diff --git a/pkg/kubelet/container/cache.go b/pkg/kubelet/container/cache.go index 9d2859a5e09..5067c9ddc97 100644 --- a/pkg/kubelet/container/cache.go +++ b/pkg/kubelet/container/cache.go @@ -40,6 +40,10 @@ type Cache interface { // if the data is newer than the cache based on the provided // time stamp. Returns if the cache was updated. Set(types.UID, *PodStatus, error, time.Time) (updated bool) + // SetObservedTime modifies the observed timestamp of a pod in the cache. + // This indicates that the pod's state was recently confirmed to be accurate + // by the runtime, even if its status wasn't modified. + SetObservedTime(types.UID, time.Time) Delete(types.UID) UpdateTime(time.Time) } @@ -61,6 +65,10 @@ type data struct { err error // Time when the data was last modified. modified time.Time + // Time when the data was last individually observed/verified by the runtime. + // The actual observed time should be the more recent of observedTime and the global cache timestamp. + // This is updated when the pod is relisted but its status hasn't changed. + observedTime time.Time } type subRecord struct { @@ -116,11 +124,23 @@ func (c *cache) Set(id types.UID, status *PodStatus, err error, timestamp time.T } } - c.pods[id] = &data{status: status, err: err, modified: timestamp} + c.pods[id] = &data{status: status, err: err, modified: timestamp, observedTime: timestamp} c.notify(id, timestamp) return true } +// SetObservedTime modifies the observed timestamp of a pod in the cache and +// notify subscribers if needed. +func (c *cache) SetObservedTime(id types.UID, timestamp time.Time) { + c.lock.Lock() + defer c.lock.Unlock() + if d, ok := c.pods[id]; ok { + d.observedTime = timestamp + } + // Notify all the subscribers if the condition is met. + c.notify(id, timestamp) +} + // Delete removes the entry of the pod. func (c *cache) Delete(id types.UID) { c.lock.Lock() @@ -168,9 +188,10 @@ func (c *cache) getIfNewerThan(id types.UID, minTime time.Time) *data { // minTime, return the default status. return makeDefaultData(id) } - if ok && (d.modified.After(minTime) || globalTimestampIsNewer) { + if ok && (globalTimestampIsNewer || d.modified.After(minTime) || d.observedTime.After(minTime)) { // Status is cached, return status if either of the following is true. // * status was modified after minTime + // * status was observed after minTime // * the global timestamp of the cache is newer than minTime. return d } diff --git a/pkg/kubelet/container/cache_test.go b/pkg/kubelet/container/cache_test.go index 401b8d9a928..eaab8cbf931 100644 --- a/pkg/kubelet/container/cache_test.go +++ b/pkg/kubelet/container/cache_test.go @@ -20,6 +20,7 @@ import ( "fmt" "strconv" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -61,6 +62,7 @@ func TestGetIfNewerThanWhenPodExists(t *testing.T) { cases := []struct { cacheTime time.Time modified time.Time + observed time.Time expected bool }{ { @@ -208,3 +210,110 @@ func TestRegisterNotification(t *testing.T) { // The advance of cache timestamp should've triggered the notification. verifyNotification(t, ch, true) } + +func TestGetNewerThan(t *testing.T) { + podID := types.UID("test-pod") + status := &PodStatus{ID: podID} + + cases := []struct { + name string + podExists bool + + action func(*cache, time.Time) + }{ + { + name: "UnblockBySet", + podExists: true, + action: func(c *cache, timestamp time.Time) { c.Set(podID, status, nil, timestamp) }, + }, + { + name: "UnblockBySetObservedTime", + podExists: true, + action: func(c *cache, timestamp time.Time) { c.SetObservedTime(podID, timestamp) }, + }, + { + name: "UnblockByUpdateTimeExists", + podExists: true, + action: func(c *cache, timestamp time.Time) { c.UpdateTime(timestamp) }, + }, + { + name: "Missing:UnblockBySet", + podExists: false, + action: func(c *cache, timestamp time.Time) { c.Set(podID, status, nil, timestamp) }, + }, + { + name: "Missing:UnblockByUpdateTime", + podExists: false, + action: func(c *cache, timestamp time.Time) { c.UpdateTime(timestamp) }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + initial := time.Now() + minTime := initial.Add(10 * time.Second) + + c := newTestCache() + c.UpdateTime(initial) + + if tc.podExists { + c.Set(podID, status, nil, initial) + c.SetObservedTime(podID, initial) + } + + // First call to GetNewerThan - should block initially. + resCh := make(chan *PodStatus, 1) + go func() { + s, err := c.GetNewerThan(podID, minTime) + assert.NoError(t, err) + resCh <- s + }() + + requireBlocked(t, resCh, "Initial") + + // Should still be blocked if the action is before minTime. + tc.action(c, minTime.Add(-time.Second)) + requireBlocked(t, resCh, "Before minTime") + + // Should be unblocked if the action is after minTime. + tc.action(c, minTime.Add(time.Second)) + r := requireUnblocked(t, resCh, "After minTime") + assert.Equal(t, podID, r.ID) + + // Verify a subsequent call returns immediately. + resCh2 := make(chan *PodStatus, 1) + go func() { + s, err := c.GetNewerThan(podID, minTime) + assert.NoError(t, err) + resCh2 <- s + }() + r = requireUnblocked(t, resCh2, "Already unblocked") + assert.Equal(t, podID, r.ID) + }) + }) + } +} + +func requireBlocked[T any](t *testing.T, ch <-chan T, msg string) { + t.Helper() + synctest.Wait() + select { + case r := <-ch: + t.Fatalf("%s: receive should have blocked, but got: %v", msg, r) + default: + // OK. + } +} + +func requireUnblocked[T any](t *testing.T, ch <-chan T, msg string) (received T) { + t.Helper() + synctest.Wait() + select { + case r := <-ch: + return r + default: + t.Fatalf("%s: receive should not have been blocked", msg) + return + } +} diff --git a/pkg/kubelet/container/testing/fake_cache.go b/pkg/kubelet/container/testing/fake_cache.go index e10a8f6f758..9c8c76f48a7 100644 --- a/pkg/kubelet/container/testing/fake_cache.go +++ b/pkg/kubelet/container/testing/fake_cache.go @@ -53,3 +53,6 @@ func (c *fakeCache) Delete(id types.UID) { func (c *fakeCache) UpdateTime(_ time.Time) { } + +func (c *fakeCache) SetObservedTime(id types.UID, timestamp time.Time) { +}