From 8988893160db47cb57c01ff52bb4058c92cf84be Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Tue, 25 Aug 2026 11:52:08 +0100 Subject: [PATCH] perf(controller): repair mirror drift from watch events instead of resync Mirror update events were filtered out, so an in-place edit of a mirrored copy was invisible to the controller. The periodic resync did not catch it either: NeedsSync only compares the source content hash with the recorded annotation, and tampered mirror data leaves both unchanged. Changes: - The mirror watch now passes real update events (resourceVersion changed, object is a managed mirror) and maps them to the source. Informer resync replays keep an unchanged resourceVersion and stay filtered. - reconcileMirror detects in-place drift by rebuilding the expected mirror in memory and comparing content hashes. Metadata differences never count as drift, so the controller's own writes converge in one no-op pass. - The resync-period default moves from 10m to 1h. Watches now cover source changes, mirror deletion, mirror drift, and namespace changes, so the resync is only a safety net for missed watch events. The chart default moves with it. - Add predicate and tamper-repair regression tests; drift repair is verified to make no direct apiserver reads. --- README.md | 4 +- charts/kubemirror/README.md | 2 +- charts/kubemirror/values.yaml | 5 +- cmd/kubemirror/main.go | 6 +- pkg/controller/source_reconciler.go | 76 ++++++++++++++++++++---- pkg/controller/source_reconciler_test.go | 73 +++++++++++++++++++++++ 6 files changed, 147 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 879c46f..c61f852 100644 --- a/README.md +++ b/README.md @@ -592,7 +592,7 @@ Complete configuration reference: | `controller.workerThreads` | Concurrent reconciliation workers | `5` | `10`, `20` | | `controller.rateLimitQPS` | API rate limit (queries per second) | `50.0` | `100.0`, `200.0` | | `controller.rateLimitBurst` | API burst allowance | `100` | `200`, `500` | -| `controller.resyncPeriod` | Full cache resync period | `10m` | `30m` | +| `controller.resyncPeriod` | Full cache resync period (safety net; drift repair is event-driven) | `1h` | `30m` | | `controller.verifySourceFreshness` | Verify cache against a direct API read before mirroring | `false` | `true`, `false` | | **Namespace Filtering** | | | | | `controller.excludedNamespaces` | Comma-separated namespace exclusion list | `""` | `kube-system,kube-public,kube-node-lease` | @@ -623,7 +623,7 @@ When running the binary directly: - `--worker-threads int` - Concurrent workers (default: 5) - `--rate-limit-qps float32` - API rate limit (default: 50.0) - `--rate-limit-burst int` - API burst limit (default: 100) -- `--resync-period duration` - Full cache resync period (default: 10m) +- `--resync-period duration` - Full cache resync period, a safety net for missed watch events (default: 1h) - `--verify-source-freshness` - Verify cache against a direct API read before mirroring (default: false) **Namespace Filtering:** diff --git a/charts/kubemirror/README.md b/charts/kubemirror/README.md index 83445d2..cd54edc 100644 --- a/charts/kubemirror/README.md +++ b/charts/kubemirror/README.md @@ -53,7 +53,7 @@ Defaults are taken from [`values.yaml`](values.yaml). | `controller.leaderElectionID` | `"kubemirror-controller-leader"` | Lease name (`--leader-election-id`). | | `controller.resourceTypes` | `[]` | Resource types to mirror (`--resource-types`). Empty enables auto-discovery. | | `controller.discoveryInterval` | `"5m"` | Auto-discovery interval (`--discovery-interval`). | -| `controller.resyncPeriod` | `"10m"` | Cache resync period (`--resync-period`). | +| `controller.resyncPeriod` | `"1h"` | Cache resync period, safety net only (`--resync-period`). | | `controller.maxTargets` | `100` | Max target namespaces per resource (`--max-targets`). | | `controller.workerThreads` | `5` | Concurrent reconcile workers (`--worker-threads`). | | `controller.rateLimitQPS` | `50.0` | API QPS limit (`--rate-limit-qps`). | diff --git a/charts/kubemirror/values.yaml b/charts/kubemirror/values.yaml index d76b64a..b883730 100644 --- a/charts/kubemirror/values.yaml +++ b/charts/kubemirror/values.yaml @@ -54,10 +54,11 @@ controller: # How often to rediscover available resources in the cluster discoveryInterval: "5m" - # Cache resync period - how often to refresh all cached resources + # Cache resync period - safety net for missed watch events; mirror drift + # is repaired event-driven, so this can stay long # Higher values reduce memory churn and API load # Default: 10m (was 30s in earlier versions) - resyncPeriod: "10m" + resyncPeriod: "1h" # Resource limits maxTargets: 100 diff --git a/cmd/kubemirror/main.go b/cmd/kubemirror/main.go index 87e9073..9aa418e 100644 --- a/cmd/kubemirror/main.go +++ b/cmd/kubemirror/main.go @@ -108,8 +108,10 @@ func main() { "QPS rate limit for API server requests.") flag.IntVar(&rateLimitBurst, "rate-limit-burst", 100, "Burst limit for API server requests.") - flag.DurationVar(&resyncPeriod, "resync-period", 10*time.Minute, - "Period for resyncing all resources (catches updates missed due to informer cache delays).") + flag.DurationVar(&resyncPeriod, "resync-period", time.Hour, + "Period for the full informer resync. Sources, mirrors, and namespaces "+ + "are watched, and mirror drift is repaired event-driven, so the resync "+ + "is only a safety net for missed watch events.") flag.BoolVar(&verifySourceFreshness, "verify-source-freshness", false, "Verify source resource freshness by comparing cache with direct API read. "+ "Converges a few seconds faster when the informer cache lags behind watch "+ diff --git a/pkg/controller/source_reconciler.go b/pkg/controller/source_reconciler.go index 2ff67bf..97513ab 100644 --- a/pkg/controller/source_reconciler.go +++ b/pkg/controller/source_reconciler.go @@ -378,8 +378,19 @@ func (r *SourceReconciler) reconcileMirror(ctx context.Context, source runtime.O } if !needsSync { - logger.V(2).Info("mirror is up to date") - return nil + // The source is unchanged since the last sync, but the mirrored + // copy itself may have been edited in place (its bookkeeping + // annotations would still match the source). Rebuild the expected + // mirror in memory and compare content hashes to catch that. + drifted, driftErr := mirrorContentDrifted(source, existing, targetNs) + if driftErr != nil { + return fmt.Errorf("failed to check mirror drift: %w", driftErr) + } + if !drifted { + logger.V(2).Info("mirror is up to date") + return nil + } + logger.Info("mirror content drifted from expected state, repairing") } // Update mirror @@ -421,6 +432,30 @@ func (r *SourceReconciler) reconcileMirror(ctx context.Context, source runtime.O return nil } +// mirrorContentDrifted reports whether the live mirror's content diverged from +// what mirroring the source into targetNs should produce. It rebuilds the +// expected mirror in memory (including transformations) and compares content +// hashes, so metadata-only differences never count as drift. This is what +// makes drift repair event-driven instead of depending on the periodic resync. +func mirrorContentDrifted(source runtime.Object, existing *unstructured.Unstructured, targetNs string) (bool, error) { + expected, err := CreateMirror(source, targetNs) + if err != nil { + return false, fmt.Errorf("failed to build expected mirror: %w", err) + } + + expectedHash, err := hash.ComputeContentHash(expected) + if err != nil { + return false, fmt.Errorf("failed to hash expected mirror: %w", err) + } + + existingHash, err := hash.ComputeContentHash(existing) + if err != nil { + return false, fmt.Errorf("failed to hash existing mirror: %w", err) + } + + return expectedHash != existingHash, nil +} + // deleteAllMirrors deletes all mirrors that this source owns across the cluster. // It verifies ownership (managed-by label + source-reference annotation) before // deleting anything to avoid destroying unrelated resources that happen to share @@ -680,26 +715,43 @@ func (r *SourceReconciler) SetupWithManagerForResourceType( mirrorObj := &unstructured.Unstructured{} mirrorObj.SetGroupVersionKind(gvk) - // Create predicates to only watch mirror deletions - mirrorDeletePredicate := predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { return false }, - UpdateFunc: func(e event.UpdateEvent) bool { return false }, - DeleteFunc: func(e event.DeleteEvent) bool { return IsMirrorResource(e.Object) }, - GenericFunc: func(e event.GenericEvent) bool { return false }, - } - return ctrl.NewControllerManagedBy(mgr). For(obj). Named(controllerName). - // Watch mirror resources - when deleted, enqueue source for reconciliation + // Watch mirror resources - deletes trigger recreation, real writes + // trigger drift repair. This keeps mirrors converged without relying + // on the periodic resync. Watches( mirrorObj, handler.EnqueueRequestsFromMapFunc(r.mapMirrorToSource), - builder.WithPredicates(mirrorDeletePredicate), + builder.WithPredicates(mirrorEventPredicate()), ). Complete(r) } +// mirrorEventPredicate gates the mirror watch. Deletes of managed mirrors +// enqueue the source so the mirror is recreated. Updates enqueue the source +// so in-place edits to a mirror are repaired event-driven; informer resync +// replays carry an unchanged resourceVersion and are filtered out, so only +// real writes pass. The controller's own mirror writes pass too and converge +// in one extra reconcile that no-ops on the content comparison. +func mirrorEventPredicate() predicate.Funcs { + return predicate.Funcs{ + CreateFunc: func(e event.CreateEvent) bool { return false }, + UpdateFunc: func(e event.UpdateEvent) bool { + if e.ObjectOld == nil || e.ObjectNew == nil { + return false + } + if e.ObjectOld.GetResourceVersion() == e.ObjectNew.GetResourceVersion() { + return false + } + return IsMirrorResource(e.ObjectNew) + }, + DeleteFunc: func(e event.DeleteEvent) bool { return IsMirrorResource(e.Object) }, + GenericFunc: func(e event.GenericEvent) bool { return false }, + } +} + // mapMirrorToSource maps a mirror resource to its source for reconciliation. func (r *SourceReconciler) mapMirrorToSource(ctx context.Context, obj client.Object) []reconcile.Request { // Only process if this is a mirror diff --git a/pkg/controller/source_reconciler_test.go b/pkg/controller/source_reconciler_test.go index a3b405f..182ad3d 100644 --- a/pkg/controller/source_reconciler_test.go +++ b/pkg/controller/source_reconciler_test.go @@ -18,6 +18,7 @@ import ( "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" "github.com/lukaszraczylo/kubemirror/pkg/config" "github.com/lukaszraczylo/kubemirror/pkg/constants" @@ -1364,3 +1365,75 @@ func TestSourceReconciler_getSourceWithFreshness_DefaultNoDirectRead(t *testing. assert.Equal(t, "test-secret", got.GetName()) assert.Zero(t, reader.gets, "default config must not verify freshness against the apiserver") } + +// TestMirrorEventPredicate locks the mirror-watch gating: real writes to a +// mirror enqueue its source for drift repair, while informer resync replays +// (unchanged resourceVersion), non-mirror objects, and creates stay silent. +func TestMirrorEventPredicate(t *testing.T) { + source := newMirrorTestSource(t) + mirrorObj, err := CreateMirror(source, "target-ns") + require.NoError(t, err) + mirror, ok := mirrorObj.(*unstructured.Unstructured) + require.True(t, ok) + mirror.SetResourceVersion("100") + + changed := mirror.DeepCopy() + changed.SetResourceVersion("101") + + nonMirrorOld := source.DeepCopy() + nonMirrorOld.SetResourceVersion("1") + nonMirrorNew := source.DeepCopy() + nonMirrorNew.SetResourceVersion("2") + + p := mirrorEventPredicate() + + assert.False(t, p.Create(event.CreateEvent{Object: mirror}), + "mirror create must not enqueue") + assert.True(t, p.Update(event.UpdateEvent{ObjectOld: mirror, ObjectNew: changed}), + "real mirror write must enqueue the source for drift repair") + assert.False(t, p.Update(event.UpdateEvent{ObjectOld: mirror, ObjectNew: mirror}), + "informer resync replay with unchanged resourceVersion must not enqueue") + assert.False(t, p.Update(event.UpdateEvent{ObjectOld: nonMirrorOld, ObjectNew: nonMirrorNew}), + "non-mirror update must not enqueue") + assert.True(t, p.Delete(event.DeleteEvent{Object: mirror}), + "mirror delete must enqueue the source for recreation") + assert.False(t, p.Delete(event.DeleteEvent{Object: source}), + "non-mirror delete must not enqueue") +} + +// TestSourceReconciler_reconcileMirror_RepairsDriftedMirror covers in-place +// mirror tampering: the mirrored data is edited but the bookkeeping +// annotations are left intact, so the source-hash check alone would consider +// the mirror up to date. The reconciler must detect the content divergence +// and rewrite the mirror — still without any direct API reads. +func TestSourceReconciler_reconcileMirror_RepairsDriftedMirror(t *testing.T) { + source := newMirrorTestSource(t) + + mirrorObj, err := CreateMirror(source, "target-ns") + require.NoError(t, err) + mirror, ok := mirrorObj.(*unstructured.Unstructured) + require.True(t, ok) + mirror.SetResourceVersion("100") + + // Tamper with the mirrored content in place; annotations stay intact. + require.NoError(t, unstructured.SetNestedStringMap(mirror.Object, + map[string]string{"tampered-field": "dGFtcGVyZWQ="}, "data")) + + mockClient := new(MockClient) + mockClient.On("Get", mock.Anything, types.NamespacedName{Namespace: "target-ns", Name: "test-secret"}, mock.Anything). + Return(nil, mirror) + mockClient.On("Update", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + reader := &countingReader{} + r := &SourceReconciler{ + Client: mockClient, + APIReader: reader, + Config: &config.Config{}, + GVK: source.GroupVersionKind(), + } + + require.NoError(t, r.reconcileMirror(context.Background(), source, source, "target-ns")) + + mockClient.AssertCalled(t, "Update", mock.Anything, mock.Anything, mock.Anything) + assert.Zero(t, reader.gets, "drift repair must not read from the apiserver directly") +}