From 62538f69e21d06c917fac59dd24039bd01e1b9c6 Mon Sep 17 00:00:00 2001 From: Alexej Disterhoft Date: Mon, 26 Oct 2026 14:44:17 +0100 Subject: [PATCH 1/6] fix(provider): confirm against the API server before skipping a status write The status updater compares the desired status with the object from the manager's cache and skips the write when they are equal. Right after its own write, the cache can still hold the previous version of the object. When updates for an object arrive out of order, as when the Gateway API runner stores a status computed from an older snapshot, the updater wrote the older status and then skipped the newer one, because the cache still showed the newer status from its write before. The object kept the older status, as seen in the flaky GatewayClassObservedGenerationBump conformance test. Remember the resourceVersion of the updater's last write per object. When the cache is behind it and the status looks unchanged, read the object from the API server before deciding. The entry is dropped once the cache has caught up or the object was read from the server. Signed-off-by: Alexej Disterhoft --- .../provider/kubernetes/status_updater.go | 28 +++++- .../kubernetes/status_updater_test.go | 91 +++++++++++++++++++ .../0-status-update-skipped-on-stale-cache.md | 1 + 3 files changed, 119 insertions(+), 1 deletion(-) create mode 100644 internal/provider/kubernetes/status_updater_test.go create mode 100644 release-notes/current/bug_fixes/0-status-update-skipped-on-stale-cache.md diff --git a/internal/provider/kubernetes/status_updater.go b/internal/provider/kubernetes/status_updater.go index 3f45d928f3e..96d8c3a04da 100644 --- a/internal/provider/kubernetes/status_updater.go +++ b/internal/provider/kubernetes/status_updater.go @@ -61,6 +61,8 @@ type UpdateHandler struct { apiReader client.Reader updateChannel chan Update ready chan struct{} + // Only apply touches lastWritten, on the single Start goroutine, so it needs no lock. + lastWritten map[string]string } func NewUpdateHandler(log logr.Logger, client client.Client, apiReader client.Reader) *UpdateHandler { @@ -70,6 +72,7 @@ func NewUpdateHandler(log logr.Logger, client client.Client, apiReader client.Re apiReader: apiReader, updateChannel: make(chan Update, 1000), ready: make(chan struct{}), + lastWritten: make(map[string]string), } return u @@ -83,6 +86,7 @@ func (u *UpdateHandler) apply(update Update) { ) log := u.log.WithValues("name", update.NamespacedName.Name, "namespace", update.NamespacedName.Namespace, "kind", objKind) + writeKey := objKind + "/" + update.NamespacedName.String() defer func() { updateDuration := time.Since(startTime) statusUpdateDurationSeconds.With(kindLabel.Value(objKind)).Record(updateDuration.Seconds()) @@ -114,6 +118,24 @@ func (u *UpdateHandler) apply(update Update) { newObj := update.Mutator.Mutate(obj) + if rv, ok := u.lastWritten[writeKey]; ok { + switch { + case rv == obj.GetResourceVersion(): + delete(u.lastWritten, writeKey) + case isStatusEqual(obj, newObj): + // The cache may not have seen this handler's last write yet and still show a + // status the API server no longer has, so confirm against the server first. + delete(u.lastWritten, writeKey) + if err := u.apiReader.Get(context.Background(), update.NamespacedName, obj); err != nil { + if kerrors.IsNotFound(err) { + return nil + } + return err + } + newObj = update.Mutator.Mutate(obj) + } + } + if isStatusEqual(obj, newObj) { log.Info("status unchanged, bypassing update") @@ -123,7 +145,11 @@ func (u *UpdateHandler) apply(update Update) { newObj.SetUID(obj.GetUID()) - return u.client.Status().Update(context.Background(), newObj) + if err := u.client.Status().Update(context.Background(), newObj); err != nil { + return err + } + u.lastWritten[writeKey] = newObj.GetResourceVersion() + return nil }); err != nil { log.Error(err, "unable to update status") diff --git a/internal/provider/kubernetes/status_updater_test.go b/internal/provider/kubernetes/status_updater_test.go new file mode 100644 index 00000000000..0eec7cec25c --- /dev/null +++ b/internal/provider/kubernetes/status_updater_test.go @@ -0,0 +1,91 @@ +// Copyright Envoy Gateway Authors +// SPDX-License-Identifier: Apache-2.0 +// The full text of the Apache license is available in the LICENSE file at +// the root of the repo. + +package kubernetes + +import ( + "context" + "testing" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" + gwapiv1 "sigs.k8s.io/gateway-api/apis/v1" + + egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" + "github.com/envoyproxy/gateway/internal/envoygateway" +) + +// laggingCache serves GatewayClass reads from a snapshot that the test advances by hand, +// standing in for an informer cache that has not seen the latest writes yet. Writes go +// straight to the embedded client, which plays the API server. +type laggingCache struct { + client.Client + snapshot *gwapiv1.GatewayClass +} + +func (c *laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if c.snapshot == nil { + return c.Client.Get(ctx, key, obj, opts...) + } + c.snapshot.DeepCopyInto(obj.(*gwapiv1.GatewayClass)) + return nil +} + +func (c *laggingCache) catchUp(t *testing.T, key types.NamespacedName) { + t.Helper() + c.snapshot = &gwapiv1.GatewayClass{} + require.NoError(t, c.Client.Get(t.Context(), key, c.snapshot)) +} + +func TestStatusUpdateIsNotSkippedWhileCacheLagsBehindOwnWrite(t *testing.T) { + key := types.NamespacedName{Name: "gc"} + api := fakeclient.NewClientBuilder(). + WithScheme(envoygateway.GetScheme()). + WithObjects(&gwapiv1.GatewayClass{ + ObjectMeta: metav1.ObjectMeta{Name: key.Name, Generation: 2}, + Spec: gwapiv1.GatewayClassSpec{ControllerName: egv1a1.GatewayControllerName}, + }). + WithStatusSubresource(&gwapiv1.GatewayClass{}). + Build() + cache := &laggingCache{Client: api} + u := NewUpdateHandler(logr.Discard(), cache, api) + + write := func(observedGeneration int64) { + u.apply(Update{ + NamespacedName: key, + Resource: new(gwapiv1.GatewayClass), + Mutator: MutatorFunc(func(obj client.Object) client.Object { + gc := obj.(*gwapiv1.GatewayClass).DeepCopy() + gc.Status.Conditions = []metav1.Condition{{ + Type: string(gwapiv1.GatewayClassConditionStatusAccepted), + Status: metav1.ConditionTrue, + Reason: string(gwapiv1.GatewayClassReasonAccepted), + ObservedGeneration: observedGeneration, + }} + return gc + }), + }) + } + storedObservedGeneration := func() int64 { + gc := &gwapiv1.GatewayClass{} + require.NoError(t, api.Get(t.Context(), key, gc)) + return gc.Status.Conditions[0].ObservedGeneration + } + + write(2) + cache.catchUp(t, key) + + // A late update from an older generation; the cache does not see this write. + write(1) + require.Equal(t, int64(1), storedObservedGeneration()) + + // Compared with the cache, which still shows the first write, this status looks unchanged. + write(2) + require.Equal(t, int64(2), storedObservedGeneration()) +} diff --git a/release-notes/current/bug_fixes/0-status-update-skipped-on-stale-cache.md b/release-notes/current/bug_fixes/0-status-update-skipped-on-stale-cache.md new file mode 100644 index 00000000000..7425adb19ff --- /dev/null +++ b/release-notes/current/bug_fixes/0-status-update-skipped-on-stale-cache.md @@ -0,0 +1 @@ +Fixed the status updater skipping a status write as unchanged when its cache had not yet seen its own previous write, which could leave a resource with an outdated status, for example a GatewayClass keeping the observedGeneration of its previous generation. From 44f7861f27a1cefb0d2b39706a40006faec183d3 Mon Sep 17 00:00:00 2001 From: Alexej Disterhoft Date: Mon, 26 Oct 2026 15:11:51 +0100 Subject: [PATCH 2/6] docs: name the release note after the PR Signed-off-by: Alexej Disterhoft --- ...ale-cache.md => 10409-status-update-skipped-on-stale-cache.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename release-notes/current/bug_fixes/{0-status-update-skipped-on-stale-cache.md => 10409-status-update-skipped-on-stale-cache.md} (100%) diff --git a/release-notes/current/bug_fixes/0-status-update-skipped-on-stale-cache.md b/release-notes/current/bug_fixes/10409-status-update-skipped-on-stale-cache.md similarity index 100% rename from release-notes/current/bug_fixes/0-status-update-skipped-on-stale-cache.md rename to release-notes/current/bug_fixes/10409-status-update-skipped-on-stale-cache.md From 2500865ad109331e252e65cb87aca372649c6b1e Mon Sep 17 00:00:00 2001 From: Alexej Disterhoft Date: Mon, 26 Oct 2026 15:50:04 +0100 Subject: [PATCH 3/6] fix(provider): keep the last-write record until a skipped status is confirmed When the cache looked behind the status updater's last write, the updater dropped its record of that write before checking the API server. If the following write then conflicted, the retry met the same stale cache without the record, took the cached status for the current one, and skipped the write. Keep the record until the server confirms the status is unchanged, the object is gone, or a write succeeds. Signed-off-by: Alexej Disterhoft --- .../provider/kubernetes/status_updater.go | 9 +- .../kubernetes/status_updater_test.go | 115 ++++++++++++------ 2 files changed, 87 insertions(+), 37 deletions(-) diff --git a/internal/provider/kubernetes/status_updater.go b/internal/provider/kubernetes/status_updater.go index 96d8c3a04da..ab89a50e519 100644 --- a/internal/provider/kubernetes/status_updater.go +++ b/internal/provider/kubernetes/status_updater.go @@ -124,15 +124,20 @@ func (u *UpdateHandler) apply(update Update) { delete(u.lastWritten, writeKey) case isStatusEqual(obj, newObj): // The cache may not have seen this handler's last write yet and still show a - // status the API server no longer has, so confirm against the server first. - delete(u.lastWritten, writeKey) + // status the API server no longer has, so confirm against the server first. The + // record stays until the server agrees or a write succeeds, so that a retry after + // a conflict confirms again. if err := u.apiReader.Get(context.Background(), update.NamespacedName, obj); err != nil { if kerrors.IsNotFound(err) { + delete(u.lastWritten, writeKey) return nil } return err } newObj = update.Mutator.Mutate(obj) + if isStatusEqual(obj, newObj) { + delete(u.lastWritten, writeKey) + } } } diff --git a/internal/provider/kubernetes/status_updater_test.go b/internal/provider/kubernetes/status_updater_test.go index 0eec7cec25c..7015973e426 100644 --- a/internal/provider/kubernetes/status_updater_test.go +++ b/internal/provider/kubernetes/status_updater_test.go @@ -7,14 +7,18 @@ package kubernetes import ( "context" + "fmt" "testing" "github.com/go-logr/logr" "github.com/stretchr/testify/require" + kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" gwapiv1 "sigs.k8s.io/gateway-api/apis/v1" egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" @@ -37,13 +41,17 @@ func (c *laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client return nil } -func (c *laggingCache) catchUp(t *testing.T, key types.NamespacedName) { - t.Helper() - c.snapshot = &gwapiv1.GatewayClass{} - require.NoError(t, c.Client.Get(t.Context(), key, c.snapshot)) +// gatewayClassStatusWriter writes the status of one GatewayClass through an UpdateHandler whose +// cache only catches up when the test says so. +type gatewayClassStatusWriter struct { + t *testing.T + key types.NamespacedName + api client.Client + cache *laggingCache + u *UpdateHandler } -func TestStatusUpdateIsNotSkippedWhileCacheLagsBehindOwnWrite(t *testing.T) { +func newGatewayClassStatusWriter(t *testing.T, funcs *interceptor.Funcs) *gatewayClassStatusWriter { key := types.NamespacedName{Name: "gc"} api := fakeclient.NewClientBuilder(). WithScheme(envoygateway.GetScheme()). @@ -52,40 +60,77 @@ func TestStatusUpdateIsNotSkippedWhileCacheLagsBehindOwnWrite(t *testing.T) { Spec: gwapiv1.GatewayClassSpec{ControllerName: egv1a1.GatewayControllerName}, }). WithStatusSubresource(&gwapiv1.GatewayClass{}). + WithInterceptorFuncs(*funcs). Build() cache := &laggingCache{Client: api} - u := NewUpdateHandler(logr.Discard(), cache, api) - - write := func(observedGeneration int64) { - u.apply(Update{ - NamespacedName: key, - Resource: new(gwapiv1.GatewayClass), - Mutator: MutatorFunc(func(obj client.Object) client.Object { - gc := obj.(*gwapiv1.GatewayClass).DeepCopy() - gc.Status.Conditions = []metav1.Condition{{ - Type: string(gwapiv1.GatewayClassConditionStatusAccepted), - Status: metav1.ConditionTrue, - Reason: string(gwapiv1.GatewayClassReasonAccepted), - ObservedGeneration: observedGeneration, - }} - return gc - }), - }) - } - storedObservedGeneration := func() int64 { - gc := &gwapiv1.GatewayClass{} - require.NoError(t, api.Get(t.Context(), key, gc)) - return gc.Status.Conditions[0].ObservedGeneration - } + return &gatewayClassStatusWriter{t: t, key: key, api: api, cache: cache, u: NewUpdateHandler(logr.Discard(), cache, api)} +} - write(2) - cache.catchUp(t, key) +func (w *gatewayClassStatusWriter) write(observedGeneration int64) { + w.u.apply(Update{ + NamespacedName: w.key, + Resource: new(gwapiv1.GatewayClass), + Mutator: MutatorFunc(func(obj client.Object) client.Object { + gc := obj.(*gwapiv1.GatewayClass).DeepCopy() + gc.Status.Conditions = []metav1.Condition{{ + Type: string(gwapiv1.GatewayClassConditionStatusAccepted), + Status: metav1.ConditionTrue, + Reason: string(gwapiv1.GatewayClassReasonAccepted), + ObservedGeneration: observedGeneration, + }} + return gc + }), + }) +} + +func (w *gatewayClassStatusWriter) catchUp() { + w.t.Helper() + w.cache.snapshot = &gwapiv1.GatewayClass{} + require.NoError(w.t, w.api.Get(w.t.Context(), w.key, w.cache.snapshot)) +} - // A late update from an older generation; the cache does not see this write. - write(1) - require.Equal(t, int64(1), storedObservedGeneration()) +func (w *gatewayClassStatusWriter) storedObservedGeneration() int64 { + w.t.Helper() + gc := &gwapiv1.GatewayClass{} + require.NoError(w.t, w.api.Get(w.t.Context(), w.key, gc)) + return gc.Status.Conditions[0].ObservedGeneration +} + +// writeOutOfOrder writes generation 2, lets the cache catch up, and then writes a late update +// from generation 1 that the cache does not see. +func (w *gatewayClassStatusWriter) writeOutOfOrder() { + w.t.Helper() + w.write(2) + w.catchUp() + w.write(1) + require.Equal(w.t, int64(1), w.storedObservedGeneration()) +} + +func TestStatusUpdateIsNotSkippedWhileCacheLagsBehindOwnWrite(t *testing.T) { + w := newGatewayClassStatusWriter(t, &interceptor.Funcs{}) + w.writeOutOfOrder() // Compared with the cache, which still shows the first write, this status looks unchanged. - write(2) - require.Equal(t, int64(2), storedObservedGeneration()) + w.write(2) + require.Equal(t, int64(2), w.storedObservedGeneration()) +} + +func TestStatusUpdateRetriedAfterConflictIsNotSkippedWhileCacheLags(t *testing.T) { + conflictOnce := false + w := newGatewayClassStatusWriter(t, &interceptor.Funcs{ + SubResourceUpdate: func(ctx context.Context, c client.Client, subResource string, obj client.Object, opts ...client.SubResourceUpdateOption) error { + if conflictOnce { + conflictOnce = false + return kerrors.NewConflict(schema.GroupResource{Resource: "gatewayclasses"}, obj.GetName(), fmt.Errorf("concurrent write")) + } + return c.SubResource(subResource).Update(ctx, obj, opts...) + }, + }) + w.writeOutOfOrder() + + // The write that follows the check against the API server conflicts once. The retry still + // meets the stale cache. + conflictOnce = true + w.write(2) + require.Equal(t, int64(2), w.storedObservedGeneration()) } From 03a4ecf80ff736f88b72c1d064fc07daae31b6ec Mon Sep 17 00:00:00 2001 From: Alexej Disterhoft Date: Mon, 26 Oct 2026 15:53:06 +0100 Subject: [PATCH 4/6] fix(provider): key last-write records by group and kind The status updater keyed its last-write records by kind and name. Extension policy kinds are unstructured, and two of them can share a Kind in different API groups, so their objects of the same namespace and name shared a record. An update for one could consume the other's record, and that object's next correction was then compared with its stale cache and skipped. Include the API group in the key. Signed-off-by: Alexej Disterhoft --- .../provider/kubernetes/status_updater.go | 5 +- .../kubernetes/status_updater_test.go | 96 ++++++++++++++++--- 2 files changed, 86 insertions(+), 15 deletions(-) diff --git a/internal/provider/kubernetes/status_updater.go b/internal/provider/kubernetes/status_updater.go index ab89a50e519..45cd2e3bddd 100644 --- a/internal/provider/kubernetes/status_updater.go +++ b/internal/provider/kubernetes/status_updater.go @@ -15,6 +15,7 @@ import ( kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" @@ -86,7 +87,9 @@ func (u *UpdateHandler) apply(update Update) { ) log := u.log.WithValues("name", update.NamespacedName.Name, "namespace", update.NamespacedName.Namespace, "kind", objKind) - writeKey := objKind + "/" + update.NamespacedName.String() + // Extension policy kinds are unstructured and may share a Kind across groups. + writeKey := schema.GroupKind{Group: obj.GetObjectKind().GroupVersionKind().Group, Kind: objKind}.String() + + "/" + update.NamespacedName.String() defer func() { updateDuration := time.Since(startTime) statusUpdateDurationSeconds.With(kindLabel.Value(objKind)).Record(updateDuration.Seconds()) diff --git a/internal/provider/kubernetes/status_updater_test.go b/internal/provider/kubernetes/status_updater_test.go index 7015973e426..2d71a52942b 100644 --- a/internal/provider/kubernetes/status_updater_test.go +++ b/internal/provider/kubernetes/status_updater_test.go @@ -8,12 +8,15 @@ package kubernetes import ( "context" "fmt" + "reflect" "testing" "github.com/go-logr/logr" "github.com/stretchr/testify/require" kerrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -25,22 +28,39 @@ import ( "github.com/envoyproxy/gateway/internal/envoygateway" ) -// laggingCache serves GatewayClass reads from a snapshot that the test advances by hand, -// standing in for an informer cache that has not seen the latest writes yet. Writes go -// straight to the embedded client, which plays the API server. +// laggingCache serves reads from snapshots the test takes by hand, standing in for an informer +// cache that has not seen the latest writes yet. Objects without a snapshot are read from the +// embedded client, which plays the API server and takes the writes. type laggingCache struct { client.Client - snapshot *gwapiv1.GatewayClass + snapshots map[string]client.Object +} + +func newLaggingCache(api client.Client) *laggingCache { + return &laggingCache{Client: api, snapshots: map[string]client.Object{}} +} + +func snapshotKey(obj client.Object, key client.ObjectKey) string { + return fmt.Sprintf("%T %s %s", obj, obj.GetObjectKind().GroupVersionKind().GroupKind(), key) } func (c *laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { - if c.snapshot == nil { + snapshot, ok := c.snapshots[snapshotKey(obj, key)] + if !ok { return c.Client.Get(ctx, key, obj, opts...) } - c.snapshot.DeepCopyInto(obj.(*gwapiv1.GatewayClass)) + reflect.ValueOf(obj).Elem().Set(reflect.ValueOf(snapshot.DeepCopyObject()).Elem()) return nil } +// catchUp lets the cache see the current state of the object; obj is an empty object of its type. +func (c *laggingCache) catchUp(t *testing.T, obj client.Object, key client.ObjectKey) { + t.Helper() + snapshot := snapshotKey(obj, key) + require.NoError(t, c.Client.Get(t.Context(), key, obj)) + c.snapshots[snapshot] = obj +} + // gatewayClassStatusWriter writes the status of one GatewayClass through an UpdateHandler whose // cache only catches up when the test says so. type gatewayClassStatusWriter struct { @@ -62,7 +82,7 @@ func newGatewayClassStatusWriter(t *testing.T, funcs *interceptor.Funcs) *gatewa WithStatusSubresource(&gwapiv1.GatewayClass{}). WithInterceptorFuncs(*funcs). Build() - cache := &laggingCache{Client: api} + cache := newLaggingCache(api) return &gatewayClassStatusWriter{t: t, key: key, api: api, cache: cache, u: NewUpdateHandler(logr.Discard(), cache, api)} } @@ -83,12 +103,6 @@ func (w *gatewayClassStatusWriter) write(observedGeneration int64) { }) } -func (w *gatewayClassStatusWriter) catchUp() { - w.t.Helper() - w.cache.snapshot = &gwapiv1.GatewayClass{} - require.NoError(w.t, w.api.Get(w.t.Context(), w.key, w.cache.snapshot)) -} - func (w *gatewayClassStatusWriter) storedObservedGeneration() int64 { w.t.Helper() gc := &gwapiv1.GatewayClass{} @@ -101,7 +115,7 @@ func (w *gatewayClassStatusWriter) storedObservedGeneration() int64 { func (w *gatewayClassStatusWriter) writeOutOfOrder() { w.t.Helper() w.write(2) - w.catchUp() + w.cache.catchUp(w.t, new(gwapiv1.GatewayClass), w.key) w.write(1) require.Equal(w.t, int64(1), w.storedObservedGeneration()) } @@ -134,3 +148,57 @@ func TestStatusUpdateRetriedAfterConflictIsNotSkippedWhileCacheLags(t *testing.T w.write(2) require.Equal(t, int64(2), w.storedObservedGeneration()) } + +func TestStatusUpdateRecordsOfSameKindInDifferentGroupsAreKeptApart(t *testing.T) { + policyA := schema.GroupVersionKind{Group: "a.example.com", Version: "v1", Kind: "Policy"} + policyB := schema.GroupVersionKind{Group: "b.example.com", Version: "v1", Kind: "Policy"} + key := types.NamespacedName{Namespace: "ns", Name: "policy"} + policy := func(gvk schema.GroupVersionKind) *unstructured.Unstructured { + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(gvk) + obj.SetNamespace(key.Namespace) + obj.SetName(key.Name) + return obj + } + mapper := meta.NewDefaultRESTMapper(nil) + mapper.Add(policyA, meta.RESTScopeNamespace) + mapper.Add(policyB, meta.RESTScopeNamespace) + api := fakeclient.NewClientBuilder(). + WithRESTMapper(mapper). + WithObjects(policy(policyA), policy(policyB)). + WithStatusSubresource(policy(policyA), policy(policyB)). + Build() + cache := newLaggingCache(api) + u := NewUpdateHandler(logr.Discard(), cache, api) + write := func(gvk schema.GroupVersionKind, observedGeneration int64) { + u.apply(Update{ + NamespacedName: key, + Resource: policy(gvk), + Mutator: MutatorFunc(func(obj client.Object) client.Object { + p := obj.(*unstructured.Unstructured).DeepCopy() + p.Object["status"] = map[string]any{"observedGeneration": observedGeneration} + return p + }), + }) + } + storedObservedGeneration := func(gvk schema.GroupVersionKind) int64 { + obj := policy(gvk) + require.NoError(t, api.Get(t.Context(), key, obj)) + gen, _, err := unstructured.NestedInt64(obj.Object, "status", "observedGeneration") + require.NoError(t, err) + return gen + } + + write(policyA, 2) + cache.catchUp(t, policy(policyA), key) + write(policyA, 1) + require.Equal(t, int64(1), storedObservedGeneration(policyA)) + + // A write for the other Policy, and an update that changes nothing once its cache has caught up. + write(policyB, 7) + cache.catchUp(t, policy(policyB), key) + write(policyB, 7) + + write(policyA, 2) + require.Equal(t, int64(2), storedObservedGeneration(policyA)) +} From 54a500dfd1fc0ae980fd6c123f133a77f941d28c Mon Sep 17 00:00:00 2001 From: Alexej Disterhoft Date: Mon, 26 Oct 2026 15:56:10 +0100 Subject: [PATCH 5/6] fix(provider): expire last-write records of the status updater A last-write record was only dropped by the next update of the same object. The last update of most objects is a write, and updates for deleted objects never arrive, so the records of every object that ever had its status written stayed for the life of the process. Record when each write happened and drop records older than a minute, at most once a minute. A record is only needed until the cache has seen the write, which takes far less. Signed-off-by: Alexej Disterhoft --- .../provider/kubernetes/status_updater.go | 41 ++++++++++++++++--- .../kubernetes/status_updater_test.go | 39 ++++++++++++++++-- 2 files changed, 70 insertions(+), 10 deletions(-) diff --git a/internal/provider/kubernetes/status_updater.go b/internal/provider/kubernetes/status_updater.go index 45cd2e3bddd..9ea566bfa3e 100644 --- a/internal/provider/kubernetes/status_updater.go +++ b/internal/provider/kubernetes/status_updater.go @@ -62,10 +62,22 @@ type UpdateHandler struct { apiReader client.Reader updateChannel chan Update ready chan struct{} - // Only apply touches lastWritten, on the single Start goroutine, so it needs no lock. - lastWritten map[string]string + // Only apply touches lastWritten and lastSwept, on the single Start goroutine, so they need no lock. + lastWritten map[string]lastWrite + lastSwept time.Time + now func() time.Time } +// lastWrite records a status write of the handler. +type lastWrite struct { + resourceVersion string + at time.Time +} + +// lastWriteTTL is how long a record of a status write is kept. The record is only needed until the +// cache has seen the write, which takes far less. +const lastWriteTTL = time.Minute + func NewUpdateHandler(log logr.Logger, client client.Client, apiReader client.Reader) *UpdateHandler { u := &UpdateHandler{ log: log, @@ -73,12 +85,28 @@ func NewUpdateHandler(log logr.Logger, client client.Client, apiReader client.Re apiReader: apiReader, updateChannel: make(chan Update, 1000), ready: make(chan struct{}), - lastWritten: make(map[string]string), + lastWritten: make(map[string]lastWrite), + now: time.Now, } return u } +// expireLastWrites drops records older than lastWriteTTL, at most once per lastWriteTTL. Most +// records are never looked up again: an object's last update is usually a write. +func (u *UpdateHandler) expireLastWrites() { + now := u.now() + if now.Sub(u.lastSwept) < lastWriteTTL { + return + } + u.lastSwept = now + for key, w := range u.lastWritten { + if now.Sub(w.at) >= lastWriteTTL { + delete(u.lastWritten, key) + } + } +} + func (u *UpdateHandler) apply(update Update) { var ( startTime = time.Now() @@ -94,6 +122,7 @@ func (u *UpdateHandler) apply(update Update) { updateDuration := time.Since(startTime) statusUpdateDurationSeconds.With(kindLabel.Value(objKind)).Record(updateDuration.Seconds()) }() + u.expireLastWrites() if err := retry.OnError(retry.DefaultBackoff, func(err error) bool { if kerrors.IsConflict(err) { @@ -121,9 +150,9 @@ func (u *UpdateHandler) apply(update Update) { newObj := update.Mutator.Mutate(obj) - if rv, ok := u.lastWritten[writeKey]; ok { + if last, ok := u.lastWritten[writeKey]; ok { switch { - case rv == obj.GetResourceVersion(): + case last.resourceVersion == obj.GetResourceVersion(): delete(u.lastWritten, writeKey) case isStatusEqual(obj, newObj): // The cache may not have seen this handler's last write yet and still show a @@ -156,7 +185,7 @@ func (u *UpdateHandler) apply(update Update) { if err := u.client.Status().Update(context.Background(), newObj); err != nil { return err } - u.lastWritten[writeKey] = newObj.GetResourceVersion() + u.lastWritten[writeKey] = lastWrite{resourceVersion: newObj.GetResourceVersion(), at: u.now()} return nil }); err != nil { log.Error(err, "unable to update status") diff --git a/internal/provider/kubernetes/status_updater_test.go b/internal/provider/kubernetes/status_updater_test.go index 2d71a52942b..688a4e92cff 100644 --- a/internal/provider/kubernetes/status_updater_test.go +++ b/internal/provider/kubernetes/status_updater_test.go @@ -10,6 +10,7 @@ import ( "fmt" "reflect" "testing" + "time" "github.com/go-logr/logr" "github.com/stretchr/testify/require" @@ -86,9 +87,9 @@ func newGatewayClassStatusWriter(t *testing.T, funcs *interceptor.Funcs) *gatewa return &gatewayClassStatusWriter{t: t, key: key, api: api, cache: cache, u: NewUpdateHandler(logr.Discard(), cache, api)} } -func (w *gatewayClassStatusWriter) write(observedGeneration int64) { - w.u.apply(Update{ - NamespacedName: w.key, +func acceptedGatewayClassUpdate(key types.NamespacedName, observedGeneration int64) Update { + return Update{ + NamespacedName: key, Resource: new(gwapiv1.GatewayClass), Mutator: MutatorFunc(func(obj client.Object) client.Object { gc := obj.(*gwapiv1.GatewayClass).DeepCopy() @@ -100,7 +101,11 @@ func (w *gatewayClassStatusWriter) write(observedGeneration int64) { }} return gc }), - }) + } +} + +func (w *gatewayClassStatusWriter) write(observedGeneration int64) { + w.u.apply(acceptedGatewayClassUpdate(w.key, observedGeneration)) } func (w *gatewayClassStatusWriter) storedObservedGeneration() int64 { @@ -202,3 +207,29 @@ func TestStatusUpdateRecordsOfSameKindInDifferentGroupsAreKeptApart(t *testing.T write(policyA, 2) require.Equal(t, int64(2), storedObservedGeneration(policyA)) } + +func TestStatusUpdateRecordsExpire(t *testing.T) { + builder := fakeclient.NewClientBuilder(). + WithScheme(envoygateway.GetScheme()). + WithStatusSubresource(&gwapiv1.GatewayClass{}) + for i := range 100 { + builder = builder.WithObjects(&gwapiv1.GatewayClass{ + ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("gc-%d", i)}, + Spec: gwapiv1.GatewayClassSpec{ControllerName: egv1a1.GatewayControllerName}, + }) + } + api := builder.Build() + u := NewUpdateHandler(logr.Discard(), api, api) + now := time.Now() + u.now = func() time.Time { return now } + + // The last update of most objects is a write, so its record is never looked up again. + for i := range 100 { + u.apply(acceptedGatewayClassUpdate(types.NamespacedName{Name: fmt.Sprintf("gc-%d", i)}, 1)) + } + require.Len(t, u.lastWritten, 100) + + now = now.Add(lastWriteTTL) + u.apply(acceptedGatewayClassUpdate(types.NamespacedName{Name: "gc-0"}, 1)) + require.Empty(t, u.lastWritten) +} From 14a957f119122d82594a531f1cb3b729dddb5893 Mon Sep 17 00:00:00 2001 From: Alexej Disterhoft Date: Mon, 26 Oct 2026 16:06:17 +0100 Subject: [PATCH 6/6] refactor(provider): simplify the last-write records of the status updater Key the records by UID instead of group, kind and name, since no two objects share a UID. Records are now only added by writes and only dropped when they expire. The early drops, once the cache showed the write or the API server confirmed the status, only spared an API read in rare cases. Sweep expired records from a ticker on the Start goroutine instead of throttling a sweep inside apply, which needs neither a sweep timestamp nor an injected clock. Signed-off-by: Alexej Disterhoft --- .../provider/kubernetes/status_updater.go | 68 +++----- .../kubernetes/status_updater_test.go | 155 +++++------------- 2 files changed, 71 insertions(+), 152 deletions(-) diff --git a/internal/provider/kubernetes/status_updater.go b/internal/provider/kubernetes/status_updater.go index 9ea566bfa3e..334004f2bbf 100644 --- a/internal/provider/kubernetes/status_updater.go +++ b/internal/provider/kubernetes/status_updater.go @@ -15,7 +15,6 @@ import ( kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" - "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" @@ -62,10 +61,8 @@ type UpdateHandler struct { apiReader client.Reader updateChannel chan Update ready chan struct{} - // Only apply touches lastWritten and lastSwept, on the single Start goroutine, so they need no lock. - lastWritten map[string]lastWrite - lastSwept time.Time - now func() time.Time + // Only the Start goroutine touches lastWritten, so it needs no lock. + lastWritten map[types.UID]lastWrite } // lastWrite records a status write of the handler. @@ -85,24 +82,24 @@ func NewUpdateHandler(log logr.Logger, client client.Client, apiReader client.Re apiReader: apiReader, updateChannel: make(chan Update, 1000), ready: make(chan struct{}), - lastWritten: make(map[string]lastWrite), - now: time.Now, + lastWritten: make(map[types.UID]lastWrite), } return u } -// expireLastWrites drops records older than lastWriteTTL, at most once per lastWriteTTL. Most -// records are never looked up again: an object's last update is usually a write. -func (u *UpdateHandler) expireLastWrites() { - now := u.now() - if now.Sub(u.lastSwept) < lastWriteTTL { - return - } - u.lastSwept = now - for key, w := range u.lastWritten { +// cacheMayLag reports whether obj, read from the cache, may not show this handler's last write of it yet. +func (u *UpdateHandler) cacheMayLag(obj client.Object) bool { + last, ok := u.lastWritten[obj.GetUID()] + return ok && last.resourceVersion != obj.GetResourceVersion() +} + +// expireLastWrites drops the records of writes older than lastWriteTTL. Most records are never +// looked up again: an object's last update is usually a write. +func (u *UpdateHandler) expireLastWrites(now time.Time) { + for uid, w := range u.lastWritten { if now.Sub(w.at) >= lastWriteTTL { - delete(u.lastWritten, key) + delete(u.lastWritten, uid) } } } @@ -115,14 +112,10 @@ func (u *UpdateHandler) apply(update Update) { ) log := u.log.WithValues("name", update.NamespacedName.Name, "namespace", update.NamespacedName.Namespace, "kind", objKind) - // Extension policy kinds are unstructured and may share a Kind across groups. - writeKey := schema.GroupKind{Group: obj.GetObjectKind().GroupVersionKind().Group, Kind: objKind}.String() + - "/" + update.NamespacedName.String() defer func() { updateDuration := time.Since(startTime) statusUpdateDurationSeconds.With(kindLabel.Value(objKind)).Record(updateDuration.Seconds()) }() - u.expireLastWrites() if err := retry.OnError(retry.DefaultBackoff, func(err error) bool { if kerrors.IsConflict(err) { @@ -150,27 +143,15 @@ func (u *UpdateHandler) apply(update Update) { newObj := update.Mutator.Mutate(obj) - if last, ok := u.lastWritten[writeKey]; ok { - switch { - case last.resourceVersion == obj.GetResourceVersion(): - delete(u.lastWritten, writeKey) - case isStatusEqual(obj, newObj): - // The cache may not have seen this handler's last write yet and still show a - // status the API server no longer has, so confirm against the server first. The - // record stays until the server agrees or a write succeeds, so that a retry after - // a conflict confirms again. - if err := u.apiReader.Get(context.Background(), update.NamespacedName, obj); err != nil { - if kerrors.IsNotFound(err) { - delete(u.lastWritten, writeKey) - return nil - } - return err - } - newObj = update.Mutator.Mutate(obj) - if isStatusEqual(obj, newObj) { - delete(u.lastWritten, writeKey) + if isStatusEqual(obj, newObj) && u.cacheMayLag(obj) { + // The cache may still show a status the API server no longer has, so confirm against the server. + if err := u.apiReader.Get(context.Background(), update.NamespacedName, obj); err != nil { + if kerrors.IsNotFound(err) { + return nil } + return err } + newObj = update.Mutator.Mutate(obj) } if isStatusEqual(obj, newObj) { @@ -185,7 +166,7 @@ func (u *UpdateHandler) apply(update Update) { if err := u.client.Status().Update(context.Background(), newObj); err != nil { return err } - u.lastWritten[writeKey] = lastWrite{resourceVersion: newObj.GetResourceVersion(), at: u.now()} + u.lastWritten[newObj.GetUID()] = lastWrite{resourceVersion: newObj.GetResourceVersion(), at: time.Now()} return nil }); err != nil { log.Error(err, "unable to update status") @@ -208,10 +189,15 @@ func (u *UpdateHandler) Start(ctx context.Context) error { // Enable Updaters to start sending updates to this handler. close(u.ready) + ticker := time.NewTicker(lastWriteTTL) + defer ticker.Stop() + for { select { case <-ctx.Done(): return nil + case now := <-ticker.C: + u.expireLastWrites(now) case update := <-u.updateChannel: u.log.Info("received a status update", "namespace", update.NamespacedName.Namespace, "name", update.NamespacedName.Name, "kind", KindOf(update.Resource)) diff --git a/internal/provider/kubernetes/status_updater_test.go b/internal/provider/kubernetes/status_updater_test.go index 688a4e92cff..2bb2f7c8761 100644 --- a/internal/provider/kubernetes/status_updater_test.go +++ b/internal/provider/kubernetes/status_updater_test.go @@ -8,16 +8,13 @@ package kubernetes import ( "context" "fmt" - "reflect" "testing" "time" "github.com/go-logr/logr" "github.com/stretchr/testify/require" kerrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -29,37 +26,45 @@ import ( "github.com/envoyproxy/gateway/internal/envoygateway" ) -// laggingCache serves reads from snapshots the test takes by hand, standing in for an informer -// cache that has not seen the latest writes yet. Objects without a snapshot are read from the +// laggingCache serves GatewayClass reads from a snapshot the test takes by hand, standing in for an +// informer cache that has not seen the latest writes yet. Until the first snapshot, reads go to the // embedded client, which plays the API server and takes the writes. type laggingCache struct { client.Client - snapshots map[string]client.Object + snapshot *gwapiv1.GatewayClass } -func newLaggingCache(api client.Client) *laggingCache { - return &laggingCache{Client: api, snapshots: map[string]client.Object{}} -} - -func snapshotKey(obj client.Object, key client.ObjectKey) string { - return fmt.Sprintf("%T %s %s", obj, obj.GetObjectKind().GroupVersionKind().GroupKind(), key) +func (c *laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if gc, ok := obj.(*gwapiv1.GatewayClass); ok && c.snapshot != nil { + c.snapshot.DeepCopyInto(gc) + return nil + } + return c.Client.Get(ctx, key, obj, opts...) } -func (c *laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { - snapshot, ok := c.snapshots[snapshotKey(obj, key)] - if !ok { - return c.Client.Get(ctx, key, obj, opts...) +// newGatewayClass sets a UID, which the API server would assign and the fake client does not. +func newGatewayClass(name string, generation int64) *gwapiv1.GatewayClass { + return &gwapiv1.GatewayClass{ + ObjectMeta: metav1.ObjectMeta{Name: name, UID: types.UID(name), Generation: generation}, + Spec: gwapiv1.GatewayClassSpec{ControllerName: egv1a1.GatewayControllerName}, } - reflect.ValueOf(obj).Elem().Set(reflect.ValueOf(snapshot.DeepCopyObject()).Elem()) - return nil } -// catchUp lets the cache see the current state of the object; obj is an empty object of its type. -func (c *laggingCache) catchUp(t *testing.T, obj client.Object, key client.ObjectKey) { - t.Helper() - snapshot := snapshotKey(obj, key) - require.NoError(t, c.Client.Get(t.Context(), key, obj)) - c.snapshots[snapshot] = obj +func acceptedGatewayClassUpdate(key types.NamespacedName, observedGeneration int64) Update { + return Update{ + NamespacedName: key, + Resource: new(gwapiv1.GatewayClass), + Mutator: MutatorFunc(func(obj client.Object) client.Object { + gc := obj.(*gwapiv1.GatewayClass).DeepCopy() + gc.Status.Conditions = []metav1.Condition{{ + Type: string(gwapiv1.GatewayClassConditionStatusAccepted), + Status: metav1.ConditionTrue, + Reason: string(gwapiv1.GatewayClassReasonAccepted), + ObservedGeneration: observedGeneration, + }} + return gc + }), + } } // gatewayClassStatusWriter writes the status of one GatewayClass through an UpdateHandler whose @@ -76,38 +81,26 @@ func newGatewayClassStatusWriter(t *testing.T, funcs *interceptor.Funcs) *gatewa key := types.NamespacedName{Name: "gc"} api := fakeclient.NewClientBuilder(). WithScheme(envoygateway.GetScheme()). - WithObjects(&gwapiv1.GatewayClass{ - ObjectMeta: metav1.ObjectMeta{Name: key.Name, Generation: 2}, - Spec: gwapiv1.GatewayClassSpec{ControllerName: egv1a1.GatewayControllerName}, - }). + WithObjects(newGatewayClass(key.Name, 2)). WithStatusSubresource(&gwapiv1.GatewayClass{}). WithInterceptorFuncs(*funcs). Build() - cache := newLaggingCache(api) + cache := &laggingCache{Client: api} return &gatewayClassStatusWriter{t: t, key: key, api: api, cache: cache, u: NewUpdateHandler(logr.Discard(), cache, api)} } -func acceptedGatewayClassUpdate(key types.NamespacedName, observedGeneration int64) Update { - return Update{ - NamespacedName: key, - Resource: new(gwapiv1.GatewayClass), - Mutator: MutatorFunc(func(obj client.Object) client.Object { - gc := obj.(*gwapiv1.GatewayClass).DeepCopy() - gc.Status.Conditions = []metav1.Condition{{ - Type: string(gwapiv1.GatewayClassConditionStatusAccepted), - Status: metav1.ConditionTrue, - Reason: string(gwapiv1.GatewayClassReasonAccepted), - ObservedGeneration: observedGeneration, - }} - return gc - }), - } -} - func (w *gatewayClassStatusWriter) write(observedGeneration int64) { w.u.apply(acceptedGatewayClassUpdate(w.key, observedGeneration)) } +// catchUp lets the cache see the current state of the GatewayClass. +func (w *gatewayClassStatusWriter) catchUp() { + w.t.Helper() + gc := &gwapiv1.GatewayClass{} + require.NoError(w.t, w.api.Get(w.t.Context(), w.key, gc)) + w.cache.snapshot = gc +} + func (w *gatewayClassStatusWriter) storedObservedGeneration() int64 { w.t.Helper() gc := &gwapiv1.GatewayClass{} @@ -120,7 +113,7 @@ func (w *gatewayClassStatusWriter) storedObservedGeneration() int64 { func (w *gatewayClassStatusWriter) writeOutOfOrder() { w.t.Helper() w.write(2) - w.cache.catchUp(w.t, new(gwapiv1.GatewayClass), w.key) + w.catchUp() w.write(1) require.Equal(w.t, int64(1), w.storedObservedGeneration()) } @@ -154,82 +147,22 @@ func TestStatusUpdateRetriedAfterConflictIsNotSkippedWhileCacheLags(t *testing.T require.Equal(t, int64(2), w.storedObservedGeneration()) } -func TestStatusUpdateRecordsOfSameKindInDifferentGroupsAreKeptApart(t *testing.T) { - policyA := schema.GroupVersionKind{Group: "a.example.com", Version: "v1", Kind: "Policy"} - policyB := schema.GroupVersionKind{Group: "b.example.com", Version: "v1", Kind: "Policy"} - key := types.NamespacedName{Namespace: "ns", Name: "policy"} - policy := func(gvk schema.GroupVersionKind) *unstructured.Unstructured { - obj := &unstructured.Unstructured{} - obj.SetGroupVersionKind(gvk) - obj.SetNamespace(key.Namespace) - obj.SetName(key.Name) - return obj - } - mapper := meta.NewDefaultRESTMapper(nil) - mapper.Add(policyA, meta.RESTScopeNamespace) - mapper.Add(policyB, meta.RESTScopeNamespace) - api := fakeclient.NewClientBuilder(). - WithRESTMapper(mapper). - WithObjects(policy(policyA), policy(policyB)). - WithStatusSubresource(policy(policyA), policy(policyB)). - Build() - cache := newLaggingCache(api) - u := NewUpdateHandler(logr.Discard(), cache, api) - write := func(gvk schema.GroupVersionKind, observedGeneration int64) { - u.apply(Update{ - NamespacedName: key, - Resource: policy(gvk), - Mutator: MutatorFunc(func(obj client.Object) client.Object { - p := obj.(*unstructured.Unstructured).DeepCopy() - p.Object["status"] = map[string]any{"observedGeneration": observedGeneration} - return p - }), - }) - } - storedObservedGeneration := func(gvk schema.GroupVersionKind) int64 { - obj := policy(gvk) - require.NoError(t, api.Get(t.Context(), key, obj)) - gen, _, err := unstructured.NestedInt64(obj.Object, "status", "observedGeneration") - require.NoError(t, err) - return gen - } - - write(policyA, 2) - cache.catchUp(t, policy(policyA), key) - write(policyA, 1) - require.Equal(t, int64(1), storedObservedGeneration(policyA)) - - // A write for the other Policy, and an update that changes nothing once its cache has caught up. - write(policyB, 7) - cache.catchUp(t, policy(policyB), key) - write(policyB, 7) - - write(policyA, 2) - require.Equal(t, int64(2), storedObservedGeneration(policyA)) -} - func TestStatusUpdateRecordsExpire(t *testing.T) { builder := fakeclient.NewClientBuilder(). WithScheme(envoygateway.GetScheme()). WithStatusSubresource(&gwapiv1.GatewayClass{}) for i := range 100 { - builder = builder.WithObjects(&gwapiv1.GatewayClass{ - ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("gc-%d", i)}, - Spec: gwapiv1.GatewayClassSpec{ControllerName: egv1a1.GatewayControllerName}, - }) + builder = builder.WithObjects(newGatewayClass(fmt.Sprintf("gc-%d", i), 1)) } api := builder.Build() u := NewUpdateHandler(logr.Discard(), api, api) - now := time.Now() - u.now = func() time.Time { return now } - - // The last update of most objects is a write, so its record is never looked up again. for i := range 100 { u.apply(acceptedGatewayClassUpdate(types.NamespacedName{Name: fmt.Sprintf("gc-%d", i)}, 1)) } + + u.expireLastWrites(time.Now()) require.Len(t, u.lastWritten, 100) - now = now.Add(lastWriteTTL) - u.apply(acceptedGatewayClassUpdate(types.NamespacedName{Name: "gc-0"}, 1)) + u.expireLastWrites(time.Now().Add(lastWriteTTL)) require.Empty(t, u.lastWritten) }