diff --git a/pkg/kubelet/apis/pods/server.go b/pkg/kubelet/apis/pods/server.go index ed73755aa59..a05beaa4f42 100644 --- a/pkg/kubelet/apis/pods/server.go +++ b/pkg/kubelet/apis/pods/server.go @@ -33,6 +33,7 @@ import ( podsv1alpha1 "k8s.io/kubelet/pkg/apis/pods/v1alpha1" "k8s.io/kubernetes/pkg/kubelet/metrics" "k8s.io/kubernetes/pkg/kubelet/pod" + podstatus "k8s.io/kubernetes/pkg/kubelet/status" ) type PodWatchEvent struct { @@ -121,23 +122,26 @@ func (b *broadcaster) Broadcast(event PodWatchEvent) { // PodsServer is the gRPC server that provides pod information. type PodsServer struct { podsv1alpha1.UnimplementedPodsServer - podManager pod.Manager - broadcaster *broadcaster + podManager pod.Manager + statusProvider podstatus.PodStatusProvider + broadcaster *broadcaster } // NewPodsServer creates a new PodServer for production use. -func NewPodsServer(broadcaster *broadcaster, podManager pod.Manager) *PodsServer { +func NewPodsServer(broadcaster *broadcaster, podManager pod.Manager, statusProvider podstatus.PodStatusProvider) *PodsServer { return &PodsServer{ - podManager: podManager, - broadcaster: broadcaster, + podManager: podManager, + statusProvider: statusProvider, + broadcaster: broadcaster, } } // NewPodsServerForTest creates a new PodServer with an injectable ticker for testing. -func NewPodsServerForTest(broadcaster *broadcaster, podManager pod.Manager) *PodsServer { +func NewPodsServerForTest(broadcaster *broadcaster, podManager pod.Manager, statusProvider podstatus.PodStatusProvider) *PodsServer { return &PodsServer{ - podManager: podManager, - broadcaster: broadcaster, + podManager: podManager, + statusProvider: statusProvider, + broadcaster: broadcaster, } } @@ -177,7 +181,13 @@ func (s *PodsServer) ListPods(ctx context.Context, req *podsv1alpha1.ListPodsReq protoPods := make([][]byte, len(podsToReturn)) for i, p := range podsToReturn { - podBytes, err := p.Marshal() + podToMarshal := p + if podStatus, ok := s.statusProvider.GetPodStatus(p.UID); ok { + podCopy := *p + podCopy.Status = podStatus + podToMarshal = &podCopy + } + podBytes, err := podToMarshal.Marshal() if err != nil { return nil, status.Errorf(codes.Internal, "failed to marshal pod: %v", err) } @@ -198,7 +208,14 @@ func (s *PodsServer) GetPod(ctx context.Context, req *podsv1alpha1.GetPodRequest return nil, status.Errorf(codes.NotFound, "pod with UID %s not found", req.PodUID) } - podBytes, err := pod.Marshal() + podToMarshal := pod + if podStatus, ok := s.statusProvider.GetPodStatus(podUID); ok { + podCopy := *pod + podCopy.Status = podStatus + podToMarshal = &podCopy + } + + podBytes, err := podToMarshal.Marshal() if err != nil { return nil, status.Errorf(codes.Internal, "failed to marshal pod: %v", err) } @@ -229,7 +246,13 @@ func (s *PodsServer) WatchPods(req *podsv1alpha1.WatchPodsRequest, stream podsv1 }) for _, p := range initialPods { - podBytes, err := p.Marshal() + podToMarshal := p + if podStatus, ok := s.statusProvider.GetPodStatus(p.UID); ok { + podCopy := *p + podCopy.Status = podStatus + podToMarshal = &podCopy + } + podBytes, err := podToMarshal.Marshal() if err != nil { logger.Error(err, "Error marshalling initial watch event pod") metrics.PodWatchEventsDroppedTotal.Inc() diff --git a/pkg/kubelet/apis/pods/server_unit_test.go b/pkg/kubelet/apis/pods/server_unit_test.go index f43fd369bb2..b149c0b14a0 100644 --- a/pkg/kubelet/apis/pods/server_unit_test.go +++ b/pkg/kubelet/apis/pods/server_unit_test.go @@ -25,6 +25,7 @@ import ( "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "google.golang.org/grpc" "google.golang.org/protobuf/proto" @@ -43,12 +44,14 @@ import ( podsapi "k8s.io/kubernetes/pkg/kubelet/apis/pods" "k8s.io/kubernetes/pkg/kubelet/metrics" kubepodtest "k8s.io/kubernetes/pkg/kubelet/pod/testing" + statustest "k8s.io/kubernetes/pkg/kubelet/status/testing" ) func TestStartEventLoop(t *testing.T) { broadcaster := podsapi.NewBroadcaster() mockManager := new(kubepodtest.MockManager) - server := podsapi.NewPodsServerForTest(broadcaster, mockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}} clientChannel := make(chan podsapi.PodWatchEvent, 100) @@ -107,11 +110,13 @@ func (m *MockWatchPodsServer) Context() context.Context { func TestWatchPods(t *testing.T) { broadcaster := podsapi.NewBroadcaster() mockManager := new(kubepodtest.MockManager) - server := podsapi.NewPodsServerForTest(broadcaster, mockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}} mockManager.On("GetPods").Return([]*v1.Pod{pod1}) mockManager.On("GetPodByUID", types.UID("pod1-uid")).Return(pod1, true) + mockStatus.On("GetPodStatus", mock.Anything).Return(v1.PodStatus{}, false) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -167,11 +172,13 @@ func TestWatchPods(t *testing.T) { func TestListPods(t *testing.T) { broadcaster := podsapi.NewBroadcaster() mockManager := new(kubepodtest.MockManager) - server := podsapi.NewPodsServerForTest(broadcaster, mockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}} pod2 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod2-uid", Name: "pod2", Namespace: "ns2"}} mockManager.On("GetPods").Return([]*v1.Pod{pod1, pod2}) + mockStatus.On("GetPodStatus", mock.Anything).Return(v1.PodStatus{}, false) resp, err := server.ListPods(context.Background(), &podsv1alpha1.ListPodsRequest{}) require.NoError(t, err) @@ -188,7 +195,8 @@ func TestListPods(t *testing.T) { func TestGetPod(t *testing.T) { broadcaster := podsapi.NewBroadcaster() mockManager := new(kubepodtest.MockManager) - server := podsapi.NewPodsServerForTest(broadcaster, mockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}} pod1.Spec.EphemeralContainers = []v1.EphemeralContainer{ { @@ -203,7 +211,8 @@ func TestGetPod(t *testing.T) { } pod1.Status = v1.PodStatus{Phase: v1.PodRunning} - mockManager.On("GetPodByUID", types.UID("pod1-uid")).Return(pod1, true) + mockManager.On("GetPodByUID", mock.Anything).Return(pod1, true) + mockStatus.On("GetPodStatus", mock.Anything).Return(v1.PodStatus{}, false) resp, err := server.GetPod(context.Background(), &podsv1alpha1.GetPodRequest{PodUID: "pod1-uid"}) require.NoError(t, err) @@ -223,7 +232,8 @@ func TestGetPod(t *testing.T) { func TestStaticPod(t *testing.T) { broadcaster := podsapi.NewBroadcaster() mockManager := new(kubepodtest.MockManager) - server := podsapi.NewPodsServerForTest(broadcaster, mockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) staticPod := &v1.Pod{ ObjectMeta: metav1.ObjectMeta{ @@ -237,6 +247,7 @@ func TestStaticPod(t *testing.T) { } mockManager.On("GetPodByUID", types.UID("static-pod-uid")).Return(staticPod, true) + mockStatus.On("GetPodStatus", mock.Anything).Return(v1.PodStatus{}, false) resp, err := server.GetPod(context.Background(), &podsv1alpha1.GetPodRequest{PodUID: "static-pod-uid"}) require.NoError(t, err) @@ -254,8 +265,10 @@ func TestErrorsAndMetrics(t *testing.T) { t.Run("DroppedWatchEventIncrementsMetric", func(t *testing.T) { broadcaster := podsapi.NewBroadcaster() mockManager := new(kubepodtest.MockManager) - server := podsapi.NewPodsServerForTest(broadcaster, mockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}} + mockStatus.On("GetPodStatus", mock.Anything).Return(v1.PodStatus{}, false) server.OnPodUpdated(pod1, pod1.Status, watch.Added) // Reset the metric before the test @@ -366,3 +379,54 @@ func TestBroadcaster_SlowClient(t *testing.T) { }) assert.NoError(t, err, "Slow client should have been dropped") } + +func TestStatusOverlay(t *testing.T) { + broadcaster := podsapi.NewBroadcaster() + mockManager := new(kubepodtest.MockManager) + mockStatus := new(statustest.MockPodStatusProvider) + server := podsapi.NewPodsServerForTest(broadcaster, mockManager, mockStatus) + + pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "pod1-uid", Name: "pod1", Namespace: "ns1"}} + pod1.Status.Phase = v1.PodPending + + overlaidStatus := v1.PodStatus{Phase: v1.PodRunning} + + mockManager.On("GetPods").Return([]*v1.Pod{pod1}) + mockManager.On("GetPodByUID", pod1.UID).Return(pod1, true) + mockStatus.On("GetPodStatus", pod1.UID).Return(overlaidStatus, true) + + t.Run("GetPod overlays status", func(t *testing.T) { + resp, err := server.GetPod(context.Background(), &podsv1alpha1.GetPodRequest{PodUID: string(pod1.UID)}) + require.NoError(t, err) + podOut := &v1.Pod{} + err = podOut.Unmarshal(resp.Pod) + require.NoError(t, err) + assert.Equal(t, v1.PodRunning, podOut.Status.Phase) + }) + + t.Run("ListPods overlays status", func(t *testing.T) { + resp, err := server.ListPods(context.Background(), &podsv1alpha1.ListPodsRequest{}) + require.NoError(t, err) + podOut := &v1.Pod{} + err = podOut.Unmarshal(resp.Pods[0]) + require.NoError(t, err) + assert.Equal(t, v1.PodRunning, podOut.Status.Phase) + }) + + t.Run("WatchPods overlays status in initial sync", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + mockStream := &MockWatchPodsServer{ + Ctx: ctx, + EventCh: make(chan *podsv1alpha1.WatchPodsEvent, 10), + } + go server.WatchPods(&podsv1alpha1.WatchPodsRequest{}, mockStream) + + event := <-mockStream.EventCh + assert.Equal(t, podsv1alpha1.EventType_ADDED, event.Type) + podOut := &v1.Pod{} + err := podOut.Unmarshal(event.Pod) + require.NoError(t, err) + assert.Equal(t, v1.PodRunning, podOut.Status.Phase) + }) +} diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index cb0a73855aa..463f2360125 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -693,12 +693,11 @@ func NewMainKubelet(ctx context.Context, klet.mirrorPodClient = kubepod.NewBasicMirrorClient(klet.kubeClient, string(nodeName), nodeLister) klet.podManager = kubepod.NewBasicPodManager() + klet.statusManager = status.NewManager(klet.kubeClient, klet.podManager, klet, kubeDeps.PodStartupLatencyTracker) if utilfeature.DefaultFeatureGate.Enabled(features.PodsAPI) { broadcaster := pods.NewBroadcaster() - klet.podsServer = pods.NewPodsServer(broadcaster, klet.podManager) + klet.podsServer = pods.NewPodsServer(broadcaster, klet.podManager, klet.statusManager) } - - klet.statusManager = status.NewManager(klet.kubeClient, klet.podManager, klet, kubeDeps.PodStartupLatencyTracker) klet.allocationManager = allocation.NewManager( klet.getRootDir(), klet.statusManager, diff --git a/pkg/kubelet/pod_workers.go b/pkg/kubelet/pod_workers.go index 5556fe55e14..a83868165d2 100644 --- a/pkg/kubelet/pod_workers.go +++ b/pkg/kubelet/pod_workers.go @@ -980,8 +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 {