diff --git a/internal/controller/applycache.go b/internal/controller/applycache.go index 6f5f861..319868a 100644 --- a/internal/controller/applycache.go +++ b/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 diff --git a/internal/controller/applycache_test.go b/internal/controller/applycache_test.go index da4ff4a..d7420b9 100644 --- a/internal/controller/applycache_test.go +++ b/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) { diff --git a/internal/controller/cluster_controller.go b/internal/controller/cluster_controller.go index 26de113..e74ca1f 100644 --- a/internal/controller/cluster_controller.go +++ b/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) { diff --git a/test.mk b/test.mk index b4a5c47..6f7b1bb 100644 --- a/test.mk +++ b/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/... diff --git a/test/integration/suite/003-targets/targets_test.go b/test/integration/suite/003-targets/targets_test.go index 363dfba..31e4d7d 100644 --- a/test/integration/suite/003-targets/targets_test.go +++ b/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), diff --git a/test/integration/suite/004-subscriptions/subscriptions_test.go b/test/integration/suite/004-subscriptions/subscriptions_test.go index ece79fc..f0c7895 100644 --- a/test/integration/suite/004-subscriptions/subscriptions_test.go +++ b/test/integration/suite/004-subscriptions/subscriptions_test.go @@ -103,7 +103,6 @@ func waitIdle(t *testing.T) { func applySubscription(t *testing.T, name string, vars map[string]any) { t.Helper() - setPipelineEnabled(t, true) base := map[string]any{ "Name": name, "Mode": "STREAM/SAMPLE", @@ -122,9 +121,13 @@ func applySubscription(t *testing.T, name string, vars map[string]any) { if err != nil { t.Fatalf("reading subscription fixture: %v", err) } + // Create the Subscription before re-enabling the pipeline. Enabling first + // can race an apply with targets but no subscriptions, which gNMIc rejects + // and can leave the collector mid-reconnect with paths=[]. if _, err := s.K8s.ApplyYAMLNoCleanup(string(b), base); err != nil { t.Fatalf("applying subscription %s: %v", name, err) } + setPipelineEnabled(t, true) waitClusterReady(t) } @@ -155,9 +158,16 @@ func deleteSubscription(t *testing.T, name string) { func waitStreams(t *testing.T, want int) { t.Helper() - for _, name := range allTargets { - s.GnmiGen.WaitStreams(t, name, want) - } + // One shared Medium budget across the fleet. Sequential per-target waits + // burn the full timeout on the first lagging target and hide siblings. + harness.Wait(t, harness.Medium, fmt.Sprintf("%d stream(s) on all targets", want), func() (bool, string) { + for _, name := range allTargets { + if n := s.GnmiGen.StreamCount(name); n != want { + return false, fmt.Sprintf("%s streams=%d", name, n) + } + } + return true, "" + }) } func waitPathOnAll(t *testing.T, path string) { @@ -167,6 +177,75 @@ func waitPathOnAll(t *testing.T, path string) { } } +// waitPathsOnAll waits until every target carries all want paths on the same +// stream snapshot. Split waitPathOnAll calls race stream reconnects: the first +// path can still be visible on the old stream, then paths go empty before the +// next assertion. +func waitPathsOnAll(t *testing.T, want ...string) { + t.Helper() + harness.Wait(t, harness.Medium, fmt.Sprintf("paths %v on all targets", want), func() (bool, string) { + for _, name := range allTargets { + paths := s.GnmiGen.Paths(name) + for _, w := range want { + found := false + for _, p := range paths { + if p == w { + found = true + break + } + } + if !found { + return false, fmt.Sprintf("%s paths=%v", name, paths) + } + } + } + return true, "" + }) +} + +// waitPathReplacedOnAll waits until every target shows have and not gone on the +// same snapshot. Sequential present-then-absent waits can pass the first check +// on a transitional stream (or see paths=[] mid-reconnect) and then burn the +// full Medium timeout on the second. +func waitPathReplacedOnAll(t *testing.T, have, gone string) { + t.Helper() + waitExactPathsOnAll(t, []string{have}, []string{gone}) +} + +// waitExactPathsOnAll waits until every target has all want paths, none of the +// gone paths, and exactly one stream — in a single snapshot per poll. +func waitExactPathsOnAll(t *testing.T, want, gone []string) { + t.Helper() + harness.Wait(t, harness.Medium, fmt.Sprintf("paths want=%v gone=%v on all targets", want, gone), func() (bool, string) { + for _, name := range allTargets { + if n := s.GnmiGen.StreamCount(name); n != 1 { + return false, fmt.Sprintf("%s streams=%d paths=%v", name, n, s.GnmiGen.Paths(name)) + } + paths := s.GnmiGen.Paths(name) + for _, w := range want { + found := false + for _, p := range paths { + if p == w { + found = true + break + } + } + if !found { + return false, fmt.Sprintf("%s paths=%v", name, paths) + } + } + for _, g := range gone { + for _, p := range paths { + if p == g { + return false, fmt.Sprintf("%s still has %s: %v", name, g, paths) + } + } + } + } + return true, "" + }) +} + func waitPathAbsentOnAll(t *testing.T, path string) { t.Helper() for _, name := range allTargets { @@ -245,6 +324,25 @@ func waitUpdatesOnly(t *testing.T, target string, want bool) { }) } +func waitUpdatesOnlyOnAll(t *testing.T, want bool) { + t.Helper() + harness.Wait(t, harness.Medium, fmt.Sprintf("updates_only=%v on all targets", want), func() (bool, string) { + for _, name := range allTargets { + subs, err := s.GnmiGen.Subscriptions(name) + if err != nil { + return false, err.Error() + } + if len(subs) != 1 { + return false, fmt.Sprintf("%s streams=%d", name, len(subs)) + } + if subs[0].UpdatesOnly != want { + return false, fmt.Sprintf("%s updates_only=%v", name, subs[0].UpdatesOnly) + } + } + return true, "" + }) +} + func waitSuppressAndHeartbeat(t *testing.T, target string, suppress bool, heartbeat string) { t.Helper() harness.Wait(t, harness.Medium, fmt.Sprintf("suppress=%v heartbeat=%s on %s", suppress, heartbeat, target), func() (bool, string) { @@ -305,9 +403,9 @@ func TestSub001_PathChangeReplacesCollectedPaths(t *testing.T) { restartsBefore := s.K8s.RestartCounts(t, cluster) patchSubscription(t, "s1", map[string]any{"paths": []string{pathCPU}}) + harness.WaitConfigApplied(t, s.K8s, cluster) - waitPathOnAll(t, pathCPU) - waitPathAbsentOnAll(t, pathIF) + waitPathReplacedOnAll(t, pathCPU, pathIF) waitStreams(t, 1) harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster)) @@ -325,12 +423,7 @@ func TestSub002_AddingPathExtendsSubscription(t *testing.T) { restartsBefore := s.K8s.RestartCounts(t, cluster) patchSubscription(t, "s1", map[string]any{"paths": []string{pathIF, pathCPU}}) - waitPathOnAll(t, pathIF) - waitPathOnAll(t, pathCPU) - harness.Wait(t, harness.Medium, "resolved path count is 2", func() (bool, string) { - n := len(s.GnmiGen.Paths(leaf1)) - return n >= 2, fmt.Sprintf("paths=%v", s.GnmiGen.Paths(leaf1)) - }) + waitPathsOnAll(t, pathIF, pathCPU) // Path edits re-establish the Subscribe stream, which resets // notifications_sent. Prove data still flows on the new stream rather // than comparing absolute counters across the teardown. @@ -386,22 +479,17 @@ func TestSub005_UpdatesOnlyReachesWire(t *testing.T) { "UpdatesOnly": false, }) waitStreams(t, 1) - waitUpdatesOnly(t, leaf1, false) + waitUpdatesOnlyOnAll(t, false) syncBefore := firstStream(t, leaf1).SyncMessagesSent restartsBefore := s.K8s.RestartCounts(t, cluster) patchSubscription(t, "s1", map[string]any{"updatesOnly": true}) - waitUpdatesOnly(t, leaf1, true) - - // updatesOnly re-establishes the stream; sync pattern should differ from - // the initial full sync (typically sync_messages_sent stays 0 or much lower). - harness.Wait(t, harness.Medium, "updates_only stream settled", func() (bool, string) { - sub := firstStream(t, leaf1) - if !sub.UpdatesOnly { - return false, "updates_only still false" - } - return true, "" - }) + harness.WaitConfigApplied(t, s.K8s, cluster) + // Assert the flag on every target in one snapshot. Waiting only on leaf1 + // then waitStreams(leaf2) races a mid-reconnect gap where leaf2 has + // streams=0 after the apply ACK (ApplyCache verify re-POST recovers it). + waitUpdatesOnlyOnAll(t, true) + syncAfter := firstStream(t, leaf1).SyncMessagesSent if syncBefore == 0 && syncAfter == 0 { t.Log("sync_messages_sent was 0 both before and after; flag still asserted on wire") @@ -409,7 +497,6 @@ func TestSub005_UpdatesOnlyReachesWire(t *testing.T) { // Soft signal only — gnmi-gen may still emit a sync response. t.Logf("sync_messages_sent before=%d after=%d", syncBefore, syncAfter) } - waitStreams(t, 1) harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster)) } @@ -474,8 +561,7 @@ func TestSub009_DeletingSubscriptionStopsPaths(t *testing.T) { applySubscription(t, "if", map[string]any{"Paths": []string{pathIF}}) applySubscription(t, "cpu", map[string]any{"Paths": []string{pathCPU}}) waitStreams(t, 2) - waitPathOnAll(t, pathIF) - waitPathOnAll(t, pathCPU) + waitPathsOnAll(t, pathIF, pathCPU) harness.WaitClusterCounts(t, s.K8s, cluster, harness.ClusterCounts{ SubscriptionsCount: harness.I32(2), }) @@ -518,11 +604,11 @@ func TestSub010_RapidSuccessiveEditsConverge(t *testing.T) { for _, p := range paths { patchSubscription(t, "s1", map[string]any{"paths": p}) } + harness.WaitConfigApplied(t, s.K8s, cluster) - waitPathOnAll(t, pathIF) - waitPathAbsentOnAll(t, pathCPU) - waitPathAbsentOnAll(t, pathSys) - waitStreams(t, 1) + // Final state in one snapshot: pathIF present, prior paths gone, one stream. + // Split present/absent waits race reconnect gaps after the patch burst. + waitExactPathsOnAll(t, []string{pathIF}, []string{pathCPU, pathSys}) s.GnmiGen.ConsistentlyCollectedOnce(t, 15*time.Second, 1, allTargets...) harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster)) harness.AssertNoPanics(t, s.K8s.OperatorLogs(t, time.Since(since)+time.Minute)) diff --git a/test/integration/suite/005-outputs/outputs_test.go b/test/integration/suite/005-outputs/outputs_test.go index bf7089d..99b532d 100644 --- a/test/integration/suite/005-outputs/outputs_test.go +++ b/test/integration/suite/005-outputs/outputs_test.go @@ -223,7 +223,9 @@ func TestOut003_ServiceTypeHonored(t *testing.T) { applyPipeline(t, "p1", []string{"out1"}, "") waitClusterReady(t) - svc := s.K8s.Service(t, harness.PromServiceName(cluster, "p1", "out1")) + // Cluster Ready can precede the per-pipeline Service reconcile; wait for + // the Service itself rather than racing it. + svc := waitPromService(t, "p1", "out1") if svc.Spec.Type != corev1.ServiceTypeNodePort { t.Fatalf("type=%s want NodePort", svc.Spec.Type) } @@ -275,6 +277,12 @@ func TestOut005_EditingOutputNoRestart(t *testing.T) { waitIdle(t) const oldFile = "/tmp/005-old.jsonl" const newFile = "/tmp/005-new.jsonl" + // Prior runs leave these files in the pod; clear while no file output is + // mounted so "non-empty" cannot pass on stale content. + pod := s.K8s.FirstCollectorPod(t, cluster) + _ = s.K8s.Exec(t, pod, harness.CollectorContainer, "sh", "-c", + fmt.Sprintf("rm -f %q %q", oldFile, newFile)) + applyFileOutput(t, "file1", oldFile) applyPromOutput(t, "out1", map[string]any{"Expiration": "60s"}) applyPipeline(t, "p1", []string{"file1", "out1"}, "") @@ -283,7 +291,6 @@ func TestOut005_EditingOutputNoRestart(t *testing.T) { s.K8s.WaitClusterPrometheusSources(t, cluster, "p1", "out1", allTargets, harness.Medium) restartsBefore := s.K8s.RestartCounts(t, cluster) - oldLen := len(s.K8s.ReadCollectorFile(t, cluster, oldFile)) out := &gnmicv1alpha1.Output{} s.K8s.WaitExists(t, "file1", out) @@ -294,14 +301,22 @@ func TestOut005_EditingOutputNoRestart(t *testing.T) { s.K8s.Patch(t, prom, `{"spec":{"config":{"expiration":"120s"}}}`) harness.WaitConfigApplied(t, s.K8s, cluster) - s.K8s.WaitCollectorFileNonEmpty(t, cluster, newFile) + newBody := s.K8s.WaitCollectorFileNonEmpty(t, cluster, newFile) s.K8s.WaitClusterPrometheusSources(t, cluster, "p1", "out1", allTargets, harness.Medium) - // Old file should stop growing once the output points elsewhere. - time.Sleep(3 * time.Second) - if got := len(s.K8s.ReadCollectorFile(t, cluster, oldFile)); got > oldLen+64 { - t.Errorf("old file still growing: was %d now %d", oldLen, got) - } + // Mark after the switch: sampling before the patch counted apply-window + // writes as "growth" and flaked. Prove live traffic hits the new file, + // then that the old one stays put. + newFloor := len(newBody) + s.K8s.WaitCollectorFileGrows(t, cluster, newFile, newFloor) + oldLen := len(s.K8s.ReadCollectorFile(t, cluster, oldFile)) + harness.Consistently(t, 5*time.Second, time.Second, "old file stopped growing", func() (bool, string) { + got := len(s.K8s.ReadCollectorFile(t, cluster, oldFile)) + if got > oldLen+64 { + return false, fmt.Sprintf("was %d now %d", oldLen, got) + } + return true, "" + }) harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster)) } @@ -323,6 +338,10 @@ func TestOut006_DeletingOutputRemovesService(t *testing.T) { harness.WaitClusterCounts(t, s.K8s, cluster, harness.ClusterCounts{ OutputsCount: harness.I32(0), }) + // The delete's config/apply briefly reloads every stream on the pod; + // settle to 1-per-target before proving it holds, or the window can start + // mid-reconnect and flake on a transient 0. + s.GnmiGen.WaitFleetStreams(t, harness.Medium, 1, allTargets) s.GnmiGen.ConsistentlyCollectedOnce(t, 10*time.Second, 1, allTargets...) harness.AssertNoRestarts(t, restartsBefore, s.K8s.RestartCounts(t, cluster)) } diff --git a/test/integration/suite/013-scale/fixtures/baseline.yaml b/test/integration/suite/013-scale/fixtures/baseline.yaml index bceb3b3..70a7e4f 100644 --- a/test/integration/suite/013-scale/fixtures/baseline.yaml +++ b/test/integration/suite/013-scale/fixtures/baseline.yaml @@ -1,5 +1,5 @@ # Shared props for 013-scale. Target CRs are generated in AfterBaseline. -# Cluster replicas come from SCALE_REPLICAS via BaselineVars. +# Cluster replicas and collector resources come from SCALE_* via BaselineVars. apiVersion: v1 kind: Secret metadata: @@ -30,11 +30,11 @@ spec: restPort: 7890 resources: requests: - cpu: 500m - memory: 512Mi + cpu: "{{ .CPURequest }}" + memory: "{{ .MemoryRequest }}" limits: - cpu: "2" - memory: 2Gi + cpu: "{{ .CPULimit }}" + memory: "{{ .MemoryLimit }}" --- apiVersion: operator.gnmic.dev/v1alpha1 kind: Subscription diff --git a/test/integration/suite/013-scale/scale_test.go b/test/integration/suite/013-scale/scale_test.go index 20a6dd6..29d074f 100644 --- a/test/integration/suite/013-scale/scale_test.go +++ b/test/integration/suite/013-scale/scale_test.go @@ -4,8 +4,14 @@ // single-target churn cost, sustained membership change, and mass reboot. // // Gated behind RUN_SCALE=1 (see TestMain). Not part of the default CI lane. -// Fleet size defaults to 200 (SCALE_TARGETS) and collector pods to 4 -// (SCALE_REPLICAS). +// Tunables (env vars, with defaults): +// +// SCALE_TARGETS=200 +// SCALE_REPLICAS=4 +// SCALE_CPU_REQUEST=500m +// SCALE_CPU_LIMIT=2 +// SCALE_MEMORY_REQUEST=512Mi +// SCALE_MEMORY_LIMIT=2Gi package scale import ( @@ -37,6 +43,10 @@ var ( targets []string fleetN int replicas int + cpuReq string + cpuLim string + memReq string + memLim string spareSim string sparePort int minPerPod int @@ -50,6 +60,10 @@ func TestMain(m *testing.M) { } fleetN = envInt("SCALE_TARGETS", 200) replicas = envInt("SCALE_REPLICAS", 4) + cpuReq = envStr("SCALE_CPU_REQUEST", "500m") + cpuLim = envStr("SCALE_CPU_LIMIT", "2") + memReq = envStr("SCALE_MEMORY_REQUEST", "512Mi") + memLim = envStr("SCALE_MEMORY_LIMIT", "2Gi") if replicas < 1 { fmt.Fprintf(os.Stderr, "013-scale: SCALE_REPLICAS=%d too small (min 1)\n", replicas) os.Exit(1) @@ -65,8 +79,8 @@ func TestMain(m *testing.M) { for i := 1; i <= fleetN; i++ { targets[i-1] = fmt.Sprintf("dev-%d", i) } - fmt.Fprintf(os.Stderr, "[harness] suite 013-scale: SCALE_TARGETS=%d SCALE_REPLICAS=%d spare=%s placement=%d..%d per pod\n", - fleetN, replicas, spareSim, minPerPod, maxPerPod) + fmt.Fprintf(os.Stderr, "[harness] suite 013-scale: SCALE_TARGETS=%d SCALE_REPLICAS=%d cpu=%s/%s mem=%s/%s spare=%s placement=%d..%d per pod\n", + fleetN, replicas, cpuReq, cpuLim, memReq, memLim, spareSim, minPerPod, maxPerPod) require := append(append([]string{}, targets...), spareSim) os.Exit(harness.RunSuite(m, harness.Options{ @@ -74,8 +88,14 @@ func TestMain(m *testing.M) { GnmiGenConfigData: renderGnmiGenConfig(fleetN + 1), RequireTargets: require, Baseline: []string{"fixtures/baseline.yaml"}, - BaselineVars: map[string]any{"Replicas": replicas}, - AfterBaseline: applyFleetTargets, + BaselineVars: map[string]any{ + "Replicas": replicas, + "CPURequest": cpuReq, + "CPULimit": cpuLim, + "MemoryRequest": memReq, + "MemoryLimit": memLim, + }, + AfterBaseline: applyFleetTargets, }, &s)) } @@ -91,6 +111,14 @@ func envInt(name string, def int) int { return n } +func envStr(name, def string) string { + v := strings.TrimSpace(os.Getenv(name)) + if v == "" { + return def + } + return v +} + // placementBand is avg±20% with a floor that matches the design's 40–60 band // at the default 200/4 fleet. func placementBand(n, r int) (int, int) {