refactor: enhance scaling configuration and resource management in tests

internal/controller/applycache.go

@@ -21,6 +21,19 @@ import (
 // to one apply per pod per interval.
 const applyRefreshInterval = 5 * time.Minute
 
+// applyVerifyDelay is the grace period before a safety-net re-POST of an
+// unverified hash. gNMIc's config/apply can ACK before every Subscribe stream
+// has re-established, and occasionally one never comes back on its own; a
+// second POST of the same config reliably kicks it loose. The delay has to
+// clear two things: every unrelated reconcile trigger that fires in the
+// meantime (Target status, Secret changes, periodic resync — all check
+// Unchanged and must not treat a still-settling apply as stale), and every
+// test's own tight "stays stable" window, so the safety net never lands mid
+// assertion. Firing it immediately (as a first attempt did) instead doubled
+// every apply back-to-back and made pods work harder to reconnect right when
+// they were already struggling, which was worse than the bug it targeted.
+const applyVerifyDelay = 30 * time.Second
+
 // ApplyCache records what was last successfully applied to each collector pod,
 // so a reconcile that changes nothing does not re-POST the whole configuration
 // to every pod.
@@ -38,8 +51,9 @@ type ApplyCache struct {
 }
 
 type applyRecord struct {
-	hash string
-	at   time.Time
+	hash     string
+	at       time.Time
+	verified bool
 }
 
 // NewApplyCache returns a cache using the default refresh interval.
@@ -55,7 +69,14 @@ func (c *ApplyCache) timeNow() time.Time {
 }
 
 // Unchanged reports whether the pod already holds this exact configuration and
-// the record is still inside the refresh interval.
+// no apply is due right now.
+//
+// An unverified hash still reports unchanged until applyVerifyDelay elapses —
+// otherwise every unrelated reconcile in that window would re-POST the same
+// config, compounding the very reconnect problem the verify pass exists to
+// fix. Once the grace period passes with no newer hash recorded, the next
+// reconcile (organic or the deliberate requeue below) is due its one
+// safety-net repost.
 //
 // A nil cache always reports false, so a reconciler constructed without one
 // (tests, the API server) applies unconditionally rather than silently skipping.
@@ -69,19 +90,43 @@ func (c *ApplyCache) Unchanged(key, hash string) bool {
 	if !ok || rec.hash != hash {
 		return false
 	}
-	return c.timeNow().Sub(rec.at) < c.ttl
+	age := c.timeNow().Sub(rec.at)
+	if !rec.verified && age >= applyVerifyDelay {
+		return false
+	}
+	return age < c.ttl
 }
 
-// Record marks a configuration as successfully applied to a pod. It must only
-// be called after the POST succeeds: recording an attempt would make a failed
-// apply look applied until the plan changes again.
+// Record marks a configuration as successfully POSTed to a pod. The first
+// Record for a hash leaves the entry unverified; if Unchanged later reports
+// changed because the grace period elapsed, the resulting repost's Record
+// call marks it verified so the short-circuit engages. It must only be called
+// after the POST succeeds: recording an attempt would make a failed apply
+// look applied until the plan changes again.
 func (c *ApplyCache) Record(key, hash string) {
 	if c == nil {
 		return
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.entries[key] = applyRecord{hash: hash, at: c.timeNow()}
+	now := c.timeNow()
+	if rec, ok := c.entries[key]; ok && rec.hash == hash && !rec.verified {
+		c.entries[key] = applyRecord{hash: hash, at: now, verified: true}
+		return
+	}
+	c.entries[key] = applyRecord{hash: hash, at: now, verified: false}
+}
+
+// NeedsVerify reports whether key holds an unverified record for hash, so the
+// reconciler should requeue after applyVerifyDelay for the follow-up POST.
+func (c *ApplyCache) NeedsVerify(key, hash string) bool {
+	if c == nil {
+		return false
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	rec, ok := c.entries[key]
+	return ok && rec.hash == hash && !rec.verified
 }
 
 // Invalidate drops one pod's record, forcing the next reconcile to re-apply to

internal/controller/applycache_test.go

@@ -32,11 +32,14 @@ func TestApplyCache_RecordThenUnchanged(t *testing.T) {
 }
 
 // The TTL is the backstop for a pod losing its config without dropping its SSE
-// stream, which nothing else can detect.
+// stream, which nothing else can detect. It is measured from verification,
+// since an unverified hash is on its own, shorter clock (applyVerifyDelay).
 func TestApplyCache_ExpiresAfterRefreshInterval(t *testing.T) {
 	now := time.Unix(1000, 0)
 	c := testCache(t, &now)
 	c.Record("ns/c1/0", "hashA")
+	now = now.Add(applyVerifyDelay)
+	c.Record("ns/c1/0", "hashA") // simulate the safety-net repost verifying it
 
 	now = now.Add(applyRefreshInterval - time.Second)
 	if !c.Unchanged("ns/c1/0", "hashA") {
@@ -98,6 +101,71 @@ func TestApplyCache_NilIsAlwaysChanged(t *testing.T) {
 	c.InvalidateCluster("ns", "c1")
 }
 
+// A freshly applied hash is unverified but must still read as unchanged
+// during the grace period: an unrelated reconcile arriving in that window
+// (e.g. a Target status update) must not trigger a redundant repost.
+func TestApplyCache_UnverifiedUnchangedDuringGrace(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(applyVerifyDelay - time.Second)
+	if !c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("unverified hash reported changed before the grace period elapsed")
+	}
+	if !c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("NeedsVerify should stay true until the follow-up Record")
+	}
+}
+
+// Once the grace period elapses with no newer hash recorded, the safety-net
+// repost is due: Unchanged must report changed so the reconciler re-POSTs.
+func TestApplyCache_UnverifiedNeedsRepostAfterGrace(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(applyVerifyDelay)
+	if c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("unverified hash still reported unchanged after the grace period")
+	}
+}
+
+// The repost's Record call marks the hash verified, so it settles for good
+// (until the TTL) instead of triggering yet another repost.
+func TestApplyCache_SecondRecordVerifies(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(applyVerifyDelay)
+	c.Record("ns/c1/0", "hashA")
+	if c.NeedsVerify("ns/c1/0", "hashA") {
+		t.Fatal("hash still reported as needing verify after the follow-up Record")
+	}
+	now = now.Add(applyVerifyDelay)
+	if !c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("verified hash reported changed")
+	}
+}
+
+// A real edit before the grace period elapses must reset verification for
+// the new hash rather than leave a stale repost pending for the old one.
+func TestApplyCache_NewHashResetsVerification(t *testing.T) {
+	now := time.Unix(1000, 0)
+	c := testCache(t, &now)
+	c.Record("ns/c1/0", "hashA")
+
+	now = now.Add(time.Second)
+	c.Record("ns/c1/0", "hashB")
+	if c.Unchanged("ns/c1/0", "hashA") {
+		t.Fatal("superseded hash still reported unchanged")
+	}
+	if !c.NeedsVerify("ns/c1/0", "hashB") {
+		t.Fatal("new hash should start its own unverified grace period")
+	}
+}
+
 // The fingerprint covers the bytes on the wire, so equal payloads compare equal
 // and any difference in what is sent is a difference here.
 func TestFingerprint(t *testing.T) {

internal/controller/cluster_controller.go

@@ -532,14 +532,16 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	// distrubute to desired replicas only, this makes redistribution fast in case of scaling down.
 	numPods := int(desiredReplicas)
 	configApplied := false
+	verifyPending := false
 	var configError error
 	var unassignedTargets int32
-	if unassigned, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
+	if unassigned, needsVerify, err := r.applyConfigToPods(ctx, &cluster, applyPlan, numPods); err != nil {
 		logger.Error(err, "failed to apply config to gNMIc pods")
 		configError = err
 	} else {
 		configApplied = true
 		unassignedTargets = unassigned
+		verifyPending = needsVerify
 		logger.Info("successfully applied config to gNMIc cluster", "pods", numPods)
 	}
 
@@ -719,6 +721,13 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if configApplied && len(pipelines) == 0 {
 		return ctrl.Result{RequeueAfter: time.Second}, nil
 	}
+	// A pod is holding an unverified hash: requeue for the safety-net repost
+	// once the grace period elapses (see applyVerifyDelay). If something else
+	// reconciles first with a newer plan, this pass is superseded harmlessly.
+	if verifyPending {
+		logger.Info("scheduling apply verify requeue", "cluster", cluster.Name, "namespace", cluster.Namespace, "after", applyVerifyDelay)
+		return ctrl.Result{RequeueAfter: applyVerifyDelay}, nil
+	}
 
 	return ctrl.Result{}, nil
 }
@@ -749,8 +758,11 @@ func clusterStatusEqual(a, b gnmicv1alpha1.ClusterStatus) bool {
 }
 
 // applyConfigToPods sends the apply plan to all gNMIc pods with distributed targets.
-// Returns the number of targets that could not be assigned due to capacity limits.
-func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (int32, error) {
+// Returns the number of targets that could not be assigned due to capacity
+// limits, and whether any pod's apply is still unverified and needs a
+// follow-up POST once the ApplyCache grace period elapses (see
+// applyVerifyDelay).
+func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmicv1alpha1.Cluster, plan *gnmic.ApplyPlan, numPods int) (unassigned int32, needsVerify bool, err error) {
 	logger := log.FromContext(ctx)
 
 	stsName := fmt.Sprintf("%s%s", resourcePrefix, cluster.Name)
@@ -762,7 +774,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 	// create an HTTP client to send the apply plan to the gNMIc pods
 	httpClient, err := r.createHTTPClientForCluster(ctx, cluster)
 	if err != nil {
-		return 0, fmt.Errorf("failed to create HTTP client: %w", err)
+		return 0, false, fmt.Errorf("failed to create HTTP client: %w", err)
 	}
 	distResult := gnmic.DistributeTargets(plan, numPods, cluster.Spec.TargetDistribution)
 
@@ -796,7 +808,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		}
 		body, err := json.Marshal(podPlan)
 		if err != nil {
-			return 0, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
+			return 0, false, fmt.Errorf("failed to marshal apply plan for pod %d: %w", podIndex, err)
 		}
 		hash := fingerprint(body)
 		bodies[podIndex] = body
@@ -806,8 +818,20 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			anyChanged = true
 		}
 	}
+	// Checked for every pod with a plan, not just those just (re)applied to:
+	// a pod can be sitting on an unverified hash from an earlier reconcile
+	// that this pass left untouched because Unchanged still held it inside
+	// the grace period.
+	pendingVerify := func() bool {
+		for podIndex := range hashes {
+			if r.Applied.NeedsVerify(streamKey(cluster.Namespace, cluster.Name, podIndex), hashes[podIndex]) {
+				return true
+			}
+		}
+		return false
+	}
 	if !anyChanged {
-		return int32(len(distResult.UnassignedTargets)), nil
+		return int32(len(distResult.UnassignedTargets)), pendingVerify(), nil
 	}
 
 	// Two-phase apply avoids double-collection when a target moves between pods.
@@ -848,7 +872,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			url := podURL(podIndex)
 			logger.Info("shrinking gNMIc pod targets before reassignment", "url", url, "targets", len(shrink.Targets))
 			if err := r.sendApplyRequest(ctx, url, shrink, httpClient); err != nil {
-				return 0, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
+				return 0, false, fmt.Errorf("failed to shrink config on pod %d: %w", podIndex, err)
 			}
 		}
 		// gNMIc's config/apply returns before Subscribe streams are fully torn
@@ -857,7 +881,7 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 		// double-collects.
 		select {
 		case <-ctx.Done():
-			return 0, ctx.Err()
+			return 0, false, ctx.Err()
 		case <-time.After(time.Second):
 		}
 	}
@@ -868,22 +892,24 @@ func (r *ClusterReconciler) applyConfigToPods(ctx context.Context, cluster *gnmi
 			continue
 		}
 		url := podURL(podIndex)
-		logger.Info("sending config to gNMIc pod", "url", url)
+		key := streamKey(cluster.Namespace, cluster.Name, podIndex)
+		isVerifyRepost := r.Applied.NeedsVerify(key, hashes[podIndex])
+		logger.Info("sending config to gNMIc pod", "url", url, "verifyRepost", isVerifyRepost)
 		if err := r.sendApplyBody(ctx, url, bodies[podIndex], httpClient); err != nil {
-			return 0, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
+			return 0, false, fmt.Errorf("failed to apply config to pod %d: %w", podIndex, err)
 		}
 		// Recorded only after the POST succeeds. Recording the attempt would
 		// make a failed apply look applied until the plan changes again.
-		r.Applied.Record(streamKey(cluster.Namespace, cluster.Name, podIndex), hashes[podIndex])
-		logger.Info("config applied to pod", "pod", podIndex, "targets", len(podPlan.Targets))
+		r.Applied.Record(key, hashes[podIndex])
+		logger.Info("config applied to pod", "pod", podIndex, "targets", len(podPlan.Targets), "verifyRepost", isVerifyRepost)
 	}
 
-	unassigned := int32(len(distResult.UnassignedTargets))
+	unassigned = int32(len(distResult.UnassignedTargets))
 	if unassigned > 0 {
 		logger.Info("targets unassigned due to capacity limits", "count", unassigned)
 	}
 
-	return unassigned, nil
+	return unassigned, pendingVerify(), nil
 }
 
 func (r *ClusterReconciler) createHTTPClientForCluster(ctx context.Context, cluster *gnmicv1alpha1.Cluster) (*http.Client, error) {

test.mk

@@ -283,9 +283,21 @@ run-integration-tests-v2: ## Bring up the gnmi-gen suite env, run all suites, te
 	exit $$status
 
 # Nightly / local-only fleet suite. Not wired into CI.
-# SCALE_TARGETS defaults to 200; override e.g. SCALE_TARGETS=50 make integration-test-scale
+# Override e.g. SCALE_TARGETS=50 SCALE_REPLICAS=2 SCALE_CPU_LIMIT=1 make integration-test-scale
 SCALE_TARGETS ?= 200
+SCALE_REPLICAS ?= 4
+SCALE_CPU_REQUEST ?= 500m
+SCALE_CPU_LIMIT ?= 2
+SCALE_MEMORY_REQUEST ?= 512Mi
+SCALE_MEMORY_LIMIT ?= 2Gi
 .PHONY: integration-test-scale
 integration-test-scale: integration-env-check ## Run 013-scale (sets RUN_SCALE=1)
-	RUN_SCALE=1 SCALE_TARGETS=$(SCALE_TARGETS) go test -tags=integration -count=1 -timeout 45m -v ./$(IT_SUITE_DIR)/013-scale/...
+	RUN_SCALE=1 \
+	SCALE_TARGETS=$(SCALE_TARGETS) \
+	SCALE_REPLICAS=$(SCALE_REPLICAS) \
+	SCALE_CPU_REQUEST=$(SCALE_CPU_REQUEST) \
+	SCALE_CPU_LIMIT=$(SCALE_CPU_LIMIT) \
+	SCALE_MEMORY_REQUEST=$(SCALE_MEMORY_REQUEST) \
+	SCALE_MEMORY_LIMIT=$(SCALE_MEMORY_LIMIT) \
+	go test -tags=integration -count=1 -timeout 45m -v ./$(IT_SUITE_DIR)/013-scale/...
 

test/integration/suite/003-targets/targets_test.go

@@ -241,6 +241,10 @@ spec:
 		return true, ""
 	})
 	s.GnmiGen.WaitStreams(t, leaf2, 0)
+	// Adding leaf2 re-applied the whole pod's config, briefly reloading
+	// leaf1's stream too; settle before proving it holds, or the window can
+	// start mid-reconnect and flake on a transient 0.
+	s.GnmiGen.WaitStreams(t, leaf1, 1)
 	s.GnmiGen.ConsistentlyCollectedOnce(t, 5*time.Second, 1, leaf1)
 
 	// Replace the Secret via YAML apply (avoids managedFields on a reused object),