github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-18 13:57:38 +00:00
Code Issues Projects Releases Wiki Activity

Overlay latest status for GET and LIST, and in the initial sync for WATCH.

Browse Source
This commit is contained in:
Brian Sonnenberg
2026-03-06 19:36:35 +00:00
parent 67441f2a11
commit 2e494fc71c
4 changed files with 107 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
pkg/kubelet/apis/pods/server.go
View File

@@ -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()

78
pkg/kubelet/apis/pods/server_unit_test.go
View File

@@ -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)
})
}

5
pkg/kubelet/kubelet.go
View File

@@ -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,

2
pkg/kubelet/pod_workers.go
View File

@@ -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 {