Per-pod cache refresh timestamp

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
 	}

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
+	}
+}

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) {
+}