diff --git a/api/v1alpha1/kollecttarget_types.go b/api/v1alpha1/kollecttarget_types.go index a45cf484..b978feec 100644 --- a/api/v1alpha1/kollecttarget_types.go +++ b/api/v1alpha1/kollecttarget_types.go @@ -78,12 +78,39 @@ type KollectTargetStatus struct { // +optional LastExtractionError string `json:"lastExtractionError,omitempty"` + // collectedCount is the number of resources this target was collecting when the + // controller last refreshed the count. It is the machine-readable source of truth + // for collection scale; the Ready condition message restates it as prose for + // backward compatibility only. + // + // A null value means the controller has never computed a count for this target + // (it has not yet reached Ready). Zero means it computed a count and it was zero. + // A target that is Degraded, or whose reconciles are failing, keeps its last known + // count rather than silently reporting a fresh-looking number — see + // collectedCountUpdatedAt for when that measurement was taken. + // +optional + CollectedCount *int64 `json:"collectedCount,omitempty"` + + // collectedCountUpdatedAt is when collectedCount last *changed* — not when it was + // last checked. A steady target whose count has not moved keeps an old timestamp + // while still being re-derived every resync, so an old timestamp on its own does + // not mean the number is stale. + // + // Read it together with the conditions: a Ready target with an old timestamp has a + // count that genuinely has not moved, while a Degraded target keeps its last known + // count and the timestamp shows how long ago that measurement was taken. + // +optional + CollectedCountUpdatedAt *metav1.Time `json:"collectedCountUpdatedAt,omitempty"` + CollectionFilterStatus `json:",inline"` } // +kubebuilder:object:root=true // +kubebuilder:subresource:status // +kubebuilder:resource:shortName=ktgt +// +kubebuilder:printcolumn:name="Collected",type=integer,JSONPath=`.status.collectedCount` +// +kubebuilder:printcolumn:name="Updated",type=date,JSONPath=`.status.collectedCountUpdatedAt` +// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp` // KollectTarget is the Schema for the kollecttargets API type KollectTarget struct { diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 011049b9..e14a33af 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -1754,6 +1754,15 @@ func (in *KollectTargetStatus) DeepCopyInto(out *KollectTargetStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.CollectedCount != nil { + in, out := &in.CollectedCount, &out.CollectedCount + *out = new(int64) + **out = **in + } + if in.CollectedCountUpdatedAt != nil { + in, out := &in.CollectedCountUpdatedAt, &out.CollectedCountUpdatedAt + *out = (*in).DeepCopy() + } in.CollectionFilterStatus.DeepCopyInto(&out.CollectionFilterStatus) } diff --git a/charts/kollect/README.md b/charts/kollect/README.md index c354b4f0..5d4eed98 100644 --- a/charts/kollect/README.md +++ b/charts/kollect/README.md @@ -30,6 +30,7 @@ unchanged (`readOnlyRootFilesystem: true`, capabilities dropped, `/tmp` `emptyDi | controller.maxConcurrentReconciles.inventory | int | `3` | Max concurrent reconciles for KollectInventory. | | controller.maxConcurrentReconciles.target | int | `5` | Max concurrent reconciles for KollectTarget. | | controller.reconcileRateLimit | string | `""` | | +| controller.targetCountResync | string | `""` | How often a Ready KollectTarget is requeued to refresh `status.collectedCount` (empty = manager default 60s). Nothing else re-enqueues a Target when objects enter or leave its matched set, so this bounds how stale the reported count can be. | | createNamespace | bool | `false` | Create the release namespace if it does not exist. | | defaultExcludedNamespaces | list | `[]` | Default namespace denylist for Target collection intent (CRD fields on KollectTarget override). | | defaultIncludedNamespaces | list | `[]` | Default namespace allowlist for Target collection intent (CRD fields on KollectTarget override). | diff --git a/charts/kollect/crds/kollect.dev_kollecttargets.yaml b/charts/kollect/crds/kollect.dev_kollecttargets.yaml index 64ca0a23..a6090a9a 100644 --- a/charts/kollect/crds/kollect.dev_kollecttargets.yaml +++ b/charts/kollect/crds/kollect.dev_kollecttargets.yaml @@ -16,7 +16,17 @@ spec: singular: kollecttarget scope: Namespaced versions: - - name: v1alpha1 + - additionalPrinterColumns: + - jsonPath: .status.collectedCount + name: Collected + type: integer + - jsonPath: .status.collectedCountUpdatedAt + name: Updated + type: date + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 schema: openAPIV3Schema: description: KollectTarget is the Schema for the kollecttargets API @@ -354,6 +364,32 @@ spec: description: activeResourceRules is the number of compiled resourceRules entries (0 when using legacy fallback). type: integer + collectedCount: + description: |- + collectedCount is the number of resources this target was collecting when the + controller last refreshed the count. It is the machine-readable source of truth + for collection scale; the Ready condition message restates it as prose for + backward compatibility only. + + A null value means the controller has never computed a count for this target + (it has not yet reached Ready). Zero means it computed a count and it was zero. + A target that is Degraded, or whose reconciles are failing, keeps its last known + count rather than silently reporting a fresh-looking number — see + collectedCountUpdatedAt for when that measurement was taken. + format: int64 + type: integer + collectedCountUpdatedAt: + description: |- + collectedCountUpdatedAt is when collectedCount last *changed* — not when it was + last checked. A steady target whose count has not moved keeps an old timestamp + while still being re-derived every resync, so an old timestamp on its own does + not mean the number is stale. + + Read it together with the conditions: a Ready target with an old timestamp has a + count that genuinely has not moved, while a Degraded target keeps its last known + count and the timestamp shows how long ago that measurement was taken. + format: date-time + type: string conditions: description: |- conditions represent the current state of the KollectTarget resource. diff --git a/charts/kollect/templates/deployment.yaml b/charts/kollect/templates/deployment.yaml index 1fdec481..cd39d693 100644 --- a/charts/kollect/templates/deployment.yaml +++ b/charts/kollect/templates/deployment.yaml @@ -58,6 +58,9 @@ spec: {{- if .Values.controller.reconcileRateLimit }} - --reconcile-rate-limit={{ .Values.controller.reconcileRateLimit }} {{- end }} + {{- if .Values.controller.targetCountResync }} + - --target-count-resync={{ .Values.controller.targetCountResync }} + {{- end }} {{- if .Values.pprof.enabled }} - --enable-pprof - --pprof-bind-address={{ .Values.pprof.bindAddress }} diff --git a/charts/kollect/values.yaml b/charts/kollect/values.yaml index bb77084f..6dc84fc5 100644 --- a/charts/kollect/values.yaml +++ b/charts/kollect/values.yaml @@ -118,6 +118,10 @@ controller: informerResyncPeriod: 12h collectMetricsSampleInterval: 30s reconcileRateLimit: "" + # -- How often a Ready KollectTarget is requeued to refresh `status.collectedCount` + # (empty = manager default 60s). Nothing else re-enqueues a Target when objects enter + # or leave its matched set, so this bounds how stale the reported count can be. + targetCountResync: "" resourcesProfile: default diff --git a/cmd/main.go b/cmd/main.go index 786bc48d..994f62b0 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -224,6 +224,7 @@ func main() { MaxConcurrentClusterTarget: cfg.maxConcurrentClusterTarget, MaxConcurrentClusterInventory: cfg.maxConcurrentClusterInventory, ReconcileRateLimitBase: cfg.reconcileRateLimit, + TargetCountResync: cfg.targetCountResync, } if err := (&controller.KollectTargetReconciler{ diff --git a/cmd/startup_flags.go b/cmd/startup_flags.go index 9be25395..c85f6de7 100644 --- a/cmd/startup_flags.go +++ b/cmd/startup_flags.go @@ -10,6 +10,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" + "github.com/platformrelay/kollect/internal/controller" "github.com/platformrelay/kollect/internal/inventory" "github.com/platformrelay/kollect/internal/validation" ) @@ -40,6 +41,7 @@ type startupConfig struct { maxConcurrentClusterTarget int maxConcurrentClusterInventory int reconcileRateLimit time.Duration + targetCountResync time.Duration enablePprof bool pprofAddr string watchNamespacesRaw string @@ -126,6 +128,8 @@ func bindStartupFlags(fs *flag.FlagSet, cfg *startupConfig) { "Max concurrent KollectClusterInventory reconciles.") fs.DurationVar(&cfg.reconcileRateLimit, "reconcile-rate-limit", 0, "Base delay for per-item exponential reconcile failure rate limiting (0 = controller-runtime default 5ms).") + fs.DurationVar(&cfg.targetCountResync, "target-count-resync", controller.DefaultTargetCountResync, + "How often a Ready KollectTarget is requeued to refresh status.collectedCount (0 = default 60s).") fs.BoolVar(&cfg.enablePprof, "enable-pprof", false, "Expose Go pprof on --pprof-bind-address (separate from metrics).") fs.StringVar(&cfg.pprofAddr, "pprof-bind-address", ":6060", diff --git a/cmd/startup_flags_test.go b/cmd/startup_flags_test.go index 0164619a..7cc7be15 100644 --- a/cmd/startup_flags_test.go +++ b/cmd/startup_flags_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/platformrelay/kollect/internal/controller" "github.com/platformrelay/kollect/internal/inventory" "github.com/platformrelay/kollect/internal/validation" @@ -51,6 +52,9 @@ func TestBindStartupFlags_Defaults(t *testing.T) { cfg.collectDispatchQueueSize, ) } + if cfg.targetCountResync != controller.DefaultTargetCountResync { + t.Fatalf("targetCountResync = %s, want %s", cfg.targetCountResync, controller.DefaultTargetCountResync) + } if cfg.informerResyncPeriod != 12*time.Hour || cfg.collectMetricsSampleInterval != 30*time.Second { t.Fatalf( "unexpected duration defaults: resync=%s sample=%s", diff --git a/config/crd/bases/kollect.dev_kollecttargets.yaml b/config/crd/bases/kollect.dev_kollecttargets.yaml index 64ca0a23..a6090a9a 100644 --- a/config/crd/bases/kollect.dev_kollecttargets.yaml +++ b/config/crd/bases/kollect.dev_kollecttargets.yaml @@ -16,7 +16,17 @@ spec: singular: kollecttarget scope: Namespaced versions: - - name: v1alpha1 + - additionalPrinterColumns: + - jsonPath: .status.collectedCount + name: Collected + type: integer + - jsonPath: .status.collectedCountUpdatedAt + name: Updated + type: date + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 schema: openAPIV3Schema: description: KollectTarget is the Schema for the kollecttargets API @@ -354,6 +364,32 @@ spec: description: activeResourceRules is the number of compiled resourceRules entries (0 when using legacy fallback). type: integer + collectedCount: + description: |- + collectedCount is the number of resources this target was collecting when the + controller last refreshed the count. It is the machine-readable source of truth + for collection scale; the Ready condition message restates it as prose for + backward compatibility only. + + A null value means the controller has never computed a count for this target + (it has not yet reached Ready). Zero means it computed a count and it was zero. + A target that is Degraded, or whose reconciles are failing, keeps its last known + count rather than silently reporting a fresh-looking number — see + collectedCountUpdatedAt for when that measurement was taken. + format: int64 + type: integer + collectedCountUpdatedAt: + description: |- + collectedCountUpdatedAt is when collectedCount last *changed* — not when it was + last checked. A steady target whose count has not moved keeps an old timestamp + while still being re-derived every resync, so an old timestamp on its own does + not mean the number is stale. + + Read it together with the conditions: a Ready target with an old timestamp has a + count that genuinely has not moved, while a Degraded target keeps its last known + count and the timestamp shows how long ago that measurement was taken. + format: date-time + type: string conditions: description: |- conditions represent the current state of the KollectTarget resource. diff --git a/docs/adr/0603-performance-scalability.md b/docs/adr/0603-performance-scalability.md index 8dd01569..d53cdbb5 100644 --- a/docs/adr/0603-performance-scalability.md +++ b/docs/adr/0603-performance-scalability.md @@ -55,7 +55,11 @@ operator**. 6. **Dispatch pool:** Tunable `--collect-dispatch-workers` / queue; enqueue wait, then block (backpressure) on the informer goroutine until queue capacity frees or ctx cancels — never processes inline, to keep worker concurrency the only path that does extract/access-check work. -7. **Resync / metrics sampling:** `--informer-resync-period`; `--collect-metrics-sample-interval`. +7. **Resync / metrics sampling:** `--informer-resync-period`; `--collect-metrics-sample-interval`; + `--target-count-resync` (default **60s**) requeues each Ready `KollectTarget` to refresh + `status.collectedCount` — costs one cluster-wide namespace `LIST` per Target per interval, **two** + when a `KollectScope` is enforced on that Target's namespace, so it is the knob to raise on large + fleets. 8. **Profiling:** Optional `--enable-pprof` on `:6060`; disabled in production Helm values. 9. **Tests:** `load`-tagged tests to **10k** (nightly when 8-core runners exist); 100k manual design proof only. 10. **100k claim gate:** Export sharding enforced + Postgres bulk upsert + **10k nightly green** (once re-enabled). diff --git a/docs/operator-manual/performance.md b/docs/operator-manual/performance.md index 0cbfb455..9996e273 100644 --- a/docs/operator-manual/performance.md +++ b/docs/operator-manual/performance.md @@ -47,6 +47,32 @@ exponential failure rate limiter (5ms base, 1000s cap). Set a positive duration `kollect_workqueue_depth` approximates queue pressure as **in-flight reconciles** per controller (not the internal client-go queue length). +## Reported collection scale + +**`KollectTarget.status.collectedCount`** is the machine-readable number of resources a Target is +collecting, surfaced as the `COLLECTED` column of `kubectl get kollecttargets`. The `Ready` +condition message restates it as prose for backward compatibility only. + +Objects entering or leaving a Target's matched set do **not** enqueue that Target, so the number is +refreshed by a periodic self-requeue: **`--target-count-resync`** (default **`60s`**, Helm +`controller.targetCountResync`). + +**Budget the resync — it is not free.** Each pass costs **one live, cluster-wide, unpaginated +namespace `LIST`** — **two** when a `KollectScope` is enforced on the Target's namespace, because +the scope check resolves the filter status as well. At the default interval, **N** Targets therefore +cost **N** namespace `LIST`s per minute — **2N** under an enforced scope. The rest is cheap: the +engine skips the informer backfill when the Target's state is unchanged, and the status write is +skipped when the number did not move. + +**`status.collectedCountUpdatedAt`** records when the number last *changed* — **not** when it was +last checked. A steady Target keeps an old timestamp while still being re-derived every resync, so +an old timestamp on its own does not mean the count is stale. + +Judge liveness from the timestamp and the conditions together: a `Ready` Target with an old +timestamp has a count that genuinely has not moved, while a `Degraded` Target keeps its last known +count and the timestamp shows how old that measurement is. Lower the interval for a more responsive +count; raise it to cut reconcile volume on very large fleets. + ## Export debouncing **`KollectInventory.spec.exportMinInterval`** (default **`30s`**) coalesces export to external sinks diff --git a/docs/reference/conditions.md b/docs/reference/conditions.md index a5bb27b3..deac10f3 100644 --- a/docs/reference/conditions.md +++ b/docs/reference/conditions.md @@ -10,6 +10,14 @@ Conditions describe the latest observed generation; use them with Events and con | `Synced` | The latest observed inventory was exported successfully. | | `Degraded` | A terminal or partial failure needs attention. | +**`KollectTarget` `Ready.lastTransitionTime` is not a flap signal.** The `Ready` message restates +the live collected count, and the controller treats a changed message as a transition. A busy +Target whose count moves therefore gets a fresh `lastTransitionTime` on every refresh — as often as +`--target-count-resync` (default `60s`) — while `Ready` never leaves `True`. Do not alert on it as +if the Target had flapped, and do not read an *old* `lastTransitionTime` as "collection has +stalled": it only means the count has not moved. Use `status.collectedCountUpdatedAt` and the +`Degraded` condition instead. + `KollectInventory.status.sinkExports[]` records timestamps, checksums, and conditions per sink. This distinguishes partial fan-out from total failure. Full collected payloads never live in CR status. diff --git a/internal/collect/engine.go b/internal/collect/engine.go index 1278af58..43bc6a56 100644 --- a/internal/collect/engine.go +++ b/internal/collect/engine.go @@ -223,7 +223,12 @@ func NewEngine( // RegisterTargetOptions carries resolved namespace and rule state for collection filtering. type RegisterTargetOptions struct { - ScopeCeiling ScopeCeiling + ScopeCeiling ScopeCeiling + // EffectiveNamespaces is the namespace set the caller already resolved. Supplying it + // carries a contract: the caller is responsible for the freshness of the engine's + // namespace metadata cache (call RefreshNamespaces first), because RegisterTarget + // then skips the cluster-wide namespace LIST. Leave it empty to have the engine + // refresh and recompute the set itself. EffectiveNamespaces []string } @@ -281,10 +286,6 @@ func (e *Engine) RegisterTarget( return nil } - if err := e.refreshNamespaceCache(ctx); err != nil { - log.FromContext(ctx).Error(err, "refresh namespace cache") - } - gvr := gvrFromProfile(profile.Spec.TargetGVK) compiled, err := CompileResourceRules(target.Spec.ResourceRules, e.extractor.celEnv) @@ -303,6 +304,16 @@ func (e *Engine) RegisterTarget( if len(effective) == 0 { namespaceSource = "recomputed" + // Only the recompute branch reads the namespace cache here, so only it has to + // pay for a live cluster-wide namespace LIST. Callers that supply + // EffectiveNamespaces have already resolved the set from a snapshot they + // refreshed themselves (RefreshNamespaces), and reconcilers re-register on every + // pass — refreshing unconditionally made every resync a LIST for every target + // (PERF-FIX-05 review finding F2). + if err := e.refreshNamespaceCache(ctx); err != nil { + log.FromContext(ctx).Error(err, "refresh namespace cache") + } + e.nsMu.RLock() matched := MatchIntentNamespaces( target.Spec.CollectionFilterSpec, diff --git a/internal/collect/register_namespace_list_test.go b/internal/collect/register_namespace_list_test.go new file mode 100644 index 00000000..11fd7fc4 --- /dev/null +++ b/internal/collect/register_namespace_list_test.go @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Konrad Heimel + +package collect + +import ( + "context" + "sync/atomic" + "testing" + + "k8s.io/apimachinery/pkg/runtime" + kubefake "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +// Reconcilers re-register their targets on every pass, and PERF-FIX-05 makes every Ready +// KollectTarget reconcile on a timer. RegisterTarget used to refresh the namespace cache +// unconditionally, which turned each of those passes into a live, cluster-wide, +// unpaginated namespace LIST per target — the cost this test pins to zero for callers +// that resolved the namespace set themselves (review finding F2). +// +// The recompute branch still has to LIST: it reads the cache it is about to match +// against, so a stale snapshot there silently drops objects in newly created namespaces. +func TestRegisterTargetNamespaceListIsOnlyPaidOnTheRecomputeBranch(t *testing.T) { + t.Parallel() + + engine, _, profile := newScopeTransitionEngine(t) + + kube, ok := engine.kube.(*kubefake.Clientset) + if !ok { + t.Fatalf("engine.kube = %T, want *kubefake.Clientset", engine.kube) + } + + var namespaceLists atomic.Int64 + kube.PrependReactor("list", "namespaces", func(k8stesting.Action) (bool, runtime.Object, error) { + namespaceLists.Add(1) + + return false, nil, nil // fall through to the tracker + }) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + if err := engine.Start(ctx); err != nil { + t.Fatalf("Start: %v", err) + } + + supplied := scopeTransitionTarget("supplied-ns", "team-a") + if err := engine.RegisterTarget(ctx, supplied, profile, RegisterTargetOptions{ + EffectiveNamespaces: []string{"team-a"}, + }); err != nil { + t.Fatalf("register with supplied namespaces: %v", err) + } + if got := namespaceLists.Load(); got != 0 { + t.Fatalf("namespace LISTs with EffectiveNamespaces supplied = %d, want 0", got) + } + + // A re-registration is the resync shape: same state, still no LIST. + if err := engine.RegisterTarget(ctx, supplied, profile, RegisterTargetOptions{ + EffectiveNamespaces: []string{"team-a"}, + }); err != nil { + t.Fatalf("re-register with supplied namespaces: %v", err) + } + if got := namespaceLists.Load(); got != 0 { + t.Fatalf("namespace LISTs after a resync re-registration = %d, want 0", got) + } + + recomputed := scopeTransitionTarget("recomputed-ns", "team-b") + if err := engine.RegisterTarget(ctx, recomputed, profile, RegisterTargetOptions{}); err != nil { + t.Fatalf("register without namespaces: %v", err) + } + if got := namespaceLists.Load(); got != 1 { + t.Fatalf("namespace LISTs on the recompute branch = %d, want 1", got) + } +} diff --git a/internal/controller/conditions.go b/internal/controller/conditions.go index e71ee04d..25f2c6f1 100644 --- a/internal/controller/conditions.go +++ b/internal/controller/conditions.go @@ -44,6 +44,13 @@ const ( reasonCollecting = "Collecting" ) +// setTargetCondition writes conditionType into conditions and persists the whole status +// subresource, skipping the API call when nothing about the condition moved. +// +// It reports whether it issued that call. Status carries fields no condition describes +// (KollectTarget.status.collectedCount), and a caller that changed one of those is +// responsible for persisting it when this skipped the write — otherwise the value stays +// in memory and the API server keeps serving the previous one (PERF-FIX-05 / F-05). func setTargetCondition( ctx context.Context, c client.Client, @@ -53,14 +60,14 @@ func setTargetCondition( conditionType string, status metav1.ConditionStatus, reason, message string, -) error { +) (written bool, err error) { existing := apimeta.FindStatusCondition(*conditions, conditionType) if existing != nil && existing.Status == status && existing.Reason == reason && existing.Message == message && existing.ObservedGeneration == generation { - return nil + return false, nil } next := metav1.Condition{ @@ -80,5 +87,5 @@ func setTargetCondition( apimeta.SetStatusCondition(conditions, next) - return c.Status().Update(ctx, target) + return true, c.Status().Update(ctx, target) } diff --git a/internal/controller/conditions_test.go b/internal/controller/conditions_test.go index 90dc70a2..64f8719d 100644 --- a/internal/controller/conditions_test.go +++ b/internal/controller/conditions_test.go @@ -47,12 +47,16 @@ func TestSetTargetCondition_skipsUnchanged(t *testing.T) { bgCtx := context.Background() msg := "profileRef \"apps\" resolved; collecting 3 resource(s)" - if err := setTargetCondition( + written, err := setTargetCondition( bgCtx, cl, target, 2, &target.Status.Conditions, conditionReady, metav1.ConditionTrue, "Collecting", msg, - ); err != nil { + ) + if err != nil { t.Fatalf("setTargetCondition: %v", err) } + if written { + t.Fatal("setTargetCondition reported a write for an unchanged condition") + } ready := apimeta.FindStatusCondition(target.Status.Conditions, conditionReady) if ready == nil { diff --git a/internal/controller/kollectclustertarget_controller.go b/internal/controller/kollectclustertarget_controller.go index 6afd9683..5938285d 100644 --- a/internal/controller/kollectclustertarget_controller.go +++ b/internal/controller/kollectclustertarget_controller.go @@ -188,6 +188,16 @@ func (r *KollectClusterTargetReconciler) syncEngineTargets( effective []string, ceiling collect.ScopeCeiling, ) error { + // The engine's namespace metadata cache backs resource-rule evaluation and the + // namespace-level watch opt-in/opt-out in ShouldCollect, and this controller resolves + // its own namespaces from the cached client rather than that snapshot. RegisterTarget + // only refreshes it when it has to recompute the namespace set, which it never does + // here — every synthetic target is registered with an explicit namespace. So refresh + // once per reconcile, ahead of the loop, instead of once per registered namespace. + if err := r.Engine.RefreshNamespaces(ctx); err != nil { + return fmt.Errorf("refresh namespace cache before cluster target registration: %w", err) + } + want := make(map[string]struct{}, len(effective)) for _, ns := range effective { want[ns] = struct{}{} diff --git a/internal/controller/kollectclustertarget_controller_test.go b/internal/controller/kollectclustertarget_controller_test.go index baed37da..6a235776 100644 --- a/internal/controller/kollectclustertarget_controller_test.go +++ b/internal/controller/kollectclustertarget_controller_test.go @@ -155,6 +155,97 @@ var _ = Describe("KollectClusterTarget Controller", func() { Expect(engine.NamespacesForClusterTarget(targetName)).To(ConsistOf(nsMatched)) }) + // Review finding F-E. syncEngineTargets refreshes the engine's namespace metadata + // cache before registering, and that call is a compensating control this lane's own + // change created: RegisterTarget no longer refreshes when the caller supplies + // EffectiveNamespaces, and this controller always does. Nothing else on the + // cluster-target path populates that cache — it resolves its own namespaces from the + // cached client, which does not even carry annotations. + // + // The cache is what ShouldCollect reads to honour `kollect.dev/namespace-watch: + // disabled`. Drop the refresh and the annotation becomes invisible: the namespace is + // still registered and its objects are collected anyway. That is over-collection with + // no error anywhere — an operator's explicit opt-out silently ignored — so assert the + // observable outcome rather than the cache contents. + It("honours a namespace watch opt-out for a cluster target", func() { + ensureNamespace(ctx, kubeClient, nsMatched, map[string]string{tenantLabel: tenantValue}) + + // Same tenant label, so this namespace IS in the target's scope and DOES get a + // synthetic target registered. Only the annotation should keep its objects out — + // and annotations reach the engine exclusively through the namespace cache. + optedOut := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: nsOther, + Labels: map[string]string{tenantLabel: tenantValue}, + Annotations: map[string]string{kollectdevv1alpha1.AnnotationNamespaceWatch: kollectdevv1alpha1.WatchValueDisabled}, + }, + } + _, err := kubeClient.CoreV1().Namespaces().Create(ctx, optedOut, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) + + for ns, name := range map[string]string{nsMatched: "cm-collected", nsOther: "cm-opted-out"} { + _, err = kubeClient.CoreV1().ConfigMaps(ns).Create(ctx, &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}, + Data: map[string]string{"name": name}, + }, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) + } + + ensureNamespace(ctx, kubeClient, sink.DefaultSecretNamespace, nil) + + profile := &kollectdevv1alpha1.KollectProfile{ + ObjectMeta: metav1.ObjectMeta{Name: profileName, Namespace: sink.DefaultSecretNamespace}, + Spec: kollectdevv1alpha1.KollectProfileSpec{ + TargetGVK: kollectdevv1alpha1.GroupVersionKind{Version: "v1", Kind: "ConfigMap"}, + Attributes: []kollectdevv1alpha1.AttributeSpec{ + {Name: "name", Path: "{.metadata.name}"}, + }, + }, + } + Expect(k8sClient.Create(ctx, profile)).To(Succeed()) + + target := &kollectdevv1alpha1.KollectClusterTarget{ + ObjectMeta: metav1.ObjectMeta{Name: targetName}, + Spec: kollectdevv1alpha1.KollectClusterTargetSpec{ + ProfileRef: kollectdevv1alpha1.NamespacedObjectReference{ + Name: profileName, + Namespace: sink.DefaultSecretNamespace, + }, + NamespaceSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{tenantLabel: tenantValue}, + }, + }, + } + Expect(k8sClient.Create(ctx, target)).To(Succeed()) + + reconciler := &KollectClusterTargetReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Engine: engine, + } + _, err = reconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: targetName}, + }) + Expect(err).NotTo(HaveOccurred()) + + // Both namespaces are registered. Without this, a zero count below could mean + // "never in scope" rather than "opt-out honoured", and the test would pass for + // the wrong reason. + Expect(engine.NamespacesForClusterTarget(targetName)).To(ConsistOf(nsMatched, nsOther)) + + // Positive control: the pipeline demonstrably works and has drained the informer's + // initial Adds, so the opted-out namespace has had its chance to be collected. + Eventually(func() int { + return engine.ItemCount(nsMatched, targetName) + }, 30*time.Second, 200*time.Millisecond).Should(Equal(1)) + + Consistently(func() int { + return engine.ItemCount(nsOther, targetName) + }, 2*time.Second, 200*time.Millisecond).Should(BeZero(), + "the namespace watch opt-out must be honoured; a non-zero count means the "+ + "engine's namespace cache was never refreshed and the annotation was invisible") + }) + It("re-enqueues targets when the referenced profile changes", func() { ensureNamespace(ctx, kubeClient, sink.DefaultSecretNamespace, nil) diff --git a/internal/controller/kollectclustertarget_nsrefresh_test.go b/internal/controller/kollectclustertarget_nsrefresh_test.go new file mode 100644 index 00000000..aa8862fa --- /dev/null +++ b/internal/controller/kollectclustertarget_nsrefresh_test.go @@ -0,0 +1,100 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Konrad Heimel + +package controller + +import ( + "context" + "errors" + "strings" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + dynfake "k8s.io/client-go/dynamic/fake" + kubefake "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + kollectdevv1alpha1 "github.com/platformrelay/kollect/api/v1alpha1" + "github.com/platformrelay/kollect/internal/collect" +) + +// The namespace refresh in syncEngineTargets is a compensating control: RegisterTarget no +// longer refreshes when the caller supplies EffectiveNamespaces, and this controller +// always does. If the refresh fails, registering anyway would mean collecting against a +// stale namespace cache — namespace watch opt-outs invisible, objects collected that an +// operator explicitly excluded, and no error to notice. Failing the sync instead surfaces +// it as Degraded/InformerRegistrationFailed and retries. +func TestSyncEngineTargetsFailsWhenTheNamespaceRefreshFails(t *testing.T) { + t.Parallel() + + scheme := runtime.NewScheme() + if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + + kube := kubefake.NewSimpleClientset() //nolint:staticcheck // SimpleClientset is sufficient here + listErr := errors.New("namespace list refused") + kube.PrependReactor("list", "namespaces", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, listErr + }) + + // A working dynamic client on purpose: registration must be prevented by the refresh + // failure itself, not by the informer being unable to start. Without this the test + // would "fail" on a nil-client panic under mutation and prove nothing. + dyn := dynfake.NewSimpleDynamicClientWithCustomListKinds( + runtime.NewScheme(), + map[schema.GroupVersionResource]string{ + {Version: "v1", Resource: "configmaps"}: "ConfigMapList", + }, + ) + + engine, err := collect.NewEngine(dyn, kube, collect.NewStore(), collect.EngineConfig{}) + if err != nil { + t.Fatalf("NewEngine: %v", err) + } + + engineCtx, cancelEngine := context.WithCancel(context.Background()) + t.Cleanup(cancelEngine) + if startErr := engine.Start(engineCtx); startErr != nil { + t.Fatalf("Start: %v", startErr) + } + + r := &KollectClusterTargetReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).Build(), + Scheme: scheme, + Engine: engine, + } + + ct := &kollectdevv1alpha1.KollectClusterTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "fleet"}, + Spec: kollectdevv1alpha1.KollectClusterTargetSpec{ + ProfileRef: kollectdevv1alpha1.NamespacedObjectReference{Name: "apps", Namespace: "kollect-system"}, + }, + } + profile := &kollectdevv1alpha1.KollectProfile{ + ObjectMeta: metav1.ObjectMeta{Name: "apps", Namespace: "kollect-system"}, + Spec: kollectdevv1alpha1.KollectProfileSpec{ + TargetGVK: kollectdevv1alpha1.GroupVersionKind{Version: "v1", Kind: "ConfigMap"}, + }, + } + + syncErr := r.syncEngineTargets(context.Background(), ct, profile, []string{"team-a"}, collect.ScopeCeiling{}) + if syncErr == nil { + t.Fatal("syncEngineTargets returned nil; a failed namespace refresh must not register against a stale cache") + } + if !errors.Is(syncErr, listErr) { + t.Fatalf("syncEngineTargets error = %v, want it to wrap %v", syncErr, listErr) + } + if !strings.Contains(syncErr.Error(), "refresh namespace cache") { + t.Fatalf("error %q should name the failed step so the Degraded message is actionable", syncErr) + } + + // Nothing was registered, so a later successful reconcile starts clean rather than + // leaving a target bound to namespaces resolved from a cache that never loaded. + if got := engine.NamespacesForClusterTarget(ct.Name); len(got) != 0 { + t.Fatalf("registered namespaces = %v, want none", got) + } +} diff --git a/internal/controller/kollecttarget_collected_count_test.go b/internal/controller/kollecttarget_collected_count_test.go new file mode 100644 index 00000000..0c34bc2a --- /dev/null +++ b/internal/controller/kollecttarget_collected_count_test.go @@ -0,0 +1,295 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Konrad Heimel + +package controller + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + kollectdevv1alpha1 "github.com/platformrelay/kollect/api/v1alpha1" +) + +// collectedCountFixture builds a target that already reached Ready reporting `stored` +// resources, together with a fake client holding it — the state a long-running target +// is in when objects start entering or leaving its matched set. +func collectedCountFixture(t *testing.T, stored int64, lastUpdate metav1.Time) ( + *kollectdevv1alpha1.KollectTarget, client.Client, +) { + t.Helper() + + count := stored + target := &kollectdevv1alpha1.KollectTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "demo", Namespace: "team-a", Generation: 2}, + Spec: kollectdevv1alpha1.KollectTargetSpec{ProfileRef: "apps"}, + Status: kollectdevv1alpha1.KollectTargetStatus{ + ObservedGeneration: 2, + CollectedCount: &count, + CollectedCountUpdatedAt: &lastUpdate, + Conditions: []metav1.Condition{{ + Type: conditionReady, + Status: metav1.ConditionTrue, + Reason: reasonCollecting, + Message: readyMessageFor(stored), + ObservedGeneration: 2, + LastTransitionTime: lastUpdate, + }}, + }, + } + + scheme := runtime.NewScheme() + if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(target). + WithStatusSubresource(target). + Build() + + return target, cl +} + +func readyMessageFor(count int64) string { + return fmt.Sprintf("profileRef %q resolved; collecting %d resource(s)", "apps", count) +} + +// PERF-FIX-05 / F-05: a target whose matched set grew must persist the new number. +// setTargetCondition skips the API write for an unchanged Ready condition, so this is +// the regression guard for "the count changed but nothing reached the API server". +func TestSetReadyPersistsGrowingCollectedCount(t *testing.T) { + t.Parallel() + + stale := metav1.NewTime(time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)) + target, cl := collectedCountFixture(t, 1000, stale) + + r := &KollectTargetReconciler{Client: cl} + result, err := r.setReady(context.Background(), target, 1200, "", "", "", "") + if err != nil { + t.Fatalf("setReady: %v", err) + } + if result.RequeueAfter != DefaultTargetCountResync { + t.Fatalf("RequeueAfter = %s, want %s", result.RequeueAfter, DefaultTargetCountResync) + } + + var stored kollectdevv1alpha1.KollectTarget + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(target), &stored); err != nil { + t.Fatalf("get: %v", err) + } + if stored.Status.CollectedCount == nil || *stored.Status.CollectedCount != 1200 { + t.Fatalf("persisted collectedCount = %v, want 1200", stored.Status.CollectedCount) + } + if stored.Status.CollectedCountUpdatedAt == nil || !stored.Status.CollectedCountUpdatedAt.After(stale.Time) { + t.Fatalf("collectedCountUpdatedAt = %v, want newer than %v", stored.Status.CollectedCountUpdatedAt, stale) + } + if !strings.Contains(stored.Status.Conditions[0].Message, "collecting 1200 resource(s)") { + t.Fatalf("Ready message = %q, want it to restate the live count", stored.Status.Conditions[0].Message) + } +} + +// Upgrade path (independent review, finding F1). A target Ready and genuinely collecting +// zero, upgraded from a binary that already wrote "collecting 0 resource(s)" into the +// Ready message, has a condition byte-identical to the one this reconcile would write — +// so setTargetCondition skips the API call. If persisting the count rode on that call, +// the measured zero would stay absent forever and `kubectl get` would show COLLECTED +// , which the shipped CRD documentation defines as "never computed". No number of +// resyncs would fix it, because the count never moves. +func TestSetReadyPersistsMeasuredZeroWhenConditionIsUnchanged(t *testing.T) { + t.Parallel() + + old := metav1.NewTime(time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)) + seed := &kollectdevv1alpha1.KollectTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "demo", Namespace: "team-a", Generation: 2}, + Spec: kollectdevv1alpha1.KollectTargetSpec{ProfileRef: "apps"}, + Status: kollectdevv1alpha1.KollectTargetStatus{ + ObservedGeneration: 2, + // collectedCount/collectedCountUpdatedAt absent: written by the old binary. + Conditions: []metav1.Condition{{ + Type: conditionReady, + Status: metav1.ConditionTrue, + Reason: reasonCollecting, + Message: readyMessageFor(0), + ObservedGeneration: 2, + LastTransitionTime: old, + }}, + }, + } + + scheme := runtime.NewScheme() + if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(seed). + WithStatusSubresource(seed). + Build() + + live := &kollectdevv1alpha1.KollectTarget{} + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(seed), live); err != nil { + t.Fatalf("get: %v", err) + } + + r := &KollectTargetReconciler{Client: cl} + for range 10 { // ten resyncs; the count genuinely never moves + if _, err := r.setReady(context.Background(), live, 0, "", "", "", ""); err != nil { + t.Fatalf("setReady: %v", err) + } + } + + var stored kollectdevv1alpha1.KollectTarget + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(seed), &stored); err != nil { + t.Fatalf("get: %v", err) + } + if stored.Status.CollectedCount == nil { + t.Fatal("after 10 resyncs collectedCount is still absent — " + + "a condition-gated write swallowed the measured zero") + } + if *stored.Status.CollectedCount != 0 { + t.Fatalf("persisted collectedCount = %d, want 0", *stored.Status.CollectedCount) + } +} + +// The forced write is the only thing carrying the count to the API server on this path, +// so its failure must surface as a reconcile error and get retried. Swallowing it would +// drop the number silently and leave status disagreeing with the cluster until it next +// happened to move — the same class of defect this lane exists to remove. +func TestSetReadyReturnsErrorWhenTheForcedCountWriteFails(t *testing.T) { + t.Parallel() + + old := metav1.NewTime(time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)) + seed := &kollectdevv1alpha1.KollectTarget{ + ObjectMeta: metav1.ObjectMeta{Name: "demo", Namespace: "team-a", Generation: 2}, + Spec: kollectdevv1alpha1.KollectTargetSpec{ProfileRef: "apps"}, + Status: kollectdevv1alpha1.KollectTargetStatus{ + ObservedGeneration: 2, + Conditions: []metav1.Condition{{ + Type: conditionReady, + Status: metav1.ConditionTrue, + Reason: reasonCollecting, + Message: readyMessageFor(0), + ObservedGeneration: 2, + LastTransitionTime: old, + }}, + }, + } + + scheme := runtime.NewScheme() + if err := kollectdevv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + + wantErr := errors.New("status update rejected") + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(seed). + WithStatusSubresource(seed). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourceUpdate: func( + context.Context, client.Client, string, client.Object, ...client.SubResourceUpdateOption, + ) error { + return wantErr + }, + }). + Build() + + live := &kollectdevv1alpha1.KollectTarget{} + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(seed), live); err != nil { + t.Fatalf("get: %v", err) + } + + // The Ready condition is byte-identical, so setTargetCondition skips its write and + // the forced count write is the only call that can fail here. + r := &KollectTargetReconciler{Client: cl} + if _, err := r.setReady(context.Background(), live, 0, "", "", "", ""); !errors.Is(err, wantErr) { + t.Fatalf("setReady error = %v, want %v", err, wantErr) + } +} + +// Objects leaving the selector must shrink the number, including all the way to zero — +// which is why collectedCount is a pointer: an absent field would be indistinguishable +// from "never measured". +func TestSetReadyPersistsShrinkingCollectedCount(t *testing.T) { + t.Parallel() + + stale := metav1.NewTime(time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)) + target, cl := collectedCountFixture(t, 1200, stale) + + r := &KollectTargetReconciler{Client: cl} + if _, err := r.setReady(context.Background(), target, 0, "", "", "", ""); err != nil { + t.Fatalf("setReady: %v", err) + } + + var stored kollectdevv1alpha1.KollectTarget + if err := cl.Get(context.Background(), client.ObjectKeyFromObject(target), &stored); err != nil { + t.Fatalf("get: %v", err) + } + if stored.Status.CollectedCount == nil { + t.Fatal("collectedCount is nil; a measured zero must be distinguishable from unmeasured") + } + if *stored.Status.CollectedCount != 0 { + t.Fatalf("persisted collectedCount = %d, want 0", *stored.Status.CollectedCount) + } +} + +// An unchanged count must not churn the timestamp: collectedCountUpdatedAt means "when +// the number last moved", and refreshing it every resync would make a frozen count look +// live — the exact failure this story exists to remove. +func TestSyncCollectedCountKeepsTimestampWhenUnchanged(t *testing.T) { + t.Parallel() + + stale := metav1.NewTime(time.Date(2026, 6, 5, 12, 0, 0, 0, time.UTC)) + count := int64(1000) + target := &kollectdevv1alpha1.KollectTarget{ + Status: kollectdevv1alpha1.KollectTargetStatus{ + CollectedCount: &count, + CollectedCountUpdatedAt: &stale, + }, + } + + got, changed := syncCollectedCount(target, 1000) + if got != 1000 { + t.Fatalf("syncCollectedCount = %d, want 1000", got) + } + if changed { + t.Fatal("syncCollectedCount reported a change for an unchanged count") + } + if !target.Status.CollectedCountUpdatedAt.Equal(&stale) { + t.Fatalf("timestamp = %v, want unchanged %v", target.Status.CollectedCountUpdatedAt, stale) + } +} + +// A target the controller has never counted reports nothing rather than a misleading zero. +func TestSyncCollectedCountFirstObservationSetsTimestamp(t *testing.T) { + t.Parallel() + + target := &kollectdevv1alpha1.KollectTarget{} + if target.Status.CollectedCount != nil { + t.Fatal("fresh target must not carry a count") + } + + got, changed := syncCollectedCount(target, 3) + if got != 3 { + t.Fatalf("syncCollectedCount = %d, want 3", got) + } + if !changed { + t.Fatal("the first observation must report a change so the caller persists it") + } + if target.Status.CollectedCount == nil || *target.Status.CollectedCount != 3 { + t.Fatalf("collectedCount = %v, want 3", target.Status.CollectedCount) + } + if target.Status.CollectedCountUpdatedAt == nil { + t.Fatal("collectedCountUpdatedAt must be set on the first observation") + } +} diff --git a/internal/controller/kollecttarget_controller.go b/internal/controller/kollecttarget_controller.go index 86927615..393797b2 100644 --- a/internal/controller/kollecttarget_controller.go +++ b/internal/controller/kollecttarget_controller.go @@ -126,6 +126,13 @@ func (r *KollectTargetReconciler) Reconcile(ctx context.Context, req ctrl.Reques return ctrl.Result{}, nil } + // ORDERING INVARIANT: this resolve must stay ahead of RegisterTarget below, and + // must not be skipped when the filter status looks unchanged. It is the only + // thing on this path that refreshes the engine's namespace metadata cache, and + // passing EffectiveNamespaces makes the caller responsible for that freshness + // (see collect.RegisterTargetOptions). The cache backs resource-rule evaluation + // and the namespace watch opt-in/opt-out in ShouldCollect, so letting it go stale + // silently drops objects rather than failing. matched, effective, activeRules, ceiling := resolveTargetFilterStatus(ctx, r.Client, r.Engine, &target) updateTargetFilterStatus(&target, matched, effective, activeRules) @@ -241,10 +248,29 @@ func (r *KollectTargetReconciler) setDegraded( apimeta.RemoveStatusCondition(&target.Status.Conditions, conditionReady) apimeta.RemoveStatusCondition(&target.Status.Conditions, conditionSynced) setSinkReachableCondition(&target.Status.Conditions, target.Generation, false, reason, message) - return setTargetCondition( + _, err := setTargetCondition( ctx, r.Client, target, target.Generation, &target.Status.Conditions, conditionDegraded, metav1.ConditionTrue, reason, message, ) + + return err +} + +// syncCollectedCount records the freshly derived resource count on status, returning the +// stored value and whether it moved. The timestamp marks when the number last *changed*, +// so an operator can tell a live count from one frozen by a degraded or failing target — +// the whole point of PERF-FIX-05: the old prose-only count could not signal staleness. +func syncCollectedCount(target *kollectdevv1alpha1.KollectTarget, collected int) (int64, bool) { + next := int64(collected) + if target.Status.CollectedCount != nil && *target.Status.CollectedCount == next { + return next, false + } + + now := metav1.Now() + target.Status.CollectedCount = &next + target.Status.CollectedCountUpdatedAt = &now + + return next, true } func (r *KollectTargetReconciler) setReady( @@ -260,8 +286,19 @@ func (r *KollectTargetReconciler) setReady( target, target.Status.MatchedNamespaces, target.Status.EffectiveNamespaces, target.Status.ActiveResourceRules, ) + // status.collectedCount is the machine-readable source of truth and backs the + // COLLECTED printer column; the prose message is kept for backward compatibility and + // restates the stored number so the two can never disagree. + // + // Persisting the count must NOT depend on that message changing. An operator + // upgrading from a binary that already wrote "collecting 0 resource(s)" has a target + // whose Ready condition is byte-identical to the one this reconcile would write, so + // setTargetCondition skips the API call — and a measured zero would stay absent + // forever, indistinguishable from "never computed". countChanged below is what makes + // the write independent of the condition text (PERF-FIX-05 / F-05). + count, countChanged := syncCollectedCount(target, collected) msg := fmt.Sprintf("profileRef %q resolved; collecting %d resource(s)", - target.Spec.ProfileRef, collected) + target.Spec.ProfileRef, count) if sinkMsg == "" { sinkMsg = "namespace inventory sinks reachable" } @@ -273,15 +310,26 @@ func (r *KollectTargetReconciler) setReady( syncedReason, syncedMsg = scopeReason, scopeMsg } setSyncedCondition(&target.Status.Conditions, target.Generation, true, syncedReason, syncedMsg) - if err := setTargetCondition( + written, err := setTargetCondition( ctx, r.Client, target, target.Generation, &target.Status.Conditions, conditionReady, metav1.ConditionTrue, reasonCollecting, msg, - ); err != nil { + ) + if err != nil { return ctrl.Result{}, err } - return ctrl.Result{}, nil + // The condition write already carried the new number; only pay for a second call + // when it was skipped and the count still has to reach the API server. + if countChanged && !written { + if updateErr := r.Status().Update(ctx, target); updateErr != nil { + return ctrl.Result{}, updateErr + } + } + + // Objects entering or leaving the matched set do not enqueue their target, so a + // collecting target requeues itself to keep status.collectedCount live (F-05). + return ctrl.Result{RequeueAfter: r.Options.targetCountResync()}, nil } // SetupWithManager sets up the controller with the Manager. diff --git a/internal/controller/kollecttarget_controller_test.go b/internal/controller/kollecttarget_controller_test.go index d1aebef7..2cabdc97 100644 --- a/internal/controller/kollecttarget_controller_test.go +++ b/internal/controller/kollecttarget_controller_test.go @@ -96,7 +96,10 @@ var _ = Describe("KollectTarget Controller", func() { NamespacedName: typeNamespacedName, }) Expect(err).NotTo(HaveOccurred()) - Expect(result.RequeueAfter).To(BeZero()) + // PERF-FIX-05: a collecting target now requeues itself on the count resync + // interval. Before that it returned an empty Result, which is exactly why + // status.collectedCount could never refresh on its own. + Expect(result.RequeueAfter).To(Equal(DefaultTargetCountResync)) updated := &kollectdevv1alpha1.KollectTarget{} Expect(k8sClient.Get(reconcileCtx, typeNamespacedName, updated)).To(Succeed()) @@ -275,5 +278,179 @@ var _ = Describe("KollectTarget Controller", func() { Expect(ready).NotTo(BeNil()) Expect(ready.Status).To(Equal(metav1.ConditionTrue)) }) + + // Review finding F2 follow-up. RegisterTarget only refreshes the engine's namespace + // metadata cache on the recompute branch, and this reconciler always supplies + // EffectiveNamespaces — so the resolve at the top of Reconcile is the only thing + // keeping that cache fresh on this path. + // + // The discriminating assertion is on the resolved namespace set, not on the cache + // itself: if the resolve stops refreshing, it computes against an empty snapshot + // and a target in a freshly created namespace resolves an EMPTY effective set. + // (Asserting only that the cache ends up populated proves nothing — the engine's + // recompute branch refreshes it as a fallback precisely when the set comes in + // empty, so that assertion passes either way. Verified by mutation.) + It("resolves the target's own namespace against a freshly refreshed cache", func() { + reconcileCtx := context.Background() + profileName := "nsrefresh-profile-" + testNameSuffix() + targetName := "nsrefresh-target-" + testNameSuffix() + + Expect(engine.NamespaceMetaSnapshot()).NotTo(HaveKey(testNS), + "precondition: the engine must not already know this freshly created namespace") + + profile := &kollectdevv1alpha1.KollectProfile{ + ObjectMeta: metav1.ObjectMeta{Name: profileName, Namespace: testNS}, + Spec: kollectdevv1alpha1.KollectProfileSpec{ + TargetGVK: kollectdevv1alpha1.GroupVersionKind{Version: "v1", Kind: "ConfigMap"}, + }, + } + Expect(k8sClient.Create(reconcileCtx, profile)).To(Succeed()) + defer func() { _ = k8sClient.Delete(reconcileCtx, profile) }() + + target := &kollectdevv1alpha1.KollectTarget{ + ObjectMeta: metav1.ObjectMeta{Name: targetName, Namespace: testNS}, + Spec: kollectdevv1alpha1.KollectTargetSpec{ + ProfileRef: profileName, + CollectionFilterSpec: kollectdevv1alpha1.CollectionFilterSpec{ + IncludedNamespaces: []string{testNS}, + }, + }, + } + Expect(k8sClient.Create(reconcileCtx, target)).To(Succeed()) + defer func() { _ = k8sClient.Delete(reconcileCtx, target) }() + + reconciler := &KollectTargetReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Engine: engine, + } + targetKey := types.NamespacedName{Name: targetName, Namespace: testNS} + _, err := reconciler.Reconcile(reconcileCtx, reconcile.Request{NamespacedName: targetKey}) + Expect(err).NotTo(HaveOccurred()) + + updated := &kollectdevv1alpha1.KollectTarget{} + Expect(k8sClient.Get(reconcileCtx, targetKey, updated)).To(Succeed()) + Expect(updated.Status.EffectiveNamespaces).To(ContainElement(testNS), + "the resolve must see the target's own namespace, which only a refreshed cache contains") + + Expect(engine.NamespaceMetaSnapshot()).To(HaveKey(testNS)) + }) + + // PERF-FIX-05 / F-05: the reported resource count used to be a snapshot of the + // last spec reconcile. Objects entering or leaving the matched set never + // refreshed it, so status silently misreported scale (observed on the Talos lab: + // "collecting 1000 resource(s)" while 1200 were being collected). The count is + // now a numeric status field, re-derived on every reconcile, with the target + // requeued on a periodic resync so a live cluster converges without a spec edit. + It("tracks objects entering and leaving the matched set in status.collectedCount", func() { + reconcileCtx := context.Background() + profileName := "count-profile-" + testNameSuffix() + targetName := "count-target-" + testNameSuffix() + + profile := &kollectdevv1alpha1.KollectProfile{ + ObjectMeta: metav1.ObjectMeta{Name: profileName, Namespace: testNS}, + Spec: kollectdevv1alpha1.KollectProfileSpec{ + TargetGVK: kollectdevv1alpha1.GroupVersionKind{Version: "v1", Kind: "ConfigMap"}, + Attributes: []kollectdevv1alpha1.AttributeSpec{ + {Name: "name", Path: "{.metadata.name}"}, + }, + }, + } + Expect(k8sClient.Create(reconcileCtx, profile)).To(Succeed()) + defer func() { _ = k8sClient.Delete(reconcileCtx, profile) }() + + createConfigMaps := func(names ...string) { + for _, name := range names { + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNS}, + Data: map[string]string{"k": name}, + } + Expect(k8sClient.Create(reconcileCtx, cm)).To(Succeed()) + } + } + deleteConfigMaps := func(names ...string) { + for _, name := range names { + Expect(k8sClient.Delete(reconcileCtx, &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNS}, + })).To(Succeed()) + } + } + + const initial = 2 + createConfigMaps("count-a", "count-b") + + target := &kollectdevv1alpha1.KollectTarget{ + ObjectMeta: metav1.ObjectMeta{Name: targetName, Namespace: testNS}, + Spec: kollectdevv1alpha1.KollectTargetSpec{ + ProfileRef: profileName, + CollectionFilterSpec: kollectdevv1alpha1.CollectionFilterSpec{ + IncludedNamespaces: []string{testNS}, + }, + }, + } + Expect(k8sClient.Create(reconcileCtx, target)).To(Succeed()) + defer func() { _ = k8sClient.Delete(reconcileCtx, target) }() + + targetKey := types.NamespacedName{Name: targetName, Namespace: testNS} + reconciler := &KollectTargetReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Engine: engine, + } + + // reconcileWhenEngineSees waits for the engine to observe want items, then + // runs the reconcile the periodic resync would have run, and returns the + // persisted status. + reconcileWhenEngineSees := func(want int) *kollectdevv1alpha1.KollectTarget { + GinkgoHelper() + Eventually(func() int { + return engine.ItemCount(testNS, targetName) + }, 30*time.Second, 200*time.Millisecond).Should(Equal(want)) + + result, err := reconciler.Reconcile(reconcileCtx, reconcile.Request{NamespacedName: targetKey}) + Expect(err).NotTo(HaveOccurred()) + // The count can only stay live if the target is requeued: nothing else + // re-enqueues a target when a watched object enters or leaves its set. + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + + updated := &kollectdevv1alpha1.KollectTarget{} + Expect(k8sClient.Get(reconcileCtx, targetKey, updated)).To(Succeed()) + + return updated + } + + // Registration reconcile: informers start here, so the count is only + // meaningful from the following pass onwards. + _, err := reconciler.Reconcile(reconcileCtx, reconcile.Request{NamespacedName: targetKey}) + Expect(err).NotTo(HaveOccurred()) + + By("reporting the initial matched set") + updated := reconcileWhenEngineSees(initial) + Expect(updated.Status.CollectedCount).NotTo(BeNil()) + Expect(*updated.Status.CollectedCount).To(Equal(int64(initial))) + Expect(updated.Status.CollectedCountUpdatedAt).NotTo(BeNil()) + firstStamp := *updated.Status.CollectedCountUpdatedAt + + By("growing the count when more objects enter the selector") + createConfigMaps("count-c", "count-d", "count-e") + updated = reconcileWhenEngineSees(initial + 3) + Expect(updated.Status.CollectedCount).NotTo(BeNil()) + Expect(*updated.Status.CollectedCount).To(Equal(int64(initial + 3))) + Expect(updated.Status.CollectedCountUpdatedAt).NotTo(BeNil()) + Expect(updated.Status.CollectedCountUpdatedAt.Before(&firstStamp)).To(BeFalse()) + + By("shrinking the count when objects leave the selector") + deleteConfigMaps("count-c", "count-d", "count-e") + updated = reconcileWhenEngineSees(initial) + Expect(updated.Status.CollectedCount).NotTo(BeNil()) + Expect(*updated.Status.CollectedCount).To(Equal(int64(initial))) + + deleteConfigMaps("count-a", "count-b") + + By("reporting a genuine zero rather than an absent count") + updated = reconcileWhenEngineSees(0) + Expect(updated.Status.CollectedCount).NotTo(BeNil()) + Expect(*updated.Status.CollectedCount).To(BeZero()) + }) }) }) diff --git a/internal/controller/runtime_options.go b/internal/controller/runtime_options.go index ae1789b7..88602d59 100644 --- a/internal/controller/runtime_options.go +++ b/internal/controller/runtime_options.go @@ -11,6 +11,21 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" ) +// DefaultTargetCountResync bounds how stale KollectTarget.status.collectedCount may get. +// Nothing re-enqueues a target when a watched object enters or leaves its matched set, so +// without a periodic requeue the count is frozen at the last spec change (PERF-FIX-05 / +// F-05). +// +// The resync is not free. Each pass costs one live, cluster-wide, unpaginated namespace +// LIST — two when a KollectScope is enforced on the target's namespace, because the scope +// check resolves the filter status as well. The engine work itself is cheap +// (RegisterTarget skips the backfill on an unchanged fingerprint and no longer refreshes +// the namespace cache when the caller supplies EffectiveNamespaces), and the status write +// is skipped unless the number actually moved. Budget accordingly: at the default +// interval, N targets cost N namespace LISTs per minute. Raise the interval on large +// fleets; lower it for a more responsive count. +const DefaultTargetCountResync = 60 * time.Second + // RuntimeOptions configures controller parallelism and workqueue rate limiting. type RuntimeOptions struct { MaxConcurrentTarget int @@ -21,6 +36,9 @@ type RuntimeOptions struct { // failure rate limiter on each controller. When zero, controller-runtime defaults apply // (5ms base, 1000s max — see controller-runtime pkg/controller/controller.go). ReconcileRateLimitBase time.Duration + // TargetCountResync is how often a Ready KollectTarget is requeued to refresh + // status.collectedCount. Zero or negative selects DefaultTargetCountResync. + TargetCountResync time.Duration } // DefaultRuntimeOptions returns production-oriented defaults (ADR-0603). @@ -30,9 +48,20 @@ func DefaultRuntimeOptions() RuntimeOptions { MaxConcurrentInventory: 3, MaxConcurrentClusterTarget: 2, MaxConcurrentClusterInventory: 2, + TargetCountResync: DefaultTargetCountResync, } } +// targetCountResync returns the configured count resync interval, falling back to the +// default so a zero-value RuntimeOptions still keeps the count live. +func (o RuntimeOptions) targetCountResync() time.Duration { + if o.TargetCountResync > 0 { + return o.TargetCountResync + } + + return DefaultTargetCountResync +} + func (o RuntimeOptions) controllerOptions(maxConcurrent int) controller.Options { opts := controller.Options{ MaxConcurrentReconciles: maxConcurrent, diff --git a/internal/controller/runtime_options_test.go b/internal/controller/runtime_options_test.go index 2ca3516a..321ffc5a 100644 --- a/internal/controller/runtime_options_test.go +++ b/internal/controller/runtime_options_test.go @@ -15,6 +15,25 @@ func TestDefaultRuntimeOptions(t *testing.T) { if opts.MaxConcurrentTarget != 5 { t.Fatalf("defaults = %#v", opts) } + if opts.TargetCountResync != DefaultTargetCountResync { + t.Fatalf("TargetCountResync = %s, want %s", opts.TargetCountResync, DefaultTargetCountResync) + } +} + +// PERF-FIX-05: a zero-value RuntimeOptions must still requeue, otherwise any caller that +// forgets the field silently reintroduces the frozen-count defect. +func TestRuntimeOptionsTargetCountResync(t *testing.T) { + t.Parallel() + + if got := (RuntimeOptions{}).targetCountResync(); got != DefaultTargetCountResync { + t.Fatalf("zero-value resync = %s, want %s", got, DefaultTargetCountResync) + } + if got := (RuntimeOptions{TargetCountResync: -1}).targetCountResync(); got != DefaultTargetCountResync { + t.Fatalf("negative resync = %s, want %s", got, DefaultTargetCountResync) + } + if got := (RuntimeOptions{TargetCountResync: 5 * time.Second}).targetCountResync(); got != 5*time.Second { + t.Fatalf("explicit resync = %s, want 5s", got) + } } func TestRuntimeOptionsControllerOptionsRateLimiter(t *testing.T) { diff --git a/test/schema/printer_columns_test.go b/test/schema/printer_columns_test.go new file mode 100644 index 00000000..a7e8077e --- /dev/null +++ b/test/schema/printer_columns_test.go @@ -0,0 +1,75 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Konrad Heimel + +package schema + +import ( + "os" + "testing" + + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + "sigs.k8s.io/yaml" +) + +// PERF-FIX-05: an operator must be able to read collection scale straight off +// `kubectl get kollecttargets`. That only works if the shipped CRD declares the printer +// columns, and both the kubebuilder CRDs and the Helm chart copy must carry them — +// nothing else in the suite would notice if a regeneration dropped them. +func TestKollectTargetPrinterColumns(t *testing.T) { + t.Parallel() + + root := repoRoot(t) + // Age is load-bearing, not decoration: declaring any additionalPrinterColumns + // suppresses the apiserver's default AGE column, so dropping this entry in a + // regeneration would silently remove AGE from `kubectl get kollecttargets`. + want := map[string]string{ + "Collected": ".status.collectedCount", + "Updated": ".status.collectedCountUpdatedAt", + "Age": ".metadata.creationTimestamp", + } + + paths := map[string]string{ + "config/crd/bases": CRDPath(root, "kollect.dev_kollecttargets.yaml"), + "charts/kollect/crds": root + + "/charts/kollect/crds/kollect.dev_kollecttargets.yaml", + } + + for source, path := range paths { + t.Run(source, func(t *testing.T) { + t.Parallel() + + got := printerColumns(t, path) + for name, jsonPath := range want { + if got[name] != jsonPath { + t.Fatalf("printer column %q = %q, want %q (columns: %v)", name, got[name], jsonPath, got) + } + } + }) + } +} + +// printerColumns returns every additionalPrinterColumn declared by a CRD manifest, keyed +// by column name. +func printerColumns(t *testing.T, path string) map[string]string { + t.Helper() + + //nolint:gosec // G304: path is a committed manifest in this repository. + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read crd: %v", err) + } + + var crd apiextensionsv1.CustomResourceDefinition + if unmarshalErr := yaml.Unmarshal(raw, &crd); unmarshalErr != nil { + t.Fatalf("parse crd: %v", unmarshalErr) + } + + cols := map[string]string{} + for i := range crd.Spec.Versions { + for _, col := range crd.Spec.Versions[i].AdditionalPrinterColumns { + cols[col.Name] = col.JSONPath + } + } + + return cols +}