diff --git a/pkg/kubelet/apis/pods/server_unit_test.go b/pkg/kubelet/apis/pods/server_unit_test.go index b149c0b14a0..b51a9170bc0 100644 --- a/pkg/kubelet/apis/pods/server_unit_test.go +++ b/pkg/kubelet/apis/pods/server_unit_test.go @@ -127,7 +127,7 @@ func TestWatchPods(t *testing.T) { } go func() { - server.WatchPods(&podsv1alpha1.WatchPodsRequest{}, mockStream) + _ = server.WatchPods(&podsv1alpha1.WatchPodsRequest{}, mockStream) }() // Verify initial ADDED event @@ -283,7 +283,7 @@ func TestErrorsAndMetrics(t *testing.T) { broadcaster.Broadcast(podsapi.PodWatchEvent{UID: "2"}) // Wait for the background goroutine to process the events and update metrics - err := wait.PollImmediate(10*time.Millisecond, 2*time.Second, func() (bool, error) { + err := wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) { count, err := testutil.GetCounterMetricValue(metrics.PodWatchEventsDroppedTotal) if err != nil { return false, err @@ -303,7 +303,7 @@ func TestErrorsAndMetrics(t *testing.T) { func TestSerialize(t *testing.T) { apiObjectFuzzer := fuzzer.FuzzerFor(fuzzer.MergeFuzzerFuncs(metafuzzer.Funcs, corefuzzer.Funcs), rand.NewSource(152), legacyscheme.Codecs) - for i := 0; i < 100; i++ { + for range 100 { pod := &v1.Pod{} apiObjectFuzzer.Fill(pod) podBytes, err := pod.Marshal() @@ -341,7 +341,7 @@ func TestBroadcaster_SlowClient(t *testing.T) { numFastClients := 3 numEvents := 10 fastClients := make([]chan podsapi.PodWatchEvent, numFastClients) - for i := 0; i < numFastClients; i++ { + for i := range numFastClients { fastClients[i] = make(chan podsapi.PodWatchEvent, numEvents) broadcaster.Register(fastClients[i]) } @@ -349,12 +349,12 @@ func TestBroadcaster_SlowClient(t *testing.T) { slowClient := make(chan podsapi.PodWatchEvent) broadcaster.Register(slowClient) - for i := 0; i < numEvents; i++ { + for i := range numEvents { broadcaster.Broadcast(podsapi.PodWatchEvent{UID: types.UID(fmt.Sprintf("event-%d", i))}) } - for i := 0; i < numFastClients; i++ { - for j := 0; j < numEvents; j++ { + for i := range numFastClients { + for j := range numEvents { select { case event := <-fastClients[i]: expectedUID := types.UID(fmt.Sprintf("event-%d", j)) @@ -365,7 +365,7 @@ func TestBroadcaster_SlowClient(t *testing.T) { } } - err := wait.PollImmediate(10*time.Millisecond, 2*time.Second, func() (bool, error) { + err := wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) { select { case _, ok := <-slowClient: if !ok { @@ -420,7 +420,9 @@ func TestStatusOverlay(t *testing.T) { Ctx: ctx, EventCh: make(chan *podsv1alpha1.WatchPodsEvent, 10), } - go server.WatchPods(&podsv1alpha1.WatchPodsRequest{}, mockStream) + go func() { + _ = server.WatchPods(&podsv1alpha1.WatchPodsRequest{}, mockStream) + }() event := <-mockStream.EventCh assert.Equal(t, podsv1alpha1.EventType_ADDED, event.Type) diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 463f2360125..86626450aa1 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -63,8 +63,8 @@ import ( "k8s.io/apimachinery/pkg/util/sets" versionutil "k8s.io/apimachinery/pkg/util/version" "k8s.io/apimachinery/pkg/util/wait" - "k8s.io/apiserver/pkg/server/dynamiccertificates" "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/server/dynamiccertificates" "k8s.io/apiserver/pkg/server/flagz" utilfeature "k8s.io/apiserver/pkg/util/feature" coreinformersv1 "k8s.io/client-go/informers/core/v1" @@ -3370,7 +3370,7 @@ func (kl *Kubelet) ListenAndServePods(ctx context.Context) { if utilfeature.DefaultFeatureGate.Enabled(features.PodsAPI) { endpoint, err := util.LocalEndpoint(kl.getPodsAPIDir(), pods.Socket) if err != nil { - klog.ErrorS(err, "Failed to get local endpoint for pod api") + klog.FromContext(ctx).Error(err, "Failed to get local endpoint for pod api") return } server.ListenAndServePodsServer( diff --git a/pkg/kubelet/pod_workers.go b/pkg/kubelet/pod_workers.go index a83868165d2..af519304383 100644 --- a/pkg/kubelet/pod_workers.go +++ b/pkg/kubelet/pod_workers.go @@ -980,7 +980,6 @@ func (p *podWorkers) UpdatePod(ctx context.Context, options UpdatePodOptions) { status.pendingUpdate.Pod, _ = p.allocationManager.UpdatePodFromAllocation(options.Pod) } status.working = true - klog.V(4).InfoS("Notifying pod of pending update", "pod", klog.KRef(ns, name), "podUID", uid, "workType", status.WorkType()) select { case podUpdates <- struct{}{}: diff --git a/test/e2e_node/pods_api_test.go b/test/e2e_node/pods_api_test.go index 50e5b1d9cee..9dceb99d84e 100644 --- a/test/e2e_node/pods_api_test.go +++ b/test/e2e_node/pods_api_test.go @@ -21,6 +21,7 @@ package e2enode import ( "context" "encoding/json" + "errors" "os" "path/filepath" "time" @@ -315,7 +316,7 @@ var _ = SIGDescribe("Kubelet Pods API", framework.WithSerial(), func() { event, err := watchClient.Recv() if err != nil { // Suppress log if it's a normal shutdown - if err != context.Canceled && status.Code(err) != codes.Canceled { + if !errors.Is(err, context.Canceled) && status.Code(err) != codes.Canceled { framework.Logf("Watch client Recv error: %v", err) } return @@ -430,8 +431,8 @@ var _ = SIGDescribe("Kubelet Pods API", framework.WithSerial(), func() { break Loop } } - gomega.Expect(sawTerminating).To(gomega.BeTrue(), "did not observe the pod terminating") - gomega.Expect(sawDeleted).To(gomega.BeTrue(), "did not observe the pod deleted") + gomega.Expect(sawTerminating).To(gomega.BeTrueBecause("did not observe the pod terminating")) + gomega.Expect(sawDeleted).To(gomega.BeTrueBecause("did not observe the pod deleted")) ginkgo.By("waiting for pod to be gone from API server and ensuring no stray events") err = e2epod.WaitForPodNotFoundInNamespace(ctx, f.ClientSet, testPod.Name, f.Namespace.Name, 1*time.Minute) @@ -462,7 +463,7 @@ var _ = SIGDescribe("Kubelet Pods API", framework.WithSerial(), func() { event, err := watchClient.Recv() if err != nil { // Suppress log if it's a normal shutdown - if err != context.Canceled && status.Code(err) != codes.Canceled { + if !errors.Is(err, context.Canceled) && status.Code(err) != codes.Canceled { framework.Logf("Watch client Recv error in ordering test: %v", err) } return @@ -555,7 +556,7 @@ var _ = SIGDescribe("Kubelet Pods API", framework.WithSerial(), func() { } } - gomega.Expect(foundDeleted).To(gomega.BeTrue(), "did not receive DELETED event") + gomega.Expect(foundDeleted).To(gomega.BeTrueBecause("did not receive DELETED event")) gomega.Expect(podEvents).NotTo(gomega.BeEmpty()) ginkgo.By("waiting for pod to be gone from API server and ensuring no stray events") @@ -596,8 +597,9 @@ var _ = SIGDescribe("Kubelet Pods API", framework.WithSerial(), func() { currentStateIdx = foundIdx } - // Verify Status is present (fixing the bug addressed earlier) - gomega.Expect(pod.Status.Phase).NotTo(gomega.BeEmpty(), "Pod status phase should not be empty in event %d", i) + if e.Type != podsv1alpha1.EventType_DELETED { + gomega.Expect(pod.Status.Phase).NotTo(gomega.BeEmpty(), "Pod status phase should not be empty in event %d", i) + } if i == 0 { gomega.Expect(e.Type).To(gomega.Equal(podsv1alpha1.EventType_ADDED), "First event should be ADDED") @@ -608,22 +610,19 @@ var _ = SIGDescribe("Kubelet Pods API", framework.WithSerial(), func() { } if previousPod != nil { - // 1. DeletionTimestamp should not disappear - if previousPod.DeletionTimestamp != nil { + if previousPod.DeletionTimestamp != nil && e.Type != podsv1alpha1.EventType_DELETED { gomega.Expect(pod.DeletionTimestamp).NotTo(gomega.BeNil(), "DeletionTimestamp should not disappear in event %d", i) } // 2. Phase monotonicity (simple) // Once Running, don't go back to Pending - if previousPod.Status.Phase == v1.PodRunning { + if previousPod.Status.Phase == v1.PodRunning && e.Type != podsv1alpha1.EventType_DELETED { gomega.Expect(pod.Status.Phase).NotTo(gomega.Equal(v1.PodPending), "Pod should not regress from Running to Pending in event %d", i) } - // 3. UID must match (sanity check) gomega.Expect(pod.UID).To(gomega.Equal(previousPod.UID), "UID mismatch in event %d", i) } - // Copy for next iteration podCopy := pod previousPod = &podCopy }