diff --git a/README.md b/README.md index 0cfe343..879c46f 100644 --- a/README.md +++ b/README.md @@ -624,7 +624,7 @@ When running the binary directly: - `--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) -- `--verify-source-freshness` - Verify cache against a direct API read before mirroring (default: true) +- `--verify-source-freshness` - Verify cache against a direct API read before mirroring (default: false) **Namespace Filtering:** - `--excluded-namespaces string` - Comma-separated exclusion list diff --git a/charts/kubemirror/templates/deployment.yaml b/charts/kubemirror/templates/deployment.yaml index b0f897f..704dc08 100644 --- a/charts/kubemirror/templates/deployment.yaml +++ b/charts/kubemirror/templates/deployment.yaml @@ -43,9 +43,7 @@ spec: - --worker-threads={{ .Values.controller.workerThreads }} - --rate-limit-qps={{ .Values.controller.rateLimitQPS }} - --rate-limit-burst={{ .Values.controller.rateLimitBurst }} - {{- if .Values.controller.verifySourceFreshness }} - - --verify-source-freshness=true - {{- end }} + - --verify-source-freshness={{ .Values.controller.verifySourceFreshness }} {{- if .Values.controller.lazyWatcherInit }} - --lazy-watcher-init=true {{- end }} diff --git a/cmd/kubemirror/main.go b/cmd/kubemirror/main.go index f3fe138..87e9073 100644 --- a/cmd/kubemirror/main.go +++ b/cmd/kubemirror/main.go @@ -110,12 +110,12 @@ func main() { "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.BoolVar(&verifySourceFreshness, "verify-source-freshness", true, + flag.BoolVar(&verifySourceFreshness, "verify-source-freshness", false, "Verify source resource freshness by comparing cache with direct API read. "+ - "Prevents mirroring stale data and missed orphan cleanups when the informer "+ - "cache lags behind watch events. Trade-off: one extra API call per reconcile "+ - "when the cache is stale. Disable only if you are confident your cluster's "+ - "watch latency is negligible.") + "Converges a few seconds faster when the informer cache lags behind watch "+ + "events, at the cost of one direct apiserver GET per reconcile. Off by "+ + "default: a stale read self-corrects on the next watch event, and update "+ + "conflicts requeue automatically.") flag.BoolVar(&lazyWatcherInit, "lazy-watcher-init", false, "Enable lazy watcher initialization - only create informers for resource types that have resources marked for mirroring. "+ "Significantly reduces memory usage by avoiding watchers for unused resource types. "+ diff --git a/pkg/controller/namespace_reconciler.go b/pkg/controller/namespace_reconciler.go index a6eb55e..c4f5c74 100644 --- a/pkg/controller/namespace_reconciler.go +++ b/pkg/controller/namespace_reconciler.go @@ -196,8 +196,9 @@ func (r *NamespaceReconciler) reconcileResourceType(ctx context.Context, rt conf // (pattern parsing/validation, namespace listing, max-targets clamping) lives in // exactly one place, mirroring the reconcileMirror delegation below. func (r *NamespaceReconciler) resolveTargetNamespaces(ctx context.Context, source *unstructured.Unstructured) ([]string, error) { - return r.newSourceReconciler(source.GroupVersionKind()). + targets, _, err := r.newSourceReconciler(source.GroupVersionKind()). resolveTargetNamespaces(ctx, source) + return targets, err } // reconcileMirror creates or updates a mirror in the target namespace by diff --git a/pkg/controller/source_reconciler.go b/pkg/controller/source_reconciler.go index b9252f1..2ff67bf 100644 --- a/pkg/controller/source_reconciler.go +++ b/pkg/controller/source_reconciler.go @@ -238,8 +238,9 @@ func (r *SourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr return ctrl.Result{Requeue: true}, nil } - // Get target namespaces - targetNamespaces, err := r.resolveTargetNamespaces(ctx, sourceObj) + // Get target namespaces (nsInfo is reused for orphan cleanup below so a + // reconcile pass performs a single namespace LIST instead of two) + targetNamespaces, nsInfo, err := r.resolveTargetNamespaces(ctx, sourceObj) if err != nil { logger.Error(err, "failed to resolve target namespaces") if r.CircuitBreaker != nil { @@ -268,7 +269,7 @@ func (r *SourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr } // Clean up orphaned mirrors (namespaces that no longer match the target criteria) - orphanedCount, err := r.cleanupOrphanedMirrors(ctx, sourceObj, targetNamespaces) + orphanedCount, err := r.cleanupOrphanedMirrors(ctx, sourceObj, targetNamespaces, nsInfo) if err != nil { logger.Error(err, "failed to cleanup orphaned mirrors") // Don't fail reconciliation for cleanup errors, just log them @@ -356,19 +357,12 @@ func (r *SourceReconciler) reconcileMirror(ctx context.Context, source runtime.O return fmt.Errorf("failed to get existing mirror: %w", err) } - // If freshness verification is enabled and mirror exists, verify it's fresh too - if err == nil && r.Config.VerifySourceFreshness && r.APIReader != nil { - fresh := &unstructured.Unstructured{} - fresh.SetGroupVersionKind(sourceUnstructured.GroupVersionKind()) - if apiErr := r.APIReader.Get(ctx, client.ObjectKey{Namespace: targetNs, Name: sourceObj.GetName()}, fresh); apiErr == nil { - if fresh.GetResourceVersion() != existing.GetResourceVersion() { - logger.V(2).Info("mirror cache stale, using fresh API version", - "cachedRV", existing.GetResourceVersion(), - "freshRV", fresh.GetResourceVersion()) - existing = fresh - } - } - } + // The cached mirror is intentionally NOT re-verified against the apiserver. + // A stale cache read is harmless here: content drift is detected via the + // source hash annotation, and a stale resourceVersion makes the Update below + // fail with a conflict, which requeues the source. A direct read at this + // point costs one apiserver GET per target namespace on every reconcile + // (sources x targets x resync), which overloads the control plane. if err == nil { // Mirror exists - check if it's managed by us @@ -475,13 +469,17 @@ func (r *SourceReconciler) deleteAllMirrors(ctx context.Context, sourceObj metav // cleanupOrphanedMirrors removes mirrors that exist but are no longer in the target list. // This handles cases where target-namespaces annotation changes (e.g., "all" → "all-labeled" or "app-*" → "prod-*"). -func (r *SourceReconciler) cleanupOrphanedMirrors(ctx context.Context, sourceObj metav1.Object, targetNamespaces []string) (int, error) { +// The caller passes the NamespaceInfo it already listed during target +// resolution; a nil nsInfo triggers a fresh list as a fallback. +func (r *SourceReconciler) cleanupOrphanedMirrors(ctx context.Context, sourceObj metav1.Object, targetNamespaces []string, nsInfo *NamespaceInfo) (int, error) { logger := log.FromContext(ctx) - // List all namespaces using unified method for consistency - nsInfo, err := r.NamespaceLister.ListNamespacesWithLabels(ctx) - if err != nil { - return 0, fmt.Errorf("failed to list namespaces: %w", err) + if nsInfo == nil { + var err error + nsInfo, err = r.NamespaceLister.ListNamespacesWithLabels(ctx) + if err != nil { + return 0, fmt.Errorf("failed to list namespaces: %w", err) + } } allNamespaces := nsInfo.All @@ -525,21 +523,23 @@ func (r *SourceReconciler) cleanupOrphanedMirrors(ctx context.Context, sourceObj } // resolveTargetNamespaces determines which namespaces should receive mirrors. -func (r *SourceReconciler) resolveTargetNamespaces(ctx context.Context, sourceObj metav1.Object) ([]string, error) { +// It also returns the NamespaceInfo it listed (nil when resolution ended before +// listing) so callers can reuse it instead of issuing a second namespace LIST. +func (r *SourceReconciler) resolveTargetNamespaces(ctx context.Context, sourceObj metav1.Object) ([]string, *NamespaceInfo, error) { annotations := sourceObj.GetAnnotations() if annotations == nil { - return nil, nil + return nil, nil, nil } targetNsAnnotation := annotations[constants.AnnotationTargetNamespaces] if targetNsAnnotation == "" { - return nil, nil + return nil, nil, nil } // Parse patterns patterns := filter.ParseTargetNamespaces(targetNsAnnotation) if len(patterns) == 0 { - return nil, nil + return nil, nil, nil } // Validate patterns and log warnings for invalid ones @@ -567,14 +567,14 @@ func (r *SourceReconciler) resolveTargetNamespaces(ctx context.Context, sourceOb // If no valid patterns remain, return empty if len(patterns) == 0 { - return nil, nil + return nil, nil, nil } } // Get all namespace info in a single API call (more efficient than 3 separate calls) nsInfo, err := r.NamespaceLister.ListNamespacesWithLabels(ctx) if err != nil { - return nil, fmt.Errorf("failed to list namespaces: %w", err) + return nil, nil, fmt.Errorf("failed to list namespaces: %w", err) } // Resolve target namespaces using the pre-categorized namespace info @@ -592,7 +592,7 @@ func (r *SourceReconciler) resolveTargetNamespaces(ctx context.Context, sourceOb targetNamespaces = targetNamespaces[:r.Config.MaxTargetsPerResource] } - return targetNamespaces, nil + return targetNamespaces, nsInfo, nil } // updateLastSyncStatus updates the source resource's sync-status annotation. diff --git a/pkg/controller/source_reconciler_test.go b/pkg/controller/source_reconciler_test.go index a970672..a3b405f 100644 --- a/pkg/controller/source_reconciler_test.go +++ b/pkg/controller/source_reconciler_test.go @@ -381,7 +381,7 @@ func TestSourceReconciler_resolveTargetNamespaces(t *testing.T) { }, } - got, err := r.resolveTargetNamespaces(context.Background(), sourceObj) + got, _, err := r.resolveTargetNamespaces(context.Background(), sourceObj) if tt.wantError { require.Error(t, err) @@ -555,7 +555,7 @@ func BenchmarkResolveTargetNamespaces(b *testing.B) { b.ResetTimer() b.ReportAllocs() for i := 0; i < b.N; i++ { - _, _ = r.resolveTargetNamespaces(ctx, sourceObj) + _, _, _ = r.resolveTargetNamespaces(ctx, sourceObj) } } @@ -640,7 +640,7 @@ func TestSourceReconciler_cleanupOrphanedMirrors(t *testing.T) { } ctx := context.Background() - deletedCount, err := r.cleanupOrphanedMirrors(ctx, sourceObj, targetNamespaces) + deletedCount, err := r.cleanupOrphanedMirrors(ctx, sourceObj, targetNamespaces, nil) require.NoError(t, err) assert.Equal(t, 1, deletedCount, "should have deleted 1 orphaned mirror") @@ -1273,3 +1273,94 @@ func TestSourceReconciler_updateLastSyncStatus_writesWhenChanged(t *testing.T) { assert.Equal(t, "reconciled:3,errors:0", source.GetAnnotations()[constants.AnnotationSyncStatus]) mockClient.AssertCalled(t, "Update", mock.Anything, mock.Anything, mock.Anything) } + +// countingReader is a client.Reader that counts direct API reads. It stands in +// for the manager's APIReader so tests can assert the reconcile hot path never +// bypasses the informer cache. +type countingReader struct { + gets int + lists int +} + +func (c *countingReader) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + c.gets++ + return errors.NewNotFound(schema.GroupResource{Resource: "secrets"}, key.Name) +} + +func (c *countingReader) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + c.lists++ + return nil +} + +// newMirrorTestSource builds an unstructured Secret source enabled for mirroring. +func newMirrorTestSource(t *testing.T) *unstructured.Unstructured { + t.Helper() + source := &unstructured.Unstructured{} + source.SetGroupVersionKind(schema.GroupVersionKind{Version: "v1", Kind: "Secret"}) + source.SetNamespace("default") + source.SetName("test-secret") + source.SetLabels(map[string]string{constants.LabelEnabled: "true"}) + source.SetAnnotations(map[string]string{ + constants.AnnotationSync: "true", + constants.AnnotationTargetNamespaces: "all", + }) + require.NoError(t, unstructured.SetNestedField(source.Object, "Opaque", "type")) + return source +} + +// TestSourceReconciler_reconcileMirror_NoDirectAPIReads is a load regression +// test: with an up-to-date mirror in the cache, reconcileMirror must complete +// without a single direct API read. A per-target APIReader.Get here multiplies +// into (sources x target namespaces) apiserver GETs on every resync and has +// melted a control-plane node before. +func TestSourceReconciler_reconcileMirror_NoDirectAPIReads(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, "unstructured source must produce an unstructured mirror") + mirror.SetResourceVersion("100") + + mockClient := new(MockClient) + mockClient.On("Get", mock.Anything, types.NamespacedName{Namespace: "target-ns", Name: "test-secret"}, mock.Anything). + Return(nil, mirror) + + reader := &countingReader{} + r := &SourceReconciler{ + Client: mockClient, + APIReader: reader, + Config: &config.Config{VerifySourceFreshness: true}, + GVK: source.GroupVersionKind(), + } + + require.NoError(t, r.reconcileMirror(context.Background(), source, source, "target-ns")) + + assert.Zero(t, reader.gets, "steady-state mirror reconcile must not GET from the apiserver directly") + assert.Zero(t, reader.lists, "steady-state mirror reconcile must not LIST from the apiserver directly") + mockClient.AssertNotCalled(t, "Update", mock.Anything, mock.Anything, mock.Anything) +} + +// TestSourceReconciler_getSourceWithFreshness_DefaultNoDirectRead locks the +// default behaviour: freshness verification is opt-in, so the default config +// must serve the source straight from the informer cache. +func TestSourceReconciler_getSourceWithFreshness_DefaultNoDirectRead(t *testing.T) { + source := newMirrorTestSource(t) + + mockClient := new(MockClient) + mockClient.On("Get", mock.Anything, types.NamespacedName{Namespace: "default", Name: "test-secret"}, mock.Anything). + Return(nil, source) + + reader := &countingReader{} + r := &SourceReconciler{ + Client: mockClient, + APIReader: reader, + Config: &config.Config{}, + GVK: source.GroupVersionKind(), + } + + got, err := r.getSourceWithFreshness(context.Background(), client.ObjectKey{Namespace: "default", Name: "test-secret"}, r.GVK) + require.NoError(t, err) + assert.Equal(t, "test-secret", got.GetName()) + assert.Zero(t, reader.gets, "default config must not verify freshness against the apiserver") +}