mirror of
https://github.com/lukaszraczylo/kubemirror.git
synced 2026-10-17 04:57:29 +00:00
perf(controller): stop per-target apiserver reads in mirror reconcile
The mirror reconcile path issued one direct apiserver GET per target namespace on every pass. With N sources and M target namespaces each resync produced N*M GET requests. On a 65-namespace cluster this generated 63 GET/s and overloaded one control-plane node. Changes: - reconcileMirror now trusts the informer cache. Content drift is detected via the source hash annotation. A stale resourceVersion makes Update return a conflict, which requeues the source. - The verify-source-freshness flag now defaults to false. The Helm chart passes the value explicitly, so the chart default of false matches the binary behaviour. - Reconcile performs one namespace LIST per pass. Target resolution and orphan cleanup share the result. - Add load regression tests that count direct API reads in the reconcile hot path.
This commit is contained in:
@@ -624,7 +624,7 @@ When running the binary directly:
|
|||||||
- `--rate-limit-qps float32` - API rate limit (default: 50.0)
|
- `--rate-limit-qps float32` - API rate limit (default: 50.0)
|
||||||
- `--rate-limit-burst int` - API burst limit (default: 100)
|
- `--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 (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:**
|
**Namespace Filtering:**
|
||||||
- `--excluded-namespaces string` - Comma-separated exclusion list
|
- `--excluded-namespaces string` - Comma-separated exclusion list
|
||||||
|
|||||||
@@ -43,9 +43,7 @@ spec:
|
|||||||
- --worker-threads={{ .Values.controller.workerThreads }}
|
- --worker-threads={{ .Values.controller.workerThreads }}
|
||||||
- --rate-limit-qps={{ .Values.controller.rateLimitQPS }}
|
- --rate-limit-qps={{ .Values.controller.rateLimitQPS }}
|
||||||
- --rate-limit-burst={{ .Values.controller.rateLimitBurst }}
|
- --rate-limit-burst={{ .Values.controller.rateLimitBurst }}
|
||||||
{{- if .Values.controller.verifySourceFreshness }}
|
- --verify-source-freshness={{ .Values.controller.verifySourceFreshness }}
|
||||||
- --verify-source-freshness=true
|
|
||||||
{{- end }}
|
|
||||||
{{- if .Values.controller.lazyWatcherInit }}
|
{{- if .Values.controller.lazyWatcherInit }}
|
||||||
- --lazy-watcher-init=true
|
- --lazy-watcher-init=true
|
||||||
{{- end }}
|
{{- end }}
|
||||||
|
|||||||
@@ -110,12 +110,12 @@ func main() {
|
|||||||
"Burst limit for API server requests.")
|
"Burst limit for API server requests.")
|
||||||
flag.DurationVar(&resyncPeriod, "resync-period", 10*time.Minute,
|
flag.DurationVar(&resyncPeriod, "resync-period", 10*time.Minute,
|
||||||
"Period for resyncing all resources (catches updates missed due to informer cache delays).")
|
"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. "+
|
"Verify source resource freshness by comparing cache with direct API read. "+
|
||||||
"Prevents mirroring stale data and missed orphan cleanups when the informer "+
|
"Converges a few seconds faster when the informer cache lags behind watch "+
|
||||||
"cache lags behind watch events. Trade-off: one extra API call per reconcile "+
|
"events, at the cost of one direct apiserver GET per reconcile. Off by "+
|
||||||
"when the cache is stale. Disable only if you are confident your cluster's "+
|
"default: a stale read self-corrects on the next watch event, and update "+
|
||||||
"watch latency is negligible.")
|
"conflicts requeue automatically.")
|
||||||
flag.BoolVar(&lazyWatcherInit, "lazy-watcher-init", false,
|
flag.BoolVar(&lazyWatcherInit, "lazy-watcher-init", false,
|
||||||
"Enable lazy watcher initialization - only create informers for resource types that have resources marked for mirroring. "+
|
"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. "+
|
"Significantly reduces memory usage by avoiding watchers for unused resource types. "+
|
||||||
|
|||||||
@@ -196,8 +196,9 @@ func (r *NamespaceReconciler) reconcileResourceType(ctx context.Context, rt conf
|
|||||||
// (pattern parsing/validation, namespace listing, max-targets clamping) lives in
|
// (pattern parsing/validation, namespace listing, max-targets clamping) lives in
|
||||||
// exactly one place, mirroring the reconcileMirror delegation below.
|
// exactly one place, mirroring the reconcileMirror delegation below.
|
||||||
func (r *NamespaceReconciler) resolveTargetNamespaces(ctx context.Context, source *unstructured.Unstructured) ([]string, error) {
|
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)
|
resolveTargetNamespaces(ctx, source)
|
||||||
|
return targets, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// reconcileMirror creates or updates a mirror in the target namespace by
|
// reconcileMirror creates or updates a mirror in the target namespace by
|
||||||
|
|||||||
@@ -238,8 +238,9 @@ func (r *SourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
|
|||||||
return ctrl.Result{Requeue: true}, nil
|
return ctrl.Result{Requeue: true}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get target namespaces
|
// Get target namespaces (nsInfo is reused for orphan cleanup below so a
|
||||||
targetNamespaces, err := r.resolveTargetNamespaces(ctx, sourceObj)
|
// reconcile pass performs a single namespace LIST instead of two)
|
||||||
|
targetNamespaces, nsInfo, err := r.resolveTargetNamespaces(ctx, sourceObj)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Error(err, "failed to resolve target namespaces")
|
logger.Error(err, "failed to resolve target namespaces")
|
||||||
if r.CircuitBreaker != nil {
|
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)
|
// 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 {
|
if err != nil {
|
||||||
logger.Error(err, "failed to cleanup orphaned mirrors")
|
logger.Error(err, "failed to cleanup orphaned mirrors")
|
||||||
// Don't fail reconciliation for cleanup errors, just log them
|
// 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)
|
return fmt.Errorf("failed to get existing mirror: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// If freshness verification is enabled and mirror exists, verify it's fresh too
|
// The cached mirror is intentionally NOT re-verified against the apiserver.
|
||||||
if err == nil && r.Config.VerifySourceFreshness && r.APIReader != nil {
|
// A stale cache read is harmless here: content drift is detected via the
|
||||||
fresh := &unstructured.Unstructured{}
|
// source hash annotation, and a stale resourceVersion makes the Update below
|
||||||
fresh.SetGroupVersionKind(sourceUnstructured.GroupVersionKind())
|
// fail with a conflict, which requeues the source. A direct read at this
|
||||||
if apiErr := r.APIReader.Get(ctx, client.ObjectKey{Namespace: targetNs, Name: sourceObj.GetName()}, fresh); apiErr == nil {
|
// point costs one apiserver GET per target namespace on every reconcile
|
||||||
if fresh.GetResourceVersion() != existing.GetResourceVersion() {
|
// (sources x targets x resync), which overloads the control plane.
|
||||||
logger.V(2).Info("mirror cache stale, using fresh API version",
|
|
||||||
"cachedRV", existing.GetResourceVersion(),
|
|
||||||
"freshRV", fresh.GetResourceVersion())
|
|
||||||
existing = fresh
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err == nil {
|
if err == nil {
|
||||||
// Mirror exists - check if it's managed by us
|
// 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.
|
// 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-*").
|
// 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)
|
logger := log.FromContext(ctx)
|
||||||
|
|
||||||
// List all namespaces using unified method for consistency
|
if nsInfo == nil {
|
||||||
nsInfo, err := r.NamespaceLister.ListNamespacesWithLabels(ctx)
|
var err error
|
||||||
if err != nil {
|
nsInfo, err = r.NamespaceLister.ListNamespacesWithLabels(ctx)
|
||||||
return 0, fmt.Errorf("failed to list namespaces: %w", err)
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("failed to list namespaces: %w", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
allNamespaces := nsInfo.All
|
allNamespaces := nsInfo.All
|
||||||
|
|
||||||
@@ -525,21 +523,23 @@ func (r *SourceReconciler) cleanupOrphanedMirrors(ctx context.Context, sourceObj
|
|||||||
}
|
}
|
||||||
|
|
||||||
// resolveTargetNamespaces determines which namespaces should receive mirrors.
|
// 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()
|
annotations := sourceObj.GetAnnotations()
|
||||||
if annotations == nil {
|
if annotations == nil {
|
||||||
return nil, nil
|
return nil, nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
targetNsAnnotation := annotations[constants.AnnotationTargetNamespaces]
|
targetNsAnnotation := annotations[constants.AnnotationTargetNamespaces]
|
||||||
if targetNsAnnotation == "" {
|
if targetNsAnnotation == "" {
|
||||||
return nil, nil
|
return nil, nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse patterns
|
// Parse patterns
|
||||||
patterns := filter.ParseTargetNamespaces(targetNsAnnotation)
|
patterns := filter.ParseTargetNamespaces(targetNsAnnotation)
|
||||||
if len(patterns) == 0 {
|
if len(patterns) == 0 {
|
||||||
return nil, nil
|
return nil, nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Validate patterns and log warnings for invalid ones
|
// 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 no valid patterns remain, return empty
|
||||||
if len(patterns) == 0 {
|
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)
|
// Get all namespace info in a single API call (more efficient than 3 separate calls)
|
||||||
nsInfo, err := r.NamespaceLister.ListNamespacesWithLabels(ctx)
|
nsInfo, err := r.NamespaceLister.ListNamespacesWithLabels(ctx)
|
||||||
if err != nil {
|
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
|
// 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]
|
targetNamespaces = targetNamespaces[:r.Config.MaxTargetsPerResource]
|
||||||
}
|
}
|
||||||
|
|
||||||
return targetNamespaces, nil
|
return targetNamespaces, nsInfo, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// updateLastSyncStatus updates the source resource's sync-status annotation.
|
// updateLastSyncStatus updates the source resource's sync-status annotation.
|
||||||
|
|||||||
@@ -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 {
|
if tt.wantError {
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
@@ -555,7 +555,7 @@ func BenchmarkResolveTargetNamespaces(b *testing.B) {
|
|||||||
b.ResetTimer()
|
b.ResetTimer()
|
||||||
b.ReportAllocs()
|
b.ReportAllocs()
|
||||||
for i := 0; i < b.N; i++ {
|
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()
|
ctx := context.Background()
|
||||||
deletedCount, err := r.cleanupOrphanedMirrors(ctx, sourceObj, targetNamespaces)
|
deletedCount, err := r.cleanupOrphanedMirrors(ctx, sourceObj, targetNamespaces, nil)
|
||||||
|
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Equal(t, 1, deletedCount, "should have deleted 1 orphaned mirror")
|
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])
|
assert.Equal(t, "reconciled:3,errors:0", source.GetAnnotations()[constants.AnnotationSyncStatus])
|
||||||
mockClient.AssertCalled(t, "Update", mock.Anything, mock.Anything, mock.Anything)
|
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")
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user