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

Remove unused Snapshot config source modes.

Browse Source
This commit is contained in:
Tim Allclair
2026-01-06 14:54:48 -08:00
parent fa1dfe4181
commit 414f4e2770
3 changed files with 50 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
pkg/kubelet/config/config.go
View File

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

99
pkg/kubelet/config/config_test.go
View File

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

2
pkg/kubelet/kubelet.go
View File

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