From 8f14ea36b829d8d04ab8a8bdf47ca354010d33ee Mon Sep 17 00:00:00 2001 From: as51340 Date: Fri, 9 Oct 2026 14:47:52 +0200 Subject: [PATCH 1/2] fix: Wait for the cache to show a StatefulSet shrink before the next pass replicaCounts applies the larger of the declared count and the one the cached StatefulSet carries. A pass that read the StatefulSet from before shedRetiredPods lowered it applied the old count again, then shrank it a second time once it found the member already gone; the StatefulSet controller could recreate the shed pod in between. The MemgraphCluster status write that ends the shedding pass requeues the cluster on a separate watch stream, so that pass can start before the StatefulSet event reaches the cache. shedRetiredPods now records the UID and generation the apply returned in an in-memory expectation (internal/expectations), and a pass whose cached StatefulSet is the same object at a lower generation ends before applying anything, until the StatefulSet event clears it. A restarted operator lists into a fresh cache that already holds the shrink, so nothing is persisted. --- CLAUDE.md | 1 + .../controller/memgraphcluster_controller.go | 79 +++++++++++----- .../memgraphcluster_controller_test.go | 59 ++++++++++++ internal/expectations/expectations.go | 91 +++++++++++++++++++ internal/expectations/expectations_test.go | 90 ++++++++++++++++++ 5 files changed, 298 insertions(+), 22 deletions(-) create mode 100644 internal/expectations/expectations.go create mode 100644 internal/expectations/expectations_test.go diff --git a/CLAUDE.md b/CLAUDE.md index 087508c..bd54cb6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -77,6 +77,7 @@ Test philosophy: assert external behavior, never internal call ordering or priva - **Logs live on the lib claim** (decided 2026-10-07; the separate `log-storage` claim and its four `log*` knobs are gone): `storage..fileLogging` (bool, default `true`, a default-on toggle like `configureCorePattern`) pins `--log-file=/var/lib/memgraph/logs/memgraph.log`, off pins `--log-file=""`. Never a path knob and `log_file` stays denied in `spec.flags`: Memgraph's `CleanLogsDir` deletes files older than `log_retention_days` from the log file's **parent directory**, so the file must sit in a directory of its own, never the claim root, and any path off a writable mount crash-loops on the read-only root. The empty value is load-bearing: the image's `memgraph.conf` sets `log_file` to `/var/log/memgraph/memgraph.log`. Toggling is an ordinary roll (no CEL pin, no claim template involved); turning it off leaves old log files on the claim, never cleaned by the operator. Logs count toward `libPVCSize`, which is why the defaults moved to 10Gi and `log_level=INFO`. - Both StatefulSets use `updateStrategy: OnDelete`, so **nothing but the operator ever restarts a workload pod**. A pod-template change no reconcile acts on takes effect never, which is what the `Updated` condition exists to report. A run-time flag change is deliberately *not* a pod-template change (see Memgraph flags above). `RollingUpdate` cannot express the required order (it sweeps highest ordinal to lowest, and `partition` is a descending cutoff, not a set), so a MAIN on any ordinal but 0 would be restarted mid-sweep and each such restart buys another failover. - The operator never promotes a MAIN outside bootstrap and the scale-down handover: a roll deletes the MAIN's pod and lets the Raft coordinators promote. That is only safe on a Memgraph reporting an unreachable MAIN as `role=main, health=down` — a release that vacates the `main` row makes `planner.Plan` believe there is no MAIN and race the failover. +- **The shrink is the one write with an expectation** (`internal/expectations`, decided 2026-10-09). The operator reads its StatefulSets from the informer cache, and `replicaCounts` applies `max(declared, cached spec.replicas)`, so a pass reading the StatefulSet from before `shedRetiredPods` lowered it would grow it back (and shrink it again once it found the member gone, the pod recreated in between). `shedRetiredPods` records the UID and generation the apply returned; `replicaCounts` marks a role whose cached StatefulSet is the same UID at a lower generation, and the pass ends before applying anything, waiting on the StatefulSet event (`Owns`). It is in-memory on purpose: a restarted operator lists into a fresh cache that already holds every write it made, so only the writing process can be ahead of its cache. Every other cached read is safe without one (the apply restates desired state, pod deletes carry a UID precondition, `rollout`'s `Stale` covers status, membership is read live over Bolt); a new write whose stale read would undo it, or a decision pushed into Memgraph from a cached count, needs one too. - Log messages follow Kubernetes style: capital first letter, no trailing period, past tense, object type named (see AGENTS.md for examples). - **External access** (`spec.externalAccess`, `type: LoadBalancer | Gateway`) is operator-owned on both sides: the external objects *and* the announced `bolt_server` of every exposed member. Both modes build the same Services, one shared by the coordinators plus one per data instance (selected by the StatefulSet pod-name label); with `LoadBalancer` they are the way in, with `Gateway` they are ClusterIPs behind one operator-owned Gateway (`gatewayClassName` required, listeners `coordinators-bolt` on 7687 and `data--bolt` on `dataPortBase + n`, default 9000) fed by one TCPRoute each. The address is discovered, never declared: external-dns hostname annotation (on the Service, or in Gateway mode on the TCPRoute, then the Gateway), then the reported hostname, then the IP, pod DNS until any exists (`Converged=False/ExternalAddressPending`). `{ordinal}` in per-instance annotation values is substituted with the pod ordinal and CEL requires it on a data hostname annotation. Only the bolt address ever leaves the cluster network; the operator itself always dials coordinators on pod DNS (`resources.CoordinatorEndpoints`), never through the LoadBalancer or Gateway. External objects follow the data pods the operator *runs* (applied count), are pruned by label once the spec stops describing them, and removing the block reverts every `bolt_server` to pod DNS. The Gateway API CRDs are never bundled: `main.go` discovers once at start (`controller.GatewayAPIServed`) whether `Gateway` and `TCPRoute` are served at `v1`, watches the kinds only if so, and a cluster asking for `Gateway` without that gets `Converged=False/ApplyFailed` naming what is missing. TCPRoute is `v1` only from Gateway API v1.6 (Envoy Gateway ≥ v1.9); older releases serve it as `v1alpha2` alone and v1.6 stops serving `v1alpha2`, so no single version works on both sides — the operator requires ≥ v1.6. The e2e suite gets LoadBalancer addresses from MetalLB and a Gateway from Envoy Gateway (`hack/kind-metallb.sh`, `hack/kind-envoy-gateway.sh`, run by `make setup-test-e2e`); envtest loads the Gateway API CRDs from the `sigs.k8s.io/gateway-api` module. - **TLS** (`spec.tls`) is presence-based per mode and takes one shared `kubernetes.io/tls` Secret per mode for both roles, fixed keys, no key-name knobs: a StatefulSet has one pod template, and Memgraph verifies no member hostname, so per-pod identities buy nothing. `bolt` (`secretName`; `tls.crt`/`tls.key` mounted read-only at `/etc/memgraph/ssl`, `--bolt-cert-file`/`--bolt-key-file`) also moves 9091 to https because Memgraph serves metrics from the Bolt server context. Everything downstream is derived, never a knob: the operator dials `bolt+ssc` (TLS, unverified — verification would force pod-DNS SANs and a CA onto a user-facing cert and buys nothing while Bolt is unauthenticated; Bolt auth is the trigger to revisit) and falls back to the other scheme once, because a roll turning TLS on or off leaves coordinators in the old mode while the spec says the new one (`memgraph.dialSchemes`); the ServiceMonitor switches to https with `insecureSkipVerify`. Probes are TCP and untouched. Rotation is not the operator's: the Secret is mounted without `subPath` so the kubelet refreshes it, and Memgraph picks it up on `RELOAD BOLT SERVER TLS` until the core reloads by mtime (planned upstream); the operator watches no Secret and never restarts for an in-place update, and a `secretName` change is an ordinary roll. `intraCluster` (`secretName`; `tls.crt`/`tls.key`/`ca.crt` mounted read-only at `/etc/memgraph/intra_cluster_tls`, the three `--cluster-*-file` flags always together) is mTLS proving membership, not identity: Memgraph verifies the chain and demands a client cert but checks no hostname, so one shared cert is the shape (nine of thirteen surveyed operators do the same). Its presence is pinned by a CEL transition rule on `MemgraphClusterSpec`, not on `TLSSpec`, because a rule on the block does not fire when the block is added or removed whole: a member with it cannot talk to one without, so a one-pod-at-a-time roll deadlocks on lag. `secretName` changes roll normally; a CA cut-over needs a bundled `ca.crt` in both Secrets, documented, not detected. Renewal is `RELOAD INTRA_CLUSTER TLS` per instance until the core reloads by mtime (Raft already does). diff --git a/internal/controller/memgraphcluster_controller.go b/internal/controller/memgraphcluster_controller.go index 805aebb..8cb40c3 100644 --- a/internal/controller/memgraphcluster_controller.go +++ b/internal/controller/memgraphcluster_controller.go @@ -52,6 +52,7 @@ import ( monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1" memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/expectations" "github.com/memgraph/kubernetes-operator/internal/memgraph" "github.com/memgraph/kubernetes-operator/internal/planner" "github.com/memgraph/kubernetes-operator/internal/resources" @@ -130,6 +131,11 @@ type MemgraphClusterReconciler struct { // ServiceMonitorAPIMissing says what the discovery found lacking when // ServiceMonitorAPI is false, so the resource is told what to install. ServiceMonitorAPIMissing string + + // shrinks are the StatefulSet shrinks this process applied and its cache + // may not show yet; a pass waits for them before resolving replica + // counts (see replicaCounts). The zero value is ready to use. + shrinks expectations.StatefulSets } // The install chart's ClusterRole is generated from these markers, so they are @@ -249,6 +255,13 @@ func (r *MemgraphClusterReconciler) reconcileCluster( if err != nil { return ctrl.Result{}, err } + if len(replicas.unseenShrinks) > 0 { + // The StatefulSet event that clears it requeues the cluster, so the + // delay is only a backstop. + log.Info("Deferred the pass until the cache shows the StatefulSet shrink it applied", + "statefulsets", replicas.unseenShrinks) + return ctrl.Result{RequeueAfter: requeueWhilePending}, nil + } // A grown storage size goes to the claims before anything is applied: the // apply restates the live claim templates, which Kubernetes refuses to @@ -685,13 +698,13 @@ func (r *MemgraphClusterReconciler) applyDesired( ctx context.Context, cluster *memgraphcomv1alpha1.MemgraphCluster, desired ...client.Object, -) (appliedGenerations, error) { - applied := appliedGenerations{} +) (appliedObjects, error) { + applied := appliedObjects{} for _, obj := range desired { if err := controllerutil.SetControllerReference(cluster, obj, r.Scheme); err != nil { return nil, fmt.Errorf("setting owner reference on %T %s: %w", obj, obj.GetName(), err) } - generation, err := r.apply(ctx, obj) + written, err := r.apply(ctx, obj) if err != nil { applyErr := fmt.Errorf("applying %T %s: %w", obj, obj.GetName(), err) msg := truncateMessage(applyErr.Error()) @@ -700,24 +713,31 @@ func (r *MemgraphClusterReconciler) applyDesired( notConvergedCondition(memgraphcomv1alpha1.ReasonApplyFailed, msg)) return nil, applyErr } - applied[appliedKey(obj)] = generation + applied[appliedKey(obj)] = appliedObject{uid: written.GetUID(), generation: written.GetGeneration()} } return applied, nil } -// appliedGenerations is the metadata.generation the API server assigned each -// object a pass applied, keyed by kind and name — two kinds share a name here, -// the headless Service and the StatefulSet of a role. -type appliedGenerations map[string]int64 +// appliedObjects is the UID and metadata.generation the API server +// returned for each object a pass applied, keyed by kind and name — two kinds +// share a name here, the headless Service and the StatefulSet of a role. +type appliedObjects map[string]appliedObject + +// appliedObject is one applied object as the API server returned it. +type appliedObject struct { + uid types.UID + generation int64 +} func appliedKey(obj client.Object) string { return obj.GetObjectKind().GroupVersionKind().Kind + "/" + obj.GetName() } -// statefulSet is the generation this pass applied to the named StatefulSet, or -// zero when the pass applied none — which is what a StatefulSet read from the -// cache is then measured against, and zero is what any generation clears. -func (a appliedGenerations) statefulSet(name string) int64 { +// statefulSet is the named StatefulSet as this pass applied it, or the zero +// value when the pass applied none — whose generation is what a StatefulSet +// read from the cache is then measured against, and zero is what any +// generation clears. +func (a appliedObjects) statefulSet(name string) appliedObject { return a["StatefulSet/"+name] } @@ -741,6 +761,10 @@ type roleReplicas struct { type replicaCounts struct { coordinators roleReplicas data roleReplicas + // unseenShrinks names the StatefulSets whose shrink this process applied + // and its cache does not show yet. While it is non-empty the counts above + // are not to be acted on. + unseenShrinks []string } // retirementMessage names the members a lowered count is shedding, and is empty @@ -828,6 +852,12 @@ func (r *MemgraphClusterReconciler) replicaCounts( if err != nil { return replicaCounts{}, err } + // A cached StatefulSet older than the shrink a previous pass applied + // still carries the count before it, and the max below would apply + // that count again: the pods just shed would come back. + if !r.shrinks.Satisfied(types.NamespacedName{Namespace: cluster.Namespace, Name: role.name}, sts) { + counts.unseenShrinks = append(counts.unseenShrinks, role.name) + } // The count the operator's own previous apply left, zero before the // first one. var current int32 @@ -1096,18 +1126,18 @@ func (r *MemgraphClusterReconciler) observeRollout( ctx context.Context, cluster *memgraphcomv1alpha1.MemgraphCluster, replicas replicaCounts, - applied appliedGenerations, + applied appliedObjects, ) (rolloutRoles, error) { var roles rolloutRoles coordinators, err := r.observeRolloutRole(ctx, cluster, replicas.coordinators, resources.CoordinatorPodSelector(cluster), resources.CoordinatorInstanceName, - applied.statefulSet(replicas.coordinators.name)) + applied.statefulSet(replicas.coordinators.name).generation) if err != nil { return rolloutRoles{}, err } data, err := r.observeRolloutRole(ctx, cluster, replicas.data, resources.DataPodSelector(cluster), resources.DataInstanceName, - applied.statefulSet(replicas.data.name)) + applied.statefulSet(replicas.data.name).generation) if err != nil { return rolloutRoles{}, err } @@ -2298,9 +2328,13 @@ func (r *MemgraphClusterReconciler) shedRetiredPods( } shrunk := role.build(cluster, role.replicas.declared) volumes.KeepLiveSizes(shrunk, role.replicas.statefulSet) - if _, err := r.applyDesired(ctx, cluster, shrunk); err != nil { + applied, err := r.applyDesired(ctx, cluster, shrunk) + if err != nil { return err } + written := applied.statefulSet(role.replicas.name) + r.shrinks.Expect(types.NamespacedName{Namespace: cluster.Namespace, Name: role.replicas.name}, + written.uid, written.generation) log.Info("Shrank a StatefulSet to the declared replica count", "statefulset", role.replicas.name, "replicas", role.replicas.declared) } @@ -2613,12 +2647,13 @@ func (r *MemgraphClusterReconciler) showInstances( // apply server-side-applies a desired object built by the resource builders. // Builders set only the fields the operator owns, so the converted apply // configuration claims exactly those fields for this controller. It returns the -// generation the API server assigned the object: the response is decoded back -// into the applied configuration, so no second read is needed for it. -func (r *MemgraphClusterReconciler) apply(ctx context.Context, obj client.Object) (int64, error) { +// object as the API server answered, its UID and generation included: the +// response is decoded back into the applied configuration, so no second read +// is needed for them. +func (r *MemgraphClusterReconciler) apply(ctx context.Context, obj client.Object) (metav1.Object, error) { content, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj) if err != nil { - return 0, fmt.Errorf("converting to unstructured: %w", err) + return nil, fmt.Errorf("converting to unstructured: %w", err) } u := &unstructured.Unstructured{Object: content} // Zero-valued struct fields survive the conversion; drop them so the @@ -2629,9 +2664,9 @@ func (r *MemgraphClusterReconciler) apply(ctx context.Context, obj client.Object if err := r.Apply(ctx, client.ApplyConfigurationFromUnstructured(u), client.FieldOwner(fieldOwner), client.ForceOwnership); err != nil { - return 0, err + return nil, err } - return u.GetGeneration(), nil + return u, nil } // SetupWithManager sets up the controller with the Manager. The Gateway API diff --git a/internal/controller/memgraphcluster_controller_test.go b/internal/controller/memgraphcluster_controller_test.go index d260c6f..6765657 100644 --- a/internal/controller/memgraphcluster_controller_test.go +++ b/internal/controller/memgraphcluster_controller_test.go @@ -115,6 +115,24 @@ func libClaim(sts *appsv1.StatefulSet) corev1.PersistentVolumeClaimSpec { return corev1.PersistentVolumeClaimSpec{} } +// laggingCache is a client whose reads of the given StatefulSets return the +// copies it holds, the way an informer cache serves an object until the watch +// event of a write reaches it. Everything else, writes included, goes through. +type laggingCache struct { + client.Client + statefulSets map[types.NamespacedName]*appsv1.StatefulSet +} + +func (c laggingCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if sts, ok := obj.(*appsv1.StatefulSet); ok { + if held, ok := c.statefulSets[key]; ok { + held.DeepCopyInto(sts) + return nil + } + } + return c.Client.Get(ctx, key, obj, opts...) +} + var _ = Describe("MemgraphCluster Controller", func() { const resourceNamespace = "default" @@ -1000,6 +1018,47 @@ var _ = Describe("MemgraphCluster Controller", func() { memgraphcomv1alpha1.ConditionConverged)).To(BeTrue()) }) + // The shrink is the one write a lagging cache would undo: the count a pass + // applies is the larger of the declared one and the one the cached + // StatefulSet carries, so a pass reading the pre-shrink StatefulSet would + // apply the old count again and bring the shed pod back. + It("should not grow a StatefulSet it shrank back while the cache shows the old count", func() { + baseline := bootstrapped() + setCounts(3, 1) + reconcileCluster(resourceName) + Expect(sinceBootstrap(baseline)).To(HaveLen(1)) + + beforeShrink := &appsv1.StatefulSet{} + get(resourceName+dataSuffix, beforeShrink) + reconcileCluster(resourceName) + shrunk := &appsv1.StatefulSet{} + get(resourceName+dataSuffix, shrunk) + Expect(*shrunk.Spec.Replicas).To(Equal(int32(1))) + + By("reconciling while the cache still serves the StatefulSet from before the shrink") + reconciler.Client = laggingCache{Client: k8sClient, statefulSets: map[types.NamespacedName]*appsv1.StatefulSet{ + client.ObjectKeyFromObject(beforeShrink): beforeShrink, + }} + result := reconcileCluster(resourceName) + // The generation, not just the count: a pass that grew the StatefulSet + // back would shrink it again further on, once it found the member + // already gone, and the end state alone would hide the pod the + // StatefulSet controller recreated in between. + afterLag := &appsv1.StatefulSet{} + get(resourceName+dataSuffix, afterLag) + Expect(afterLag.Generation).To(Equal(shrunk.Generation), "the StatefulSet must not be written at all") + Expect(result.RequeueAfter).To(BeNumerically(">", 0), "the pass waits for the cache, it does not give up") + Expect(sinceBootstrap(baseline)).To(HaveLen(1), "nothing is issued while the cache lags") + + By("converging once the cache shows the shrink") + reconciler.Client = k8sClient + reconcileCluster(resourceName) + Expect(replicas(dataSuffix)).To(Equal(int32(1))) + Expect(status().DataInstances).To(Equal(int32(1))) + Expect(apimeta.IsStatusConditionTrue(status().Conditions, + memgraphcomv1alpha1.ConditionConverged)).To(BeTrue()) + }) + // Memgraph refuses to unregister the MAIN, so a retiring instance holding it // is demoted and a survivor promoted in its place — within the same pass, on // the same leader connection, so the cluster is MAIN-less for the time diff --git a/internal/expectations/expectations.go b/internal/expectations/expectations.go new file mode 100644 index 0000000..e1d8e31 --- /dev/null +++ b/internal/expectations/expectations.go @@ -0,0 +1,91 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package expectations remembers the StatefulSet writes the operator made and +// its informer cache has not shown back yet. A pass reads its StatefulSets +// from the cache, which can lag a write the previous pass made; for most +// decisions that is harmless, because the apply restates the same desired +// state, but a shrink is the one write whose stale read undoes it: the replica +// count a pass applies is the larger of the declared one and the one the +// StatefulSet carries, so a cache still showing the pre-shrink count would +// apply it again and recreate the pods the shrink just shed. +// +// The pattern is the Kubernetes controllers' expectations, in the shape ECK +// uses for StatefulSets: the write's UID and generation are recorded, and a +// cached copy is trusted again once it is at least that generation, or is a +// different object, or is gone. Nothing is persisted. A restarted operator +// fills its cache from a fresh list that already holds every write it made, +// so only the process that made a write can be ahead of its own cache, and +// that process is the one holding the expectation. +package expectations + +import ( + "sync" + + appsv1 "k8s.io/api/apps/v1" + "k8s.io/apimachinery/pkg/types" +) + +// StatefulSets holds the outstanding expectation of each StatefulSet the +// operator wrote. Its zero value is ready to use, and it is safe for +// concurrent use. +// +// An expectation for a StatefulSet whose cluster is deleted before the cache +// catches up is never cleared, which leaves a few bytes behind and nothing +// else: a StatefulSet created under the same name later has a different UID, +// which satisfies it. +type StatefulSets struct { + mu sync.Mutex + expected map[types.NamespacedName]written +} + +// written is the StatefulSet a write left behind, as the API server returned +// it. +type written struct { + uid types.UID + generation int64 +} + +// Expect records a write to the named StatefulSet that the API server answered +// with the given UID and generation. A later write replaces an earlier one, so +// only the newest is waited for. +func (s *StatefulSets) Expect(name types.NamespacedName, uid types.UID, generation int64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.expected == nil { + s.expected = map[types.NamespacedName]written{} + } + s.expected[name] = written{uid: uid, generation: generation} +} + +// Satisfied reports whether cached, the named StatefulSet as the cache shows +// it (nil when the cache has none), can be trusted: it is no older than the +// last write expected of it. A satisfied expectation is forgotten. +func (s *StatefulSets) Satisfied(name types.NamespacedName, cached *appsv1.StatefulSet) bool { + s.mu.Lock() + defer s.mu.Unlock() + want, ok := s.expected[name] + if !ok { + return true + } + // Gone, or deleted and created again under the same name: either way the + // object written is not the one a pass would act on any more. + if cached == nil || cached.UID != want.uid || cached.Generation >= want.generation { + delete(s.expected, name) + return true + } + return false +} diff --git a/internal/expectations/expectations_test.go b/internal/expectations/expectations_test.go new file mode 100644 index 0000000..23e0b5f --- /dev/null +++ b/internal/expectations/expectations_test.go @@ -0,0 +1,90 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package expectations + +import ( + "testing" + + appsv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +var name = types.NamespacedName{Namespace: "default", Name: "cluster-data"} + +func statefulSet(uid types.UID, generation int64) *appsv1.StatefulSet { + return &appsv1.StatefulSet{ObjectMeta: metav1.ObjectMeta{ + Namespace: name.Namespace, Name: name.Name, UID: uid, Generation: generation, + }} +} + +func TestSatisfied(t *testing.T) { + for _, tc := range []struct { + desc string + cached *appsv1.StatefulSet + want bool + }{ + {"an older generation of the object written", statefulSet("a", 4), false}, + {"the generation written", statefulSet("a", 5), true}, + {"a newer generation, written since by someone else", statefulSet("a", 6), true}, + {"no StatefulSet at all", nil, true}, + {"another object under the same name", statefulSet("b", 1), true}, + } { + t.Run(tc.desc, func(t *testing.T) { + var s StatefulSets + s.Expect(name, "a", 5) + if got := s.Satisfied(name, tc.cached); got != tc.want { + t.Fatalf("Satisfied() = %v, want %v", got, tc.want) + } + }) + } +} + +func TestNothingExpected(t *testing.T) { + var s StatefulSets + if !s.Satisfied(name, statefulSet("a", 1)) { + t.Fatal("a StatefulSet nothing was written to must be trusted") + } +} + +// A satisfied expectation is forgotten, so a cache that later shows an older +// copy again — which an informer never does, but a test double can — is +// trusted rather than waited on forever. +func TestSatisfiedIsForgotten(t *testing.T) { + var s StatefulSets + s.Expect(name, "a", 5) + if !s.Satisfied(name, statefulSet("a", 5)) { + t.Fatal("the generation written must satisfy the expectation") + } + if !s.Satisfied(name, statefulSet("a", 4)) { + t.Fatal("a satisfied expectation must be forgotten") + } +} + +// Only the newest write is waited for: a second shrink before the cache shows +// the first raises the bar to the second. +func TestLaterWriteReplacesEarlier(t *testing.T) { + var s StatefulSets + s.Expect(name, "a", 5) + s.Expect(name, "a", 6) + if s.Satisfied(name, statefulSet("a", 5)) { + t.Fatal("the first write's generation must not satisfy the second") + } + if !s.Satisfied(name, statefulSet("a", 6)) { + t.Fatal("the second write's generation must satisfy it") + } +} From 29f23c1a322261d237c9d73ba0320d3db0668eb9 Mon Sep 17 00:00:00 2001 From: as51340 Date: Fri, 9 Oct 2026 15:08:02 +0200 Subject: [PATCH 2/2] test: Poll for the vmagent Deployment instead of waiting on it kubectl wait fails at once on an object that does not exist yet, and the operator creates the Deployment a moment after the patch, so the spec raced it. It now polls the Available condition, like the init container spec polls its pod. --- test/e2e/memgraphcluster_test.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/test/e2e/memgraphcluster_test.go b/test/e2e/memgraphcluster_test.go index f432ef3..1a76c0b 100644 --- a/test/e2e/memgraphcluster_test.go +++ b/test/e2e/memgraphcluster_test.go @@ -472,11 +472,17 @@ var _ = Describe("MemgraphCluster", Ordered, func() { _, err = utils.Run(cmd) Expect(err).NotTo(HaveOccurred()) + // Polled rather than `kubectl wait`, which fails at once on a Deployment + // that does not exist yet, and right after the patch the operator has + // not created it. By("waiting for the vmagent Deployment to become available under the restricted policy") - cmd = exec.Command("kubectl", "wait", "--for=condition=Available", "deployment/"+quickstartCluster.name+"-vmagent", - "-n", clusterNamespace, "--timeout=3m") - _, err = utils.Run(cmd) - Expect(err).NotTo(HaveOccurred(), "the vmagent never became available") + Eventually(func(g Gomega) { + cmd := exec.Command("kubectl", "get", "deployment", quickstartCluster.name+"-vmagent", "-n", clusterNamespace, + "-o", `jsonpath={.status.conditions[?(@.type=="Available")].status}`) + output, err := utils.Run(cmd) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(strings.TrimSpace(output)).To(Equal("True")) + }, 3*time.Minute, 5*time.Second).Should(Succeed(), "the vmagent never became available") By("waiting for every instance's up series to reach the sink with the external label") declared := quickstartCluster.coordinators + quickstartCluster.dataInstances