From 414f4e27707e60dd95b2f39073b62a13406c3cb4 Mon Sep 17 00:00:00 2001 From: Tim Allclair Date: Tue, 6 Jan 2026 14:54:48 -0800 Subject: [PATCH] Remove unused Snapshot config source modes. --- pkg/kubelet/config/config.go | 92 ++++++++-------------------- pkg/kubelet/config/config_test.go | 99 ++++++++----------------------- pkg/kubelet/kubelet.go | 2 +- 3 files changed, 50 insertions(+), 143 deletions(-) diff --git a/pkg/kubelet/config/config.go b/pkg/kubelet/config/config.go index f85aa4d3e0a..485481a22b3 100644 --- a/pkg/kubelet/config/config.go +++ b/pkg/kubelet/config/config.go @@ -18,7 +18,6 @@ package config import ( "context" - "fmt" "reflect" "sync" "time" @@ -34,23 +33,6 @@ import ( "k8s.io/kubernetes/pkg/kubelet/util/format" ) -// PodConfigNotificationMode describes how changes are sent to the update channel. -type PodConfigNotificationMode int - -const ( - // PodConfigNotificationUnknown is the default value for - // PodConfigNotificationMode when uninitialized. - PodConfigNotificationUnknown PodConfigNotificationMode = iota - // PodConfigNotificationSnapshot delivers the full configuration as a SET whenever - // any change occurs. - PodConfigNotificationSnapshot - // PodConfigNotificationSnapshotAndUpdates delivers an UPDATE and DELETE message whenever pods are - // changed, and a SET message if there are any additions or removals. - PodConfigNotificationSnapshotAndUpdates - // PodConfigNotificationIncremental delivers ADD, UPDATE, DELETE, REMOVE, RECONCILE to the update channel. - PodConfigNotificationIncremental -) - type podStartupSLIObserver interface { ObservedPodOnWatch(pod *v1.Pod, when time.Time) } @@ -76,9 +58,9 @@ type sourceUpdate struct { // NewPodConfig creates an object that can merge many configuration sources into a stream // of normalized updates to a pod configuration. -func NewPodConfig(mode PodConfigNotificationMode, recorder record.EventRecorder, startupSLIObserver podStartupSLIObserver) *PodConfig { +func NewPodConfig(recorder record.EventRecorder, startupSLIObserver podStartupSLIObserver) *PodConfig { updates := make(chan kubetypes.PodUpdate, 50) - storage := newPodStorage(updates, mode, recorder, startupSLIObserver) + storage := newPodStorage(updates, recorder, startupSLIObserver) podConfig := &PodConfig{ pods: storage, mux: newMux(storage), @@ -130,7 +112,6 @@ type podStorage struct { podLock sync.RWMutex // map of source name to pod uid to pod reference pods map[string]map[types.UID]*v1.Pod - mode PodConfigNotificationMode // ensures that updates are delivered in strict order // on the updates channel @@ -150,10 +131,9 @@ type podStorage struct { // TODO: PodConfigNotificationMode could be handled by a listener to the updates channel // in the future, especially with multiple listeners. // TODO: allow initialization of the current state of the store with snapshotted version. -func newPodStorage(updates chan<- kubetypes.PodUpdate, mode PodConfigNotificationMode, recorder record.EventRecorder, startupSLIObserver podStartupSLIObserver) *podStorage { +func newPodStorage(updates chan<- kubetypes.PodUpdate, recorder record.EventRecorder, startupSLIObserver podStartupSLIObserver) *podStorage { return &podStorage{ pods: make(map[string]map[types.UID]*v1.Pod), - mode: mode, updates: updates, sourcesSeen: sets.Set[string]{}, recorder: recorder, @@ -173,51 +153,27 @@ func (s *podStorage) Merge(ctx context.Context, source string, update sourceUpda firstSet := !seenBefore && s.sourcesSeen.Has(source) // deliver update notifications - switch s.mode { - case PodConfigNotificationIncremental: - if len(removes.Pods) > 0 { - s.updates <- *removes - } - if len(adds.Pods) > 0 { - s.updates <- *adds - } - if len(updates.Pods) > 0 { - s.updates <- *updates - } - if len(deletes.Pods) > 0 { - s.updates <- *deletes - } - if firstSet && len(adds.Pods) == 0 && len(updates.Pods) == 0 && len(deletes.Pods) == 0 { - // Send an empty update when first seeing the source and there are - // no ADD or UPDATE or DELETE pods from the source. This signals kubelet that - // the source is ready. - s.updates <- *adds - } - // Only add reconcile support here, because kubelet doesn't support Snapshot update now. - if len(reconciles.Pods) > 0 { - s.updates <- *reconciles - } - - case PodConfigNotificationSnapshotAndUpdates: - if len(removes.Pods) > 0 || len(adds.Pods) > 0 || firstSet { - s.updates <- kubetypes.PodUpdate{Pods: s.mergedState().([]*v1.Pod), Op: kubetypes.SET, Source: source} - } - if len(updates.Pods) > 0 { - s.updates <- *updates - } - if len(deletes.Pods) > 0 { - s.updates <- *deletes - } - - case PodConfigNotificationSnapshot: - if len(updates.Pods) > 0 || len(deletes.Pods) > 0 || len(adds.Pods) > 0 || len(removes.Pods) > 0 || firstSet { - s.updates <- kubetypes.PodUpdate{Pods: s.mergedState().([]*v1.Pod), Op: kubetypes.SET, Source: source} - } - - case PodConfigNotificationUnknown: - fallthrough - default: - panic(fmt.Sprintf("unsupported PodConfigNotificationMode: %#v", s.mode)) + if len(removes.Pods) > 0 { + s.updates <- *removes + } + if len(adds.Pods) > 0 { + s.updates <- *adds + } + if len(updates.Pods) > 0 { + s.updates <- *updates + } + if len(deletes.Pods) > 0 { + s.updates <- *deletes + } + if firstSet && len(adds.Pods) == 0 && len(updates.Pods) == 0 && len(deletes.Pods) == 0 { + // Send an empty update when first seeing the source and there are + // no ADD or UPDATE or DELETE pods from the source. This signals kubelet that + // the source is ready. + s.updates <- *adds + } + // Only add reconcile support here, because kubelet doesn't support Snapshot update now. + if len(reconciles.Pods) > 0 { + s.updates <- *reconciles } return nil diff --git a/pkg/kubelet/config/config_test.go b/pkg/kubelet/config/config_test.go index df72f85d84f..8a96ce67dfb 100644 --- a/pkg/kubelet/config/config_test.go +++ b/pkg/kubelet/config/config_test.go @@ -97,9 +97,9 @@ func createSourceUpdate(pods ...*v1.Pod) sourceUpdate { return sourceUpdate{pods} } -func createPodConfigTester(ctx context.Context, mode PodConfigNotificationMode) (chan<- sourceUpdate, <-chan kubetypes.PodUpdate, *PodConfig) { +func createPodConfigTester(ctx context.Context) (chan<- sourceUpdate, <-chan kubetypes.PodUpdate, *PodConfig) { eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx)) - config := NewPodConfig(mode, eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "kubelet"}), &mockPodStartupSLIObserver{}) + config := NewPodConfig(eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "kubelet"}), &mockPodStartupSLIObserver{}) channel := config.Channel(ctx, TestSource) ch := config.Updates() return channel, ch, config @@ -144,7 +144,7 @@ func expectNoPodUpdate(t *testing.T, ch <-chan kubetypes.PodUpdate) { func TestNewPodAdded(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, config := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, config := createPodConfigTester(tCtx) // see an update podUpdate := createSourceUpdate(CreateValidPod("foo", "new")) @@ -158,7 +158,7 @@ func TestNewPodAdded(t *testing.T) { func TestNewPodAddedInvalidNamespace(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, config := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, config := createPodConfigTester(tCtx) // see an update podUpdate := createSourceUpdate(CreateValidPod("foo", "")) @@ -172,7 +172,7 @@ func TestNewPodAddedInvalidNamespace(t *testing.T) { func TestNewPodAddedDefaultNamespace(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, config := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, config := createPodConfigTester(tCtx) // see an update podUpdate := createSourceUpdate(CreateValidPod("foo", "default")) @@ -186,7 +186,7 @@ func TestNewPodAddedDefaultNamespace(t *testing.T) { func TestNewPodAddedDifferentNamespaces(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, config := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, config := createPodConfigTester(tCtx) // see an update pod1 := CreateValidPod("foo", "default") @@ -207,7 +207,7 @@ func TestNewPodAddedDifferentNamespaces(t *testing.T) { func TestInvalidPodFiltered(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) // see an update podUpdate := createSourceUpdate(CreateValidPod("foo", "new")) @@ -220,50 +220,10 @@ func TestInvalidPodFiltered(t *testing.T) { expectNoPodUpdate(t, ch) } -func TestNewPodAddedSnapshotAndUpdates(t *testing.T) { - tCtx := ktesting.Init(t) - - channel, ch, config := createPodConfigTester(tCtx, PodConfigNotificationSnapshotAndUpdates) - - // see an set - podUpdate := createSourceUpdate(CreateValidPod("foo", "new")) - channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.SET, TestSource, CreateValidPod("foo", "new"))) - - config.Sync() - expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.SET, kubetypes.AllSource, CreateValidPod("foo", "new"))) - - // container updates are separated as UPDATE - pod := *podUpdate.Pods[0] - pod.Spec.Containers = []v1.Container{{Name: "bar", Image: "test", ImagePullPolicy: v1.PullIfNotPresent, TerminationMessagePolicy: v1.TerminationMessageReadFile}} - channel <- createSourceUpdate(&pod) - expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.UPDATE, TestSource, &pod)) -} - -func TestNewPodAddedSnapshot(t *testing.T) { - tCtx := ktesting.Init(t) - - channel, ch, config := createPodConfigTester(tCtx, PodConfigNotificationSnapshot) - - // see an set - podUpdate := createSourceUpdate(CreateValidPod("foo", "new")) - channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.SET, TestSource, CreateValidPod("foo", "new"))) - - config.Sync() - expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.SET, kubetypes.AllSource, CreateValidPod("foo", "new"))) - - // container updates are separated as UPDATE - pod := *podUpdate.Pods[0] - pod.Spec.Containers = []v1.Container{{Name: "bar", Image: "test", ImagePullPolicy: v1.PullIfNotPresent, TerminationMessagePolicy: v1.TerminationMessageReadFile}} - channel <- createSourceUpdate(&pod) - expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.SET, TestSource, &pod)) -} - func TestNewPodAddedUpdatedRemoved(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) // should register an add podUpdate := createSourceUpdate(CreateValidPod("foo", "new")) @@ -288,7 +248,7 @@ func TestNewPodAddedUpdatedRemoved(t *testing.T) { func TestNewPodAddedDelete(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) // should register an add addedPod := CreateValidPod("foo", "new") @@ -309,7 +269,7 @@ func TestNewPodAddedDelete(t *testing.T) { func TestNewPodAddedUpdatedSet(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) // should register an add podUpdate := createSourceUpdate(CreateValidPod("foo", "new"), CreateValidPod("foo2", "new"), CreateValidPod("foo3", "new")) @@ -352,7 +312,7 @@ func TestNewPodAddedSetReconciled(t *testing.T) { } var podWithStatusChange *v1.Pod pods, _ := newTestPods(false, false) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) // Use SET to initialize the config, especially initialize the source set channel <- createSourceUpdate(pods...) @@ -376,34 +336,25 @@ func TestNewPodAddedSetReconciled(t *testing.T) { func TestInitialEmptySet(t *testing.T) { tCtx := ktesting.Init(t) - for _, test := range []struct { - mode PodConfigNotificationMode - op kubetypes.PodOperation - }{ - {PodConfigNotificationIncremental, kubetypes.ADD}, - {PodConfigNotificationSnapshot, kubetypes.SET}, - {PodConfigNotificationSnapshotAndUpdates, kubetypes.SET}, - } { - channel, ch, _ := createPodConfigTester(tCtx, test.mode) + channel, ch, _ := createPodConfigTester(tCtx) - // should register an empty PodUpdate operation - podUpdate := createSourceUpdate() - channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(test.op, TestSource)) + // should register an empty PodUpdate operation + podUpdate := createSourceUpdate() + channel <- podUpdate + expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.ADD, TestSource)) - // should ignore following empty sets - podUpdate = createSourceUpdate() - channel <- podUpdate - podUpdate = createSourceUpdate(CreateValidPod("foo", "new")) - channel <- podUpdate - expectPodUpdate(t, ch, CreatePodUpdate(test.op, TestSource, CreateValidPod("foo", "new"))) - } + // should ignore following empty sets + podUpdate = createSourceUpdate() + channel <- podUpdate + podUpdate = createSourceUpdate(CreateValidPod("foo", "new")) + channel <- podUpdate + expectPodUpdate(t, ch, CreatePodUpdate(kubetypes.ADD, TestSource, CreateValidPod("foo", "new"))) } func TestPodUpdateAnnotations(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) pod := CreateValidPod("foo2", "new") pod.Annotations = make(map[string]string) @@ -434,7 +385,7 @@ func TestPodUpdateAnnotations(t *testing.T) { func TestPodUpdateLabels(t *testing.T) { tCtx := ktesting.Init(t) - channel, ch, _ := createPodConfigTester(tCtx, PodConfigNotificationIncremental) + channel, ch, _ := createPodConfigTester(tCtx) pod := CreateValidPod("foo2", "new") pod.Labels = make(map[string]string) @@ -457,7 +408,7 @@ func TestPodConfigRace(t *testing.T) { tCtx := ktesting.Init(t) eventBroadcaster := record.NewBroadcaster(record.WithContext(tCtx)) - config := NewPodConfig(PodConfigNotificationIncremental, eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "kubelet"}), &mockPodStartupSLIObserver{}) + config := NewPodConfig(eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "kubelet"}), &mockPodStartupSLIObserver{}) seenSources := sets.New[string](TestSource) var wg sync.WaitGroup const iterations = 100 diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 9acfc620c32..e59491dea6e 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -373,7 +373,7 @@ func makePodSourceConfig(kubeCfg *kubeletconfiginternal.KubeletConfiguration, ku } // source of all configuration - cfg := config.NewPodConfig(config.PodConfigNotificationIncremental, kubeDeps.Recorder, kubeDeps.PodStartupLatencyTracker) + cfg := config.NewPodConfig(kubeDeps.Recorder, kubeDeps.PodStartupLatencyTracker) // TODO: it needs to be replaced by a proper context in the future ctx := context.TODO()