diff --git a/CLAUDE.md b/CLAUDE.md index e337bfc..0f3546e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,7 +47,7 @@ CI (`.github/workflows/`) runs `make lint-config`, `make lint`, `make test-unit` ## Architecture -The PRD defines seven modules with two pure cores and one mock seam. Keep this separation — it's what makes the logic testable without a cluster: +The PRD defines seven modules with two pure cores and one mock seam; an eighth (the rolling-restart decision) was added post-v1 as a third pure core. Keep this separation — it's what makes the logic testable without a cluster: 1. **API types** (`api/v1alpha1/`) — spec/status with kubebuilder + CEL markers. Replica counts (coordinators, data instances) are **mutable in both directions**, bounded by schema floors alone (coordinators ≥ 3 and odd, data instances ≥ 1) — enforced by the CRD, not a webhook. Defaults are declared as CRD schema defaults *and* mirrored as Go constants in `memgraphcluster_types.go` so resource builders behave correctly on specs that never passed admission (unit tests). 2. **Resource builders** — pure functions: spec in, desired Kubernetes objects out (one StatefulSet per role + headless Services; per-pod identity such as coordinator ID and advertised addresses derived from pod ordinals). No API calls, no side effects. Tested golden-style. @@ -55,14 +55,17 @@ The PRD defines seven modules with two pure cores and one mock seam. Keep this s 4. **Registration planner** — pure diff: declared topology + observed `SHOW INSTANCES` in, ordered registration commands out (empty when converged). Reconciliation semantics live here: read-before-write, idempotent, re-issue only missing registrations. A MAIN is promoted only when the cluster has none — at bootstrap, and after the planner itself demoted a data instance that is retiring; a MAIN that is staying is never overridden, because failover belongs to the Raft coordinators. A lowered count is the one removal the planner drives: `DEMOTE INSTANCE` the retiring MAIN, promote a survivor, `UNREGISTER INSTANCE` the retiring data instances, `REMOVE COORDINATOR` the retiring coordinators, and only then does the controller shed their pods (see `specs/operator-mvp/issues/15`, `16`). Moving MAIN off a retiring instance is the one step that can lose data, so it has a precondition nothing else does: a survivor that is both reachable and reported by `SHOW REPLICATION LAG` as holding every transaction the MAIN committed, in every database. Without one, the demotion, the promotion and that instance's unregistration are all left out of the plan and the retiring MAIN keeps serving — a scale-down that pauses, not one that drops writes. Because that state plans *nothing*, an empty plan is not proof a retirement finished: `planner.Retired` is what gates shedding the pods. One command breaks the pure-diff mould: `YIELD LEADERSHIP`, needed because Raft refuses to remove its own leader. It names no successor, so it is always a plan's **last** command and terminal — the controller requeues and re-observes under whichever coordinator won the election. 5. **Controller** (`internal/controller/`) — fetch CR, server-side-apply builder output, gate on pod readiness, run planner against the HA client, write status/conditions. 6. **Operator install chart** (`charts/memgraph-operator/`) — lives in this repo, cross-published to `memgraph.github.io/helm-charts` at release. Its `crds/` and `rbac/manager-rules.yaml` are **generated** (`make chart-sync`, verified by `make chart-verify`): the manager's ClusterRole comes from the `+kubebuilder:rbac` markers, so tightening or widening the controller's permissions means editing the markers, never the chart. The e2e suite installs the operator through this chart, so every scenario runs under the RBAC users get. The chart's `version` and its `appVersion` (the operator image tag) move **independently**: tag `v` releases the operator, `chart-` releases the chart alone — see `docs/releasing.md`. -7. **E2E harness** (`test/e2e/`, build tag `e2e`) — multi-node KinD with real Memgraph images. +7. **Rolling-restart decision** (`internal/rollout/`) — the third pure core: both roles' pods reduced to `{name, revisionHash, ready}` plus each StatefulSet's `UpdateRevision`, the `SHOW INSTANCES` view and `SHOW REPLICATION LAG` in, **exactly one** action out (`Done` / `Wait(reason)` / `Delete(pod)`). Both StatefulSets use `updateStrategy: OnDelete`, so the operator owns every pod restart and this decides which pod is next: data instances before coordinators, the observed MAIN last of its role, the Raft leader last of its. Nothing is persisted — pods already carrying the new revision *are* the ones already restarted, so a mid-roll spec revert or a Raft-driven MAIN move self-corrects. One action per pass and never a list, because every step re-gates on fresh lag. It issues no Bolt commands at all; a roll is invisible to the planner. +8. **E2E harness** (`test/e2e/`, build tag `e2e`) — multi-node KinD with real Memgraph images. -Test philosophy (from the PRD): assert external behavior, never internal call ordering or private state. Builders get golden tests, planner gets pure topology-diff cases, controller gets envtest with the HA client mocked. +Test philosophy (from the PRD): assert external behavior, never internal call ordering or private state. Builders get golden tests, planner gets pure topology-diff cases, the rolling-restart decision gets pure cases over pod revisions and observed cluster state, controller gets envtest with the HA client mocked. ## Conventions - Spec knob names mirror the HA Helm chart's vocabulary where the concept carries over (e.g. the `secrets.name` / `secrets.licenseKey` / `secrets.organizationKey` block) — check the chart before inventing a name. - No secret material in spec or status; secrets are consumed by reference only. - Storage is never deleted by the operator: no finalizer-based cleanup; PVC retention (deletion *and* scale-down) maps to the StatefulSet PVC retention policy (default `Retain`). The only cluster members the operator removes are the ones a lowered replica count retires; a coordinator removed from Raft keeps running and keeps its state on purpose, which is what makes re-growing onto a retained volume safe. -- Workload pods: non-root uid 101 / gid 103, seccomp RuntimeDefault, all capabilities dropped. +- Workload pods: non-root uid 101 / gid 103, seccomp RuntimeDefault, all capabilities dropped, `terminationGracePeriodSeconds: 300` (a ceiling, not a delay — an instance killed mid-shutdown recovers from its WAL and lengthens the catch-up a roll waits on). +- 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. `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. - Log messages follow Kubernetes style: capital first letter, no trailing period, past tense, object type named (see AGENTS.md for examples). diff --git a/api/v1alpha1/memgraphcluster_types.go b/api/v1alpha1/memgraphcluster_types.go index 2badccd..fab6db1 100644 --- a/api/v1alpha1/memgraphcluster_types.go +++ b/api/v1alpha1/memgraphcluster_types.go @@ -133,6 +133,16 @@ const ( // --for=condition=Converged` therefore means a scale is genuinely finished, // not merely accepted. ConditionConverged = "Converged" + + // ConditionUpdated is True when every workload pod runs the pod template the + // spec currently describes. Because both StatefulSets use updateStrategy + // OnDelete, Kubernetes replaces no pod on its own: the operator restarts them + // one at a time, data instances before coordinators, MAIN and the Raft leader + // last. It is kept apart from Converged deliberately — Converged answers + // "does the cluster have the declared members", this one answers "do they run + // the declared template", and a user looking at a False condition needs to + // know which of the two is happening. + ConditionUpdated = "Updated" ) // Condition reasons reported on MemgraphCluster status. Reasons are CamelCase @@ -198,6 +208,23 @@ const ( // survivors are down, or too far behind to catch up. ReasonNoCaughtUpSurvivor = "NoCaughtUpSurvivor" + // ReasonRollingRestartInProgress is set while the operator is restarting pods + // to bring them onto the pod template the spec currently describes. The + // message names the pod being restarted and why it is that one's turn, because + // the order is the whole safety argument: every data instance except MAIN + // first, then MAIN, then the coordinators with the Raft leader last. + ReasonRollingRestartInProgress = "RollingRestartInProgress" + + // ReasonWaitingForCatchUp is set while a rolling restart waits for the data + // instance it restarted last to hold every transaction the MAIN has committed + // again. Until it does, restarting the next pod would leave recent writes on + // the MAIN alone. + ReasonWaitingForCatchUp = "WaitingForCatchUp" + + // ReasonAllPodsUpdated is set when every workload pod runs the pod template + // the spec currently describes. + ReasonAllPodsUpdated = "AllPodsUpdated" + // ReasonMainElected is set when a data instance is observed as MAIN. ReasonMainElected = "MainElected" diff --git a/charts/memgraph-operator/rbac/manager-rules.yaml b/charts/memgraph-operator/rbac/manager-rules.yaml index 16eda07..fb4f0ab 100644 --- a/charts/memgraph-operator/rbac/manager-rules.yaml +++ b/charts/memgraph-operator/rbac/manager-rules.yaml @@ -13,6 +13,15 @@ # Generated from the +kubebuilder:rbac markers in the controller sources. # Regenerate with 'make chart-sync'; do not edit by hand. To widen or tighten # the permissions, edit the markers -- 'make chart-verify' fails if the two drift. +- apiGroups: + - "" + resources: + - pods + verbs: + - delete + - get + - list + - watch - apiGroups: - "" resources: diff --git a/cmd/main.go b/cmd/main.go index bddce0c..56cac9f 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -25,10 +25,14 @@ import ( // to ensure that exec-entrypoint and run can make use of them. _ "k8s.io/client-go/plugin/pkg/client/auth" + corev1 "k8s.io/api/core/v1" + klabels "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" "sigs.k8s.io/controller-runtime/pkg/log/zap" "sigs.k8s.io/controller-runtime/pkg/metrics/filters" @@ -38,6 +42,7 @@ import ( memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" "github.com/memgraph/kubernetes-operator/internal/controller" "github.com/memgraph/kubernetes-operator/internal/memgraph" + "github.com/memgraph/kubernetes-operator/internal/resources" // +kubebuilder:scaffold:imports ) @@ -160,8 +165,21 @@ func main() { Metrics: metricsServerOptions, WebhookServer: webhookServer, HealthProbeBindAddress: probeAddr, - LeaderElection: enableLeaderElection, - LeaderElectionID: "a5adec69.memgraph.com", + // Pods are cached, because the rolling restart needs each one's + // controller-revision-hash and readiness on every pass — but only this + // operator's own pods are. Watching every pod in the cluster to find them + // would cost memory proportional to somebody else's workload. + Cache: cache.Options{ + ByObject: map[client.Object]cache.ByObject{ + &corev1.Pod{}: { + Label: klabels.SelectorFromSet(klabels.Set{ + resources.ManagedByLabel: resources.ManagedByValue, + }), + }, + }, + }, + LeaderElection: enableLeaderElection, + LeaderElectionID: "a5adec69.memgraph.com", // LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily // when the Manager ends. This requires the binary to immediately end when the // Manager is stopped, otherwise, this setting is unsafe. Setting this significantly diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 7c6c5df..45cc667 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -4,6 +4,15 @@ kind: ClusterRole metadata: name: manager-role rules: +- apiGroups: + - "" + resources: + - pods + verbs: + - delete + - get + - list + - watch - apiGroups: - "" resources: diff --git a/internal/controller/fake_memgraph_test.go b/internal/controller/fake_memgraph_test.go index 9aef399..93dc337 100644 --- a/internal/controller/fake_memgraph_test.go +++ b/internal/controller/fake_memgraph_test.go @@ -211,7 +211,7 @@ func (c *fakeClient) ShowReplicationLag(context.Context) ([]memgraph.Replication lag = append(lag, memgraph.ReplicationLag{ Instance: instance.Name, Databases: []memgraph.DatabaseLag{{ - Database: "memgraph", + Database: memgraphDbName, CommittedTxns: 100 - behind, TxnsBehindMain: behind, }}, diff --git a/internal/controller/memgraphcluster_controller.go b/internal/controller/memgraphcluster_controller.go index 6f9eea3..58dd405 100644 --- a/internal/controller/memgraphcluster_controller.go +++ b/internal/controller/memgraphcluster_controller.go @@ -41,6 +41,7 @@ import ( "github.com/memgraph/kubernetes-operator/internal/memgraph" "github.com/memgraph/kubernetes-operator/internal/planner" "github.com/memgraph/kubernetes-operator/internal/resources" + "github.com/memgraph/kubernetes-operator/internal/rollout" ) // fieldOwner identifies this controller as the server-side-apply field @@ -86,11 +87,17 @@ type MemgraphClusterReconciler struct { // is needed to set those owner references: they block owner deletion, which // clusters running the OwnerReferencesPermissionEnforcement admission plugin // only allow with update access to the owner's finalizers. +// +// Pods are the one thing the operator deletes. Both StatefulSets use +// updateStrategy OnDelete, so replacing a pod whose template changed is the +// operator's job and nobody else's; get/list/watch reads their revision and +// readiness, and delete is the restart itself. // +kubebuilder:rbac:groups=memgraph.com,resources=memgraphclusters,verbs=get;list;watch // +kubebuilder:rbac:groups=memgraph.com,resources=memgraphclusters/status,verbs=get;patch // +kubebuilder:rbac:groups=memgraph.com,resources=memgraphclusters/finalizers,verbs=update // +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;patch // +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;patch +// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;delete // Reconcile drives the cluster toward the declared MemgraphCluster spec in // two stages. First it server-side-applies the builders' desired objects: one @@ -280,6 +287,128 @@ func (r *MemgraphClusterReconciler) replicaCounts( return counts, nil } +// rolloutRoles is both roles' pods as the rolling restart sees them. +type rolloutRoles struct { + coordinators rollout.Role + data rollout.Role +} + +// observeRollout reads both roles' pods and the revision their StatefulSet +// currently hashes its pod template to, which is everything the rolling restart +// needs about Kubernetes. It is pure observation: nothing is decided here. +func (r *MemgraphClusterReconciler) observeRollout( + ctx context.Context, + cluster *memgraphcomv1alpha1.MemgraphCluster, + replicas replicaCounts, +) (rolloutRoles, error) { + var roles rolloutRoles + coordinators, err := r.observeRolloutRole(ctx, cluster, replicas.coordinators, + resources.CoordinatorPodSelector(cluster), resources.CoordinatorInstanceName) + if err != nil { + return rolloutRoles{}, err + } + data, err := r.observeRolloutRole(ctx, cluster, replicas.data, + resources.DataPodSelector(cluster), resources.DataInstanceName) + if err != nil { + return rolloutRoles{}, err + } + roles.coordinators, roles.data = coordinators, data + return roles, nil +} + +// observeRolloutRole reads one role's pods, in ordinal order, each tagged with +// the Memgraph instance that runs on it. +// +// Pods are looked up by the name their ordinal gives them rather than by +// iterating whatever the list returned, so a pod that has been deleted and not +// yet recreated is simply absent from the result — which is how the rolling +// restart learns to wait for it, and what keeps a pod belonging to some other +// generation of the StatefulSet from being counted. +// +// A pod on its way out is not ready no matter what its conditions still say. Its +// containers keep passing their probes for as long as they take to shut down, and +// a restart that trusted that would delete the next pod while this one is still +// running. +func (r *MemgraphClusterReconciler) observeRolloutRole( + ctx context.Context, + cluster *memgraphcomv1alpha1.MemgraphCluster, + role roleReplicas, + selector map[string]string, + instanceName func(ordinal int32) string, +) (rollout.Role, error) { + observed := rollout.Role{Replicas: role.applied} + + var sts appsv1.StatefulSet + if err := r.Get(ctx, types.NamespacedName{Name: role.name, Namespace: cluster.Namespace}, &sts); err != nil { + if apierrors.IsNotFound(err) { + // Nothing has been provisioned yet, so there is no revision to measure + // pods against and nothing to restart. + return observed, nil + } + return rollout.Role{}, fmt.Errorf("getting StatefulSet %s: %w", role.name, err) + } + observed.UpdateRevision = sts.Status.UpdateRevision + + var pods corev1.PodList + if err := r.List(ctx, &pods, + client.InNamespace(cluster.Namespace), client.MatchingLabels(selector)); err != nil { + return rollout.Role{}, fmt.Errorf("listing pods of StatefulSet %s: %w", role.name, err) + } + byName := make(map[string]*corev1.Pod, len(pods.Items)) + for i := range pods.Items { + byName[pods.Items[i].Name] = &pods.Items[i] + } + + for ordinal := int32(0); ordinal < role.applied; ordinal++ { + pod, ok := byName[fmt.Sprintf("%s-%d", role.name, ordinal)] + if !ok { + continue + } + observed.Pods = append(observed.Pods, rollout.Pod{ + Name: pod.Name, + UID: string(pod.UID), + Instance: instanceName(ordinal), + Ordinal: ordinal, + RevisionHash: pod.Labels[appsv1.StatefulSetRevisionLabel], + Ready: pod.DeletionTimestamp == nil && podReady(pod), + }) + } + return observed, nil +} + +// podReady reports the pod's Ready condition. +func podReady(pod *corev1.Pod) bool { + for _, condition := range pod.Status.Conditions { + if condition.Type == corev1.PodReady { + return condition.Status == corev1.ConditionTrue + } + } + return false +} + +// restartPod deletes the pod a rolling restart picked, so its StatefulSet +// recreates it on the current pod template. The delete is conditioned on the UID +// that was observed: a pod already replaced between the observation and here is +// left alone rather than restarted twice, and a pod that is simply gone is not an +// error — the next pass re-observes and decides again. +func (r *MemgraphClusterReconciler) restartPod( + ctx context.Context, + cluster *memgraphcomv1alpha1.MemgraphCluster, + decision rollout.Decision, +) error { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: decision.Pod.Name, Namespace: cluster.Namespace}, + } + uid := types.UID(decision.Pod.UID) + err := r.Delete(ctx, pod, client.Preconditions{UID: &uid}) + switch { + case err == nil, apierrors.IsNotFound(err), apierrors.IsConflict(err): + return nil + default: + return fmt.Errorf("deleting pod %s to restart it: %w", decision.Pod.Name, err) + } +} + // currentReplicas is the replica count the operator's own previous apply left on // a role's StatefulSet, or zero when the cluster has not been provisioned yet. func (r *MemgraphClusterReconciler) currentReplicas( @@ -324,7 +453,15 @@ func (r *MemgraphClusterReconciler) reconcileRegistration( ) (ctrl.Result, error) { log := logf.FromContext(ctx) - ready, err := r.workloadsReady(ctx, cluster, replicas) + // Both roles' pods are read once per pass: the readiness gate needs to know + // whether a restart is under way to tolerate the pod it took down, and the + // restart itself needs the same view further down. + roles, err := r.observeRollout(ctx, cluster, replicas) + if err != nil { + return ctrl.Result{}, err + } + + ready, err := r.workloadsReady(ctx, cluster, replicas, roles) if err != nil { return ctrl.Result{}, err } @@ -417,13 +554,48 @@ func (r *MemgraphClusterReconciler) reconcileRegistration( return ctrl.Result{RequeueAfter: requeueAfterRegistration}, nil } - // Converged, but keep re-observing: a registration a pod loses later - // produces no watch event, so drift is only caught by resyncing. - log.Info("Confirmed cluster registration is converged") + // Registration is converged, so this is where a changed pod template gets + // rolled through the cluster. It is deliberately the only place: a + // retirement is still moving MAIN around and a pending registration means + // the cluster is not the one the spec describes, so neither is a moment to + // start deleting pods. converged := trueCondition(memgraphcomv1alpha1.ConditionConverged, memgraphcomv1alpha1.ReasonAllInstancesRegistered, fmt.Sprintf("All %d declared instances are registered", len(topology.Coordinators)+len(topology.DataInstances))) - if statusErr := r.writeStatus(ctx, cluster, latest, readyOrNot(latest.main), converged); statusErr != nil { + + switch decision := rollout.Next(roles.data, roles.coordinators, observed, lag); decision.Action { + case rollout.Delete: + // Reported before the pod goes, for the reason a rejected apply is: the + // next pass has to explain an absence it caused, and a restart nobody + // announced looks like the cluster losing a pod on its own. + if statusErr := r.writeStatus(ctx, cluster, latest, readyOrNot(latest.main), converged, + notUpdatedCondition(decision.Reason, decision.Message)); statusErr != nil { + return ctrl.Result{}, statusErr + } + if err := r.restartPod(ctx, cluster, decision); err != nil { + return ctrl.Result{}, err + } + log.Info("Deleted a workload pod to restart it onto the current pod template", + "pod", decision.Pod.Name, "reason", decision.Message) + return ctrl.Result{RequeueAfter: requeueWhilePending}, nil + + case rollout.Wait: + log.Info("Deferred the next pod restart", "reason", decision.Message) + if statusErr := r.writeStatus(ctx, cluster, latest, readyOrNot(latest.main), converged, + notUpdatedCondition(decision.Reason, decision.Message)); statusErr != nil { + return ctrl.Result{}, statusErr + } + return ctrl.Result{RequeueAfter: requeueWhilePending}, nil + } + + // Converged, but keep re-observing: a registration a pod loses later + // produces no watch event, so drift is only caught by resyncing. + log.Info("Confirmed cluster registration is converged") + updated := trueCondition(memgraphcomv1alpha1.ConditionUpdated, + memgraphcomv1alpha1.ReasonAllPodsUpdated, + "All workload pods run the pod template the spec describes") + if statusErr := r.writeStatus(ctx, cluster, latest, + readyOrNot(latest.main), converged, updated); statusErr != nil { return ctrl.Result{}, statusErr } return ctrl.Result{RequeueAfter: resyncInterval}, nil @@ -542,11 +714,18 @@ func lastObserved(cluster *memgraphcomv1alpha1.MemgraphCluster) observation { } } -// observedMain returns the name of the data instance reported as MAIN, or the -// empty string when none is elected yet. +// observedMain returns the name of the data instance reported as MAIN *and* +// reachable, or the empty string when the cluster has none it can serve writes +// from. +// +// Reachability is part of the question, not a refinement of it. A MAIN whose pod +// is gone keeps its role in the coordinators' Raft state and keeps being reported +// as MAIN, so a check on the role alone would claim the cluster serves writes for +// the whole failover window — including every window the rolling restart opens on +// purpose by deleting the MAIN's pod. func observedMain(observed []memgraph.Instance) string { for _, instance := range observed { - if instance.IsMain() { + if instance.IsMain() && instance.IsUp() { return instance.Name } } @@ -592,6 +771,12 @@ func notConvergedCondition(reason, message string) metav1.Condition { } } +func notUpdatedCondition(reason, message string) metav1.Condition { + return metav1.Condition{ + Type: memgraphcomv1alpha1.ConditionUpdated, Status: metav1.ConditionFalse, Reason: reason, Message: message, + } +} + // writeStatus patches the status subresource with the pass's observation and the // given conditions. It uses the status subresource exclusively — spec is never // touched — and skips the patch when nothing changed, so a converged cluster @@ -638,20 +823,44 @@ func (r *MemgraphClusterReconciler) writeStatus( // A StatefulSet the apply just created is not ready, not an error: the same lag // makes an absent StatefulSet the same waiting state as one whose pods have not // come up yet. +// +// One absence is tolerated: the pod a rolling restart itself took down. Without +// that, the first pod the restart deletes would make this gate false, the pass +// would return before ever connecting to a coordinator, and the restart could +// never learn whether that pod came back — a roll that deletes one pod and then +// waits forever. The exception is deliberately narrow, and both halves of the +// condition matter. It applies only to a role that has outdated pods, so a +// healthy cluster is still held to every pod being ready; and only when all of +// the role's pods exist, because a role short of its replicas is exactly the +// stale-informer case above — during a 3-to-4 scale-up a readyReplicas of 3 +// against an applied 4 would otherwise read as "one pod down, mid-roll, +// tolerated" and let registration run against a pod that does not exist yet. func (r *MemgraphClusterReconciler) workloadsReady( ctx context.Context, cluster *memgraphcomv1alpha1.MemgraphCluster, replicas replicaCounts, + roles rolloutRoles, ) (bool, error) { - for _, role := range []roleReplicas{replicas.coordinators, replicas.data} { + for _, role := range []struct { + replicas roleReplicas + rollout rollout.Role + }{ + {replicas.coordinators, roles.coordinators}, + {replicas.data, roles.data}, + } { var sts appsv1.StatefulSet - if err := r.Get(ctx, types.NamespacedName{Name: role.name, Namespace: cluster.Namespace}, &sts); err != nil { + name := role.replicas.name + if err := r.Get(ctx, types.NamespacedName{Name: name, Namespace: cluster.Namespace}, &sts); err != nil { if apierrors.IsNotFound(err) { return false, nil } - return false, fmt.Errorf("getting StatefulSet %s: %w", role.name, err) + return false, fmt.Errorf("getting StatefulSet %s: %w", name, err) + } + required := role.replicas.applied + if rollout.InProgress(role.rollout) && sts.Status.Replicas == required { + required-- } - if sts.Status.ReadyReplicas < role.applied { + if sts.Status.ReadyReplicas < required { return false, nil } } diff --git a/internal/controller/memgraphcluster_controller_test.go b/internal/controller/memgraphcluster_controller_test.go index 9997482..154bbd9 100644 --- a/internal/controller/memgraphcluster_controller_test.go +++ b/internal/controller/memgraphcluster_controller_test.go @@ -25,6 +25,7 @@ import ( . "github.com/onsi/gomega" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -41,6 +42,10 @@ import ( const ( coordinatorSuffix = "-coordinator" dataSuffix = "-data" + + // memgraphDbName is the value of the app.kubernetes.io/name label the operator + // stamps on everything, and the container name inside its pods. + memgraphDbName = "memgraph" ) // Non-default spec values the specs in this package override with, chosen so @@ -1157,13 +1162,16 @@ var _ = Describe("MemgraphCluster Controller", func() { coordinators: roleReplicas{name: resourceName + coordinatorSuffix, declared: 3, applied: 3}, data: roleReplicas{name: resourceName + dataSuffix, declared: 2, applied: 2}, } - ready, err := reconciler.workloadsReady(ctx, cluster, held) + // A zero rolloutRoles is a cluster with no restart under way, which is what + // keeps this about the count comparison alone: the gate only ever tolerates + // an unready pod while a role actually has pods left to restart. + ready, err := reconciler.workloadsReady(ctx, cluster, held, rolloutRoles{}) Expect(err).NotTo(HaveOccurred()) Expect(ready).To(BeTrue(), "the cluster is ready at the size this pass applies") grown := held grown.data.declared, grown.data.applied = 3, 3 - ready, err = reconciler.workloadsReady(ctx, cluster, grown) + ready, err = reconciler.workloadsReady(ctx, cluster, grown, rolloutRoles{}) Expect(err).NotTo(HaveOccurred()) Expect(ready).To(BeFalse(), "a pass applying 3 must not read 2-ready-of-2 as ready, whatever spec.replicas still says") @@ -1462,5 +1470,259 @@ var _ = Describe("MemgraphCluster Controller", func() { Expect(&cluster.Spec).To(Equal(specBefore), "status updates must never mutate spec") Expect(cluster.Status.Conditions).NotTo(BeEmpty()) }) + + // A MAIN whose pod is gone keeps its role in the coordinators' Raft state, so + // the role alone would claim the cluster serves writes for the whole failover + // window — including every window a rolling restart opens on purpose. + It("should report NotReady while the MAIN is unreachable", func() { + fake.setInstances(convergedCluster()) + reconcileCluster(resourceName) + markWorkloadsReady(resourceName) + reconcileCluster(resourceName) + Expect(status().Main).To(Equal("instance_0")) + + fake.setInstances([]memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + func() memgraph.Instance { + main := observedDataInstance(0, memgraph.RoleMain) + main.Health = "down" + return main + }(), + observedDataInstance(1, memgraph.RoleReplica), + }) + reconcileCluster(resourceName) + + Expect(status().Main).To(BeEmpty(), "an unreachable MAIN is not a MAIN the cluster can serve from") + ready := condition(memgraphcomv1alpha1.ConditionReady) + Expect(ready.Status).To(Equal(metav1.ConditionFalse)) + Expect(ready.Reason).To(Equal(memgraphcomv1alpha1.ReasonNoMainElected)) + Expect(fake.executedCommands()).To(BeEmpty(), + "the coordinators own the failover; the operator issues no promotion") + }) + }) + + Context("when a changed pod template has to be rolled through the cluster", func() { + const ( + resourceName = "mgc-rollout" + oldRevision = "mgc-rollout-6c9f8b7d5" + newRevision = "mgc-rollout-77b4c8f9d" + ) + + observedCoordinator := func(id int, role string) memgraph.Instance { + host := fmt.Sprintf("%s-coordinator-%d.%s-coordinator.%s.svc.cluster.local", + resourceName, id-1, resourceName, resourceNamespace) + return memgraph.Instance{ + Name: fmt.Sprintf("coordinator_%d", id), BoltServer: host + ":7687", + CoordinatorServer: host + ":12000", ManagementServer: host + ":10000", + Health: "up", Role: role, + } + } + observedDataInstance := func(i int, role string) memgraph.Instance { + return memgraph.Instance{Name: fmt.Sprintf("instance_%d", i), Health: "up", Role: role} + } + convergedCluster := func() []memgraph.Instance { + return []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + observedDataInstance(1, memgraph.RoleReplica), + } + } + condition := func(condType string) *metav1.Condition { + GinkgoHelper() + cluster := &memgraphcomv1alpha1.MemgraphCluster{} + get(resourceName, cluster) + return apimeta.FindStatusCondition(cluster.Status.Conditions, condType) + } + + // putPod stands in for the StatefulSet controller envtest does not run: it + // creates or replaces one role pod at the given revision, ready. + putPod := func(suffix, component string, ordinal int, revision string) { + GinkgoHelper() + name := fmt.Sprintf("%s%s-%d", resourceName, suffix, ordinal) + existing := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: resourceNamespace}} + Expect(client.IgnoreNotFound(k8sClient.Delete(ctx, existing))).To(Succeed()) + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: resourceNamespace, + Labels: map[string]string{ + "app.kubernetes.io/name": memgraphDbName, + "app.kubernetes.io/instance": resourceName, + "app.kubernetes.io/component": component, + "app.kubernetes.io/managed-by": "memgraph-operator", + appsv1.StatefulSetRevisionLabel: revision, + }, + }, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: memgraphDbName, Image: memgraphDbName}}}, + } + Expect(k8sClient.Create(ctx, pod)).To(Succeed()) + pod.Status.Conditions = []corev1.PodCondition{{ + Type: corev1.PodReady, Status: corev1.ConditionTrue, + LastTransitionTime: metav1.Now(), + }} + Expect(k8sClient.Status().Update(ctx, pod)).To(Succeed()) + } + + // putPods places every pod of both roles at one revision. + putPods := func(revision string) { + GinkgoHelper() + for ordinal := range 3 { + putPod(coordinatorSuffix, "coordinator", ordinal, revision) + } + for ordinal := range 2 { + putPod(dataSuffix, "data", ordinal, revision) + } + } + + // declareRevision publishes the revision both StatefulSets' current pod + // template hashes to, which is what makes the pods above outdated. + declareRevision := func(revision string) { + GinkgoHelper() + for _, suffix := range []string{coordinatorSuffix, dataSuffix} { + sts := &appsv1.StatefulSet{} + get(resourceName+suffix, sts) + sts.Status.UpdateRevision = revision + Expect(k8sClient.Status().Update(ctx, sts)).To(Succeed()) + } + } + + podExists := func(suffix string, ordinal int) bool { + GinkgoHelper() + pod := &corev1.Pod{} + err := k8sClient.Get(ctx, types.NamespacedName{ + Name: fmt.Sprintf("%s%s-%d", resourceName, suffix, ordinal), + Namespace: resourceNamespace, + }, pod) + if apierrors.IsNotFound(err) { + return false + } + Expect(err).NotTo(HaveOccurred()) + return pod.DeletionTimestamp == nil + } + + BeforeEach(func() { + resource := &memgraphcomv1alpha1.MemgraphCluster{ + ObjectMeta: metav1.ObjectMeta{Name: resourceName, Namespace: resourceNamespace}, + } + Expect(k8sClient.Create(ctx, resource)).To(Succeed()) + + fake.setInstances(convergedCluster()) + reconcileCluster(resourceName) + markWorkloadsReady(resourceName) + reconcileCluster(resourceName) + }) + + AfterEach(func() { + cluster := &memgraphcomv1alpha1.MemgraphCluster{} + get(resourceName, cluster) + Expect(k8sClient.Delete(ctx, cluster)).To(Succeed()) + deleteOwned(resourceName) + Expect(k8sClient.DeleteAllOf(ctx, &corev1.Pod{}, + client.InNamespace(resourceNamespace), + client.MatchingLabels{"app.kubernetes.io/instance": resourceName}, + client.GracePeriodSeconds(0), + )).To(Succeed()) + }) + + It("should report Updated once every pod runs the declared template", func() { + putPods(newRevision) + declareRevision(newRevision) + reconcileCluster(resourceName) + + updated := condition(memgraphcomv1alpha1.ConditionUpdated) + Expect(updated).NotTo(BeNil()) + Expect(updated.Status).To(Equal(metav1.ConditionTrue)) + Expect(updated.Reason).To(Equal(memgraphcomv1alpha1.ReasonAllPodsUpdated)) + }) + + // The whole order in one spec: replicas before MAIN, data plane before + // coordinators, Raft leader last, one pod at a time throughout. + It("should restart data pods before coordinators, MAIN and the leader last", func() { + putPods(oldRevision) + declareRevision(newRevision) + + // instance_0 is MAIN, so the replica on ordinal 1 goes first. + reconcileCluster(resourceName) + Expect(podExists(dataSuffix, 1)).To(BeFalse(), "the non-MAIN data pod is restarted first") + Expect(podExists(dataSuffix, 0)).To(BeTrue(), "the MAIN's pod is not touched yet") + Expect(podExists(coordinatorSuffix, 2)).To(BeTrue(), "coordinators wait for the data plane") + updated := condition(memgraphcomv1alpha1.ConditionUpdated) + Expect(updated.Status).To(Equal(metav1.ConditionFalse)) + Expect(updated.Reason).To(Equal(memgraphcomv1alpha1.ReasonRollingRestartInProgress)) + + // It comes back on the new revision, reachable and caught up. + putPod(dataSuffix, "data", 1, newRevision) + reconcileCluster(resourceName) + Expect(podExists(dataSuffix, 0)).To(BeFalse(), "the MAIN's pod is restarted last of its role") + Expect(podExists(coordinatorSuffix, 2)).To(BeTrue()) + + // The coordinators fail over to instance_1, and the old MAIN returns as a + // replica — which is what the operator observes rather than arranges. + fake.setInstances([]memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleReplica), + observedDataInstance(1, memgraph.RoleMain), + }) + putPod(dataSuffix, "data", 0, newRevision) + + // Data done: coordinator_1 leads on ordinal 0, so ordinal 2 goes first. + reconcileCluster(resourceName) + Expect(podExists(coordinatorSuffix, 2)).To(BeFalse()) + Expect(podExists(coordinatorSuffix, 0)).To(BeTrue(), "the Raft leader's pod is last") + + putPod(coordinatorSuffix, "coordinator", 2, newRevision) + reconcileCluster(resourceName) + Expect(podExists(coordinatorSuffix, 1)).To(BeFalse()) + Expect(podExists(coordinatorSuffix, 0)).To(BeTrue()) + + putPod(coordinatorSuffix, "coordinator", 1, newRevision) + reconcileCluster(resourceName) + Expect(podExists(coordinatorSuffix, 0)).To(BeFalse(), "the leader goes once nothing else is left") + + putPod(coordinatorSuffix, "coordinator", 0, newRevision) + reconcileCluster(resourceName) + Expect(condition(memgraphcomv1alpha1.ConditionUpdated).Status).To(Equal(metav1.ConditionTrue)) + Expect(fake.executedCommands()).To(BeEmpty(), + "a rolling restart issues no registration commands at all") + }) + + It("should not restart the MAIN while no replica is caught up", func() { + putPods(oldRevision) + putPod(dataSuffix, "data", 1, newRevision) + declareRevision(newRevision) + fake.setBehind("instance_1", 7) + + reconcileCluster(resourceName) + + Expect(podExists(dataSuffix, 0)).To(BeTrue(), "the MAIN keeps serving; the roll waits") + updated := condition(memgraphcomv1alpha1.ConditionUpdated) + Expect(updated.Status).To(Equal(metav1.ConditionFalse)) + Expect(updated.Reason).To(Equal(memgraphcomv1alpha1.ReasonNoCaughtUpSurvivor)) + }) + + // Registration convergence comes first: a cluster missing a registration is + // not the one the spec describes, so it is no moment to start deleting pods. + It("should not restart any pod while a registration is pending", func() { + putPods(oldRevision) + declareRevision(newRevision) + fake.setInstances([]memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + }) + + reconcileCluster(resourceName) + + Expect(podExists(dataSuffix, 1)).To(BeTrue()) + Expect(podExists(coordinatorSuffix, 2)).To(BeTrue()) + Expect(fake.executedCommands()).To(ContainElement(ContainSubstring("REGISTER INSTANCE instance_1"))) + }) }) }) diff --git a/internal/controller/memgraphcluster_validation_test.go b/internal/controller/memgraphcluster_validation_test.go index b047b00..266863f 100644 --- a/internal/controller/memgraphcluster_validation_test.go +++ b/internal/controller/memgraphcluster_validation_test.go @@ -328,7 +328,7 @@ var _ = Describe("MemgraphCluster CRD validation", func() { CSI: &corev1.CSIVolumeSource{ Driver: "secrets-store.csi.k8s.io", ReadOnly: ptr.To(true), - VolumeAttributes: map[string]string{"secretProviderClass": "memgraph"}, + VolumeAttributes: map[string]string{"secretProviderClass": memgraphDbName}, }, }, }}, @@ -350,7 +350,7 @@ var _ = Describe("MemgraphCluster CRD validation", func() { csi := stored.Spec.ExtraVolumes.Coordinators[0].CSI Expect(csi).NotTo(BeNil()) Expect(csi.Driver).To(Equal("secrets-store.csi.k8s.io")) - Expect(csi.VolumeAttributes).To(HaveKeyWithValue("secretProviderClass", "memgraph")) + Expect(csi.VolumeAttributes).To(HaveKeyWithValue("secretProviderClass", memgraphDbName)) Expect(stored.Spec.ExtraVolumeMounts.Data[0].MountPath).To(Equal("/etc/memgraph/ssl")) Expect(stored.Spec.ExtraVolumeMounts.Coordinators).To(BeEmpty()) diff --git a/internal/memgraph/client.go b/internal/memgraph/client.go index 9200d97..f23f9c9 100644 --- a/internal/memgraph/client.go +++ b/internal/memgraph/client.go @@ -83,9 +83,25 @@ type CoordinatorSpec struct { // Name returns the instance name Memgraph derives from the coordinator ID and // reports in SHOW INSTANCES. func (c CoordinatorSpec) Name() string { - return fmt.Sprintf("coordinator_%d", c.ID) + return fmt.Sprintf(coordinatorNameFormat, c.ID) } +// CoordinatorIDFromName is the inverse of Name: the Raft ID of the coordinator a +// view names. It sits here rather than with its callers so that the format and its +// parser cannot drift — a changed name would otherwise leave the parser silently +// matching nothing. +func CoordinatorIDFromName(name string) (int32, error) { + var id int32 + if _, err := fmt.Sscanf(name, coordinatorNameFormat, &id); err != nil { + return 0, fmt.Errorf("parsing coordinator name %q: %w", name, err) + } + return id, nil +} + +// coordinatorNameFormat is how Memgraph derives a coordinator's instance name +// from its Raft ID, stated once for both directions. +const coordinatorNameFormat = "coordinator_%d" + // DataInstanceSpec declares one data instance to register with the cluster. type DataInstanceSpec struct { Name string diff --git a/internal/planner/planner_test.go b/internal/planner/planner_test.go index 31ba7e0..13e2672 100644 --- a/internal/planner/planner_test.go +++ b/internal/planner/planner_test.go @@ -175,6 +175,17 @@ func downDataInstance(i int) memgraph.Instance { return instance } +// mainDownDataInstance is the MAIN with its pod gone: Raft still records it as the +// current MAIN, and the coordinator leader cannot reach it. This is what a +// coordinator-driven failover looks like from the moment the MAIN dies until a +// successor is promoted, and what the rolling restart deliberately creates when it +// deletes the MAIN's pod. +func mainDownDataInstance(i int) memgraph.Instance { + instance := observedDataInstance(i, memgraph.RoleMain) + instance.Health = "down" + return instance +} + // caughtUp is the SHOW REPLICATION LAG view of the given data instances with every // one of them holding all of the MAIN's transactions: the state that lets a // retiring MAIN hand over. The MAIN reports itself in the view too, at zero @@ -870,6 +881,27 @@ func TestPlan(t *testing.T) { planner.UnregisterInstance{Name: thirdInstance}, }, }, + // The property the sequenced rolling restart rests on. It deletes the MAIN's + // pod on purpose and leaves the promotion to the coordinators, so the state + // below happens on every upgrade: MAIN still holds its role in Raft, but the + // leader cannot reach it. The planner must plan nothing at all — a promotion + // here would be the operator racing the failover it just triggered, which is + // two control systems choosing a MAIN at once. + // + // It only holds on a Memgraph that keeps reporting role=main for an + // unreachable MAIN. A release that reports role=unknown instead vacates the + // main row, and this same view would take the bootstrap promotion branch. + { + name: "an unreachable MAIN is left to the coordinators, not promoted around", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + mainDownDataInstance(0), + observedDataInstance(1, memgraph.RoleReplica), + }, + want: nil, + }, } for _, tc := range cases { diff --git a/internal/resources/resources.go b/internal/resources/resources.go index 7d949e7..9c52c12 100644 --- a/internal/resources/resources.go +++ b/internal/resources/resources.go @@ -36,6 +36,13 @@ const ( coordinatorComponent = "coordinator" dataComponent = "data" + + // ManagedByLabel and ManagedByValue mark every object the operator builds. + // They are exported because the manager scopes its Pod cache to them: the + // rolling restart needs per-pod revisions, and caching every pod in the + // cluster to get them would be a rude surprise on a large one. + ManagedByLabel = "app.kubernetes.io/managed-by" + ManagedByValue = "memgraph-operator" ) // Named container and Service port names shared by both roles. @@ -70,10 +77,22 @@ func labels( l := make(map[string]string, len(custom)+4) maps.Copy(l, custom) maps.Copy(l, selectorLabels(cluster, component)) - l["app.kubernetes.io/managed-by"] = "memgraph-operator" + l[ManagedByLabel] = ManagedByValue return l } +// CoordinatorPodSelector matches the pods of this cluster's coordinator +// StatefulSet, and DataPodSelector those of its data StatefulSet. Both are the +// StatefulSets' own selectors, so they cannot drift from the pods they describe. +func CoordinatorPodSelector(cluster *memgraphcomv1alpha1.MemgraphCluster) map[string]string { + return selectorLabels(cluster, coordinatorComponent) +} + +// DataPodSelector matches the pods of this cluster's data StatefulSet. +func DataPodSelector(cluster *memgraphcomv1alpha1.MemgraphCluster) map[string]string { + return selectorLabels(cluster, dataComponent) +} + // selectorLabels returns the immutable subset of labels used as StatefulSet // and Service selectors. func selectorLabels(cluster *memgraphcomv1alpha1.MemgraphCluster, component string) map[string]string { diff --git a/internal/resources/statefulset.go b/internal/resources/statefulset.go index 770e4e6..114d124 100644 --- a/internal/resources/statefulset.go +++ b/internal/resources/statefulset.go @@ -56,6 +56,16 @@ const ( // Container names of the two optional containers core dumps bring along. corePatternContainerName = "init-core-pattern" uploaderContainerName = "core-dumps-uploader" + + // terminationGracePeriod is how long a pod gets to shut down cleanly before + // SIGKILL. Kubernetes' own default of 30 seconds was harmless while nothing + // routinely deleted these pods; it is wrong now that the operator deletes + // every one of them on every pod-template change, because an instance killed + // mid-shutdown recovers from its write-ahead log on startup and lengthens + // exactly the catch-up the rolling restart then waits on. This is a ceiling + // and not a delay — an instance that exits in two seconds costs two seconds — + // so it is a constant rather than a knob until someone needs a different one. + terminationGracePeriod int64 = 300 ) // CoordinatorStatefulSet builds the single StatefulSet running all @@ -391,7 +401,21 @@ func statefulSet( Replicas: ptr.To(replicas), ServiceName: name, PodManagementPolicy: appsv1.ParallelPodManagement, - Selector: &metav1.LabelSelector{MatchLabels: selectorLabels(cluster, component)}, + // The operator replaces pods itself, one at a time and in an order + // Kubernetes cannot express: data instances before coordinators, the + // MAIN last, the Raft leader last. RollingUpdate sweeps highest ordinal + // to lowest and `partition` is a descending cutoff rather than a set, so + // a MAIN on any ordinal but 0 would be restarted mid-sweep and every + // such restart costs another coordinator-driven failover. + // + // The cost of this is real and permanent: nothing but the operator will + // ever restart one of these pods again, so a pod-template change no + // reconcile acts on takes effect never. That is what the Updated + // condition is for. + UpdateStrategy: appsv1.StatefulSetUpdateStrategy{ + Type: appsv1.OnDeleteStatefulSetStrategyType, + }, + Selector: &metav1.LabelSelector{MatchLabels: selectorLabels(cluster, component)}, // The StatefulSet controller is the only thing that ever deletes // this cluster's storage; the operator owns no finalizer and runs // no cleanup of its own. Both halves of the policy follow the one @@ -408,8 +432,9 @@ func statefulSet( Labels: labels(cluster, component, role.podLabels), }, Spec: corev1.PodSpec{ - InitContainers: podInitContainers(spec, role), - Containers: podContainers(container, role), + TerminationGracePeriodSeconds: ptr.To(terminationGracePeriod), + InitContainers: podInitContainers(spec, role), + Containers: podContainers(container, role), SecurityContext: &corev1.PodSecurityContext{ RunAsUser: ptr.To(memgraphUserID), RunAsGroup: ptr.To(memgraphGroupID), diff --git a/internal/resources/statefulset_test.go b/internal/resources/statefulset_test.go index 1bea14a..d313de9 100644 --- a/internal/resources/statefulset_test.go +++ b/internal/resources/statefulset_test.go @@ -386,13 +386,23 @@ func TestCoordinatorStatefulSetDefaults(t *testing.T) { Replicas: ptr.To(int32(3)), ServiceName: coordinatorName, PodManagementPolicy: appsv1.ParallelPodManagement, - Selector: &metav1.LabelSelector{MatchLabels: expectedSelectorLabels(coordinatorComponent)}, + // The operator replaces these pods itself, one at a time and MAIN or Raft + // leader last, which no RollingUpdate can express. + UpdateStrategy: appsv1.StatefulSetUpdateStrategy{ + Type: appsv1.OnDeleteStatefulSetStrategyType, + }, + Selector: &metav1.LabelSelector{MatchLabels: expectedSelectorLabels(coordinatorComponent)}, PersistentVolumeClaimRetentionPolicy: expectedRetentionPolicy( appsv1.RetainPersistentVolumeClaimRetentionPolicyType), VolumeClaimTemplates: expectedClaimTemplates(), Template: corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{Labels: expectedLabels(coordinatorComponent)}, Spec: corev1.PodSpec{ + // Kubernetes' 30-second default is too short for a database that is + // now restarted on every pod-template change: an instance killed + // mid-shutdown recovers from its WAL and lengthens the catch-up the + // rolling restart waits on. + TerminationGracePeriodSeconds: ptr.To(int64(300)), Containers: []corev1.Container{{ Name: memgraphName, Image: defaultImageRef, @@ -441,13 +451,21 @@ func TestDataStatefulSetDefaults(t *testing.T) { Replicas: ptr.To(int32(2)), ServiceName: dataName, PodManagementPolicy: appsv1.ParallelPodManagement, - Selector: &metav1.LabelSelector{MatchLabels: expectedSelectorLabels(dataComponent)}, + UpdateStrategy: appsv1.StatefulSetUpdateStrategy{ + Type: appsv1.OnDeleteStatefulSetStrategyType, + }, + Selector: &metav1.LabelSelector{MatchLabels: expectedSelectorLabels(dataComponent)}, PersistentVolumeClaimRetentionPolicy: expectedRetentionPolicy( appsv1.RetainPersistentVolumeClaimRetentionPolicyType), VolumeClaimTemplates: expectedClaimTemplates(), Template: corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{Labels: expectedLabels(dataComponent)}, Spec: corev1.PodSpec{ + // Kubernetes' 30-second default is too short for a database that is + // now restarted on every pod-template change: an instance killed + // mid-shutdown recovers from its WAL and lengthens the catch-up the + // rolling restart waits on. + TerminationGracePeriodSeconds: ptr.To(int64(300)), Containers: []corev1.Container{{ Name: memgraphName, Image: defaultImageRef, diff --git a/internal/resources/topology.go b/internal/resources/topology.go index 7105de4..4625aa2 100644 --- a/internal/resources/topology.go +++ b/internal/resources/topology.go @@ -37,6 +37,53 @@ func DeclaredDataInstances(cluster *memgraphcomv1alpha1.MemgraphCluster) int32 { return normalize(cluster.Spec).dataInstances } +// CoordinatorID is the Raft coordinator ID of the coordinator running on the pod +// with the given ordinal. IDs are 1-based because Memgraph treats ID 0 as unset. +func CoordinatorID(ordinal int32) int32 { + return ordinal + 1 +} + +// CoordinatorInstanceName and DataInstanceName are the names the members running +// on a pod ordinal are known by in SHOW INSTANCES. +// +// They are exported because this mapping has to exist in exactly one place. +// Anything that matches an observed cluster row against a pod — the rolling +// restart, which has nothing but pods to work from — needs the same derivation the +// builders and the declared topology use, and a second spelling of it would fail +// quietly: a name that is merely wrong matches no row at all, so the caller +// concludes the instance is absent rather than that it asked the wrong question. +func CoordinatorInstanceName(ordinal int32) string { + return memgraph.CoordinatorSpec{ID: CoordinatorID(ordinal)}.Name() +} + +// DataInstanceName is the SHOW INSTANCES name of the data instance on the pod +// with the given ordinal. +func DataInstanceName(ordinal int32) string { + return fmt.Sprintf("instance_%d", ordinal) +} + +// CoordinatorOrdinal and DataInstanceOrdinal are the inverses: the ordinal of the +// pod running the member an observed view names. They live next to the functions +// they invert so the two cannot drift apart, and they are what anything holding a +// name and needing the pod behind it uses — the e2e suite reading MAIN out of +// SHOW INSTANCES, for one. +func CoordinatorOrdinal(name string) (int32, error) { + id, err := memgraph.CoordinatorIDFromName(name) + if err != nil { + return 0, err + } + return id - 1, nil +} + +// DataInstanceOrdinal is the ordinal of the pod running the named data instance. +func DataInstanceOrdinal(name string) (int32, error) { + var ordinal int32 + if _, err := fmt.Sscanf(name, "instance_%d", &ordinal); err != nil { + return 0, fmt.Errorf("parsing data instance name %q: %w", name, err) + } + return ordinal, nil +} + // DeclaredTopology derives the registration topology the planner drives the // cluster toward. Identity follows the pod ordinal exactly as the workload // pods advertise it: coordinator ordinal N is Raft coordinator N+1 (Memgraph @@ -117,7 +164,7 @@ func coordinator( ) memgraph.CoordinatorSpec { fqdn := podFQDN(cluster, CoordinatorName(cluster), spec, ordinal) return memgraph.CoordinatorSpec{ - ID: ordinal + 1, + ID: CoordinatorID(ordinal), BoltServer: hostPort(fqdn, spec.ports.bolt), CoordinatorServer: hostPort(fqdn, spec.ports.coordinator), ManagementServer: hostPort(fqdn, spec.ports.management), @@ -135,7 +182,7 @@ func dataInstance( ) memgraph.DataInstanceSpec { fqdn := podFQDN(cluster, DataName(cluster), spec, ordinal) return memgraph.DataInstanceSpec{ - Name: fmt.Sprintf("instance_%d", ordinal), + Name: DataInstanceName(ordinal), BoltServer: hostPort(fqdn, spec.ports.bolt), ManagementServer: hostPort(fqdn, spec.ports.management), ReplicationServer: hostPort(fqdn, spec.ports.replication), diff --git a/internal/rollout/rollout.go b/internal/rollout/rollout.go new file mode 100644 index 0000000..31c3d1b --- /dev/null +++ b/internal/rollout/rollout.go @@ -0,0 +1,458 @@ +/* +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 rollout decides which single pod a changed pod template lets the +// operator restart next. Both role StatefulSets use updateStrategy OnDelete, so +// Kubernetes replaces no pod on its own and this is the only thing that ever +// does — which is also why the trigger is any revision change and not an image +// change: a template edit nobody rolls is a cluster frozen on an old spec. +// +// The decision is a pure function of what is observed — pods with their revision +// and readiness, the coordinator leader's SHOW INSTANCES view, and SHOW +// REPLICATION LAG — and it returns exactly one action per pass. One action and +// never a list, because every step is re-gated on a fresh observation: lag +// measured one pod ago says nothing about the next. +// +// Nothing is remembered between passes. The pods already carrying the new +// revision *are* the ones already restarted, so "what is left" and "what must +// have caught up" are both read off the cluster rather than tracked. That is what +// makes a spec reverted halfway through, or Raft moving MAIN or coordinator +// leadership mid-roll, self-correcting: the next pass simply re-derives the +// answer, with nothing to unwind. +// +// The order is the whole point. Data instances roll before coordinators, the +// observed MAIN is the last data pod to go, and the observed Raft leader is the +// last coordinator. Restarting the MAIN costs one coordinator-driven failover, so +// it happens once, at the end, when every instance Raft could promote in its +// place is already running the new revision. Kubernetes' own RollingUpdate cannot +// express that — it sweeps highest ordinal to lowest, and partition is a +// descending cutoff rather than a set, so a MAIN on any ordinal but 0 is restarted +// mid-sweep and each such restart buys another failover. +// +// The operator never promotes anything here. Killing the MAIN's pod leaves the +// promotion to the Raft coordinators, which is what keeps two control systems +// from choosing a MAIN at once. That is only safe on a Memgraph that reports an +// unreachable MAIN as role=main with health=down: a release that vacates the main +// row instead leaves planner.Plan believing the cluster has no MAIN, and it will +// race the failover with a promotion of its own on every single restart. +package rollout + +import ( + "fmt" + + memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/memgraph" +) + +// Action is what the operator does with the decision. +type Action int + +const ( + // Done means every pod of both roles runs its StatefulSet's current + // revision: there is nothing to restart. + Done Action = iota + + // Wait means a restart is outstanding but may not proceed yet. The reason + // and message say what is being waited for, and go straight onto the + // resource — "why is my upgrade not moving" has to be answerable from + // kubectl describe alone. + Wait + + // Delete means the named pod is the next one to restart. Its StatefulSet + // recreates it at the current revision. + Delete +) + +// Pod is one workload pod as the decision sees it: which Memgraph instance runs +// on it, which pod-template revision it carries, and whether Kubernetes +// considers it ready. +type Pod struct { + // Name is the pod to delete. + Name string + + // UID is the pod's identity at the moment it was observed. The delete is + // conditioned on it, so a pod the StatefulSet already recreated between the + // observation and the delete is never restarted a second time. + UID string + + // Instance is the name this pod's Memgraph instance is known by in SHOW + // INSTANCES — instance_N for data pods, coordinator_N+1 for coordinators. + // The decision matches observations by this name, never by pod name. + Instance string + + // Ordinal is the pod's StatefulSet ordinal, which orders restarts within a + // role. + Ordinal int32 + + // RevisionHash is the pod's controller-revision-hash label, set by the + // StatefulSet controller. + RevisionHash string + + // Ready is the pod's Kubernetes readiness, which for these pods is a TCP + // connect to a port. It is necessary but never sufficient: an instance + // answering on its port has not necessarily rejoined replication. + Ready bool +} + +// Role is one StatefulSet's pods with the revision they are measured against. +type Role struct { + // Replicas is how many pods the role must have. A pod that has been deleted + // and not yet recreated is missing from Pods, and a role short of its + // replicas is never acted on. + Replicas int32 + + // UpdateRevision is the StatefulSet's status.updateRevision — the revision + // its current pod template hashes to. Empty while the StatefulSet has no + // status yet, which reads as nothing to do rather than as everything being + // outdated. + UpdateRevision string + + Pods []Pod +} + +// Decision is the one action a pass may take. +type Decision struct { + Action Action + + // Pod is the pod to restart, set only for Delete, and carried whole rather + // than by name. Its UID is what makes the delete conditional, and that UID has + // to be the one this decision was made against: a caller that re-read the pod + // by name to find it would get whichever pod exists by then — possibly the + // replacement — so the precondition would always match and guard nothing. + Pod Pod + + // Reason and Message describe a Wait or a Delete for the resource's + // condition. Done needs neither: the caller reports its own converged + // message. + Reason string + Message string +} + +// InProgress reports whether the role has pods still to restart. It is what +// lets the caller loosen its readiness gate by the one pod a restart took down, +// and only while one is actually under way. +func InProgress(role Role) bool { + return len(outdated(role)) > 0 +} + +// Next returns the single action to take toward both roles running their +// StatefulSets' current pod template. +// +// Data instances are dealt with first and completely; coordinators only once no +// data pod is outstanding *and* the data plane is whole again, so that at most +// one pod of the cluster is ever down — across both roles, not per role. A role +// whose pods all carry the current revision contributes nothing, which is why a +// cluster with nothing to roll returns Done regardless of how healthy it is: +// readiness is the Ready and Converged conditions' business, not this one's. +func Next( + data, coordinators Role, + observed []memgraph.Instance, + lag []memgraph.ReplicationLag, +) Decision { + instances := index(observed) + lags := indexLag(lag) + + if len(outdated(data)) > 0 { + return nextDataInstance(data, instances, lags) + } + if len(outdated(coordinators)) == 0 { + return Decision{Action: Done} + } + // A coordinator's pod does not go while a data pod is missing or unready: at + // most one pod of the cluster is down at a time, across both roles. + // + // Replication lag deliberately does not gate this. A coordinator restart + // neither reduces the number of instances holding recent writes nor forces a + // promotion, so a replica still draining its backlog is no reason to hold it — + // and making it one would let a single chronically lagging replica freeze the + // coordinators' pod template indefinitely. + if wait, ok := present(data); !ok { + return wait + } + return nextCoordinator(coordinators, instances) +} + +// nextDataInstance picks the next data pod to restart, or says what it is waiting +// for. Non-MAIN pods go first, highest ordinal down, matching the order a +// StatefulSet would have used; the MAIN goes last and alone. +func nextDataInstance( + data Role, + instances map[string]memgraph.Instance, + lags map[string]memgraph.ReplicationLag, +) Decision { + if wait, ok := present(data); !ok { + return wait + } + + // Where MAIN sits comes from the role Raft reports, not from health: an + // unreachable MAIN is still the cluster's MAIN, and restarting another pod + // while believing there is none is exactly the mistake to avoid. + main := mainInstance(instances) + if main == "" { + return waiting(memgraphcomv1alpha1.ReasonNoMainElected, + "Waiting for a MAIN data instance before restarting any data pod") + } + + pending := outdated(data) + if next, ok := highestOrdinalExcept(pending, main); ok { + // Taking another replica down reduces the number of instances holding + // recent writes, so the ones already restarted have to be back in + // replication first. This is the gate that waits out a fresh volume's full + // snapshot resync. + if wait, ok := replicating(data, instances, lags); !ok { + return wait + } + return restarting(next, fmt.Sprintf( + "Restarting data instance pod %s, which is not MAIN", next.Name)) + } + + // Only the MAIN is left. Its restart costs a failover, so it is the one step + // with a precondition of its own. + mainPod := pending[0] + if data.Replicas == 1 { + // A single data instance has no replica to fail over to and never will, + // so the precondition below can never be satisfied. Refusing would leave + // its pod template frozen forever, which protects nothing: there is no + // high availability here to preserve. + return restarting(mainPod, fmt.Sprintf( + "Restarting the only data instance pod %s, which interrupts the cluster until it is back", mainPod.Name)) + } + if !instances[main].IsUp() { + return waiting(memgraphcomv1alpha1.ReasonWorkloadsNotReady, fmt.Sprintf( + "Waiting for MAIN data instance %s to be reachable before restarting it", main)) + } + // A survivor that is reachable and holds every transaction the MAIN has + // committed. One is enough because the coordinators promote the most + // up-to-date instance they can reach, so whichever wins is at least as + // current as the one proven here. Instances observed down are deliberately + // not counted and deliberately not disqualifying: Raft cannot promote them, + // and a permanently sick replica must not freeze the cluster's pod template. + for _, pod := range data.Pods { + if pod.Instance == main { + continue + } + if instances[pod.Instance].IsUp() && lags[pod.Instance].IsCaughtUp() { + return restarting(mainPod, fmt.Sprintf( + "Restarting MAIN data instance pod %s last; %s is caught up and can be promoted in its place", + mainPod.Name, pod.Instance)) + } + } + return waiting(memgraphcomv1alpha1.ReasonNoCaughtUpSurvivor, fmt.Sprintf( + "Waiting for a data instance that is reachable and caught up with MAIN %s before restarting it; "+ + "the cluster keeps serving until one is", main)) +} + +// nextCoordinator picks the next coordinator pod to restart. Non-leaders go +// first, highest ordinal down, and the Raft leader last — its restart costs an +// election, which is harmless while the data plane has a MAIN, but there is no +// reason to pay it more than once. +func nextCoordinator(coordinators Role, instances map[string]memgraph.Instance) Decision { + if wait, ok := present(coordinators); !ok { + return wait + } + // A coordinator is proven back by the leader reaching it, which with three or + // more coordinators and one pod down at a time is the quorum question itself. + // Raft membership is no use here: it survives a pod restart untouched, so it + // never reads as absent. + for _, pod := range updated(coordinators) { + if !instances[pod.Instance].IsUp() { + return waiting(memgraphcomv1alpha1.ReasonWorkloadsNotReady, fmt.Sprintf( + "Waiting for restarted coordinator %s to be reachable before restarting the next one", pod.Instance)) + } + } + + leader := leaderInstance(instances) + if leader == "" { + return waiting(memgraphcomv1alpha1.ReasonNoCoordinatorLeader, + "Waiting for the coordinators to elect a leader before restarting any coordinator pod") + } + + pending := outdated(coordinators) + if next, ok := highestOrdinalExcept(pending, leader); ok { + return restarting(next, fmt.Sprintf( + "Restarting coordinator pod %s, which does not hold Raft leadership", next.Name)) + } + return restarting(pending[0], fmt.Sprintf( + "Restarting coordinator pod %s last; it holds Raft leadership, so the surviving members elect a successor", + pending[0].Name)) +} + +// replicating reports whether every data pod already carrying the current +// revision — which is exactly the set this roll has restarted — is back in +// replication: reachable by the coordinator leader, and holding every transaction +// the MAIN has committed. +// +// Pods still on the old revision are held to readiness alone on purpose. They have +// not been touched yet, so an instance that was already lagging before the roll +// began does not get to block it; the step that genuinely needs a caught-up +// instance asks for one directly, and asks for one rather than all. +// +// The restarted pods are not required to report role=replica, even though that is +// what they will normally be. A failover unrelated to the roll can move MAIN onto +// one of them, and demanding replica there would deadlock the roll against a +// perfectly healthy cluster. Reachable and caught up is the property that matters, +// and the MAIN reports itself caught up by definition. +func replicating( + role Role, + instances map[string]memgraph.Instance, + lags map[string]memgraph.ReplicationLag, +) (Decision, bool) { + for _, pod := range updated(role) { + if !instances[pod.Instance].IsUp() { + return waiting(memgraphcomv1alpha1.ReasonWorkloadsNotReady, fmt.Sprintf( + "Waiting for restarted data instance %s to be reachable before restarting the next pod", + pod.Instance)), false + } + if !lags[pod.Instance].IsCaughtUp() { + return waiting(memgraphcomv1alpha1.ReasonWaitingForCatchUp, fmt.Sprintf( + "Waiting for restarted data instance %s to catch up with MAIN before restarting the next pod", + pod.Instance)), false + } + } + return Decision{}, true +} + +// present reports whether every pod of the role exists and is ready — the pod a +// previous pass deleted included, which is what serialises the restarts down to +// one at a time. +func present(role Role) (Decision, bool) { + if int32(len(role.Pods)) != role.Replicas { + return waiting(memgraphcomv1alpha1.ReasonWorkloadsNotReady, + "Waiting for the pod restarted last to be recreated"), false + } + for _, pod := range role.Pods { + if !pod.Ready { + return waiting(memgraphcomv1alpha1.ReasonWorkloadsNotReady, fmt.Sprintf( + "Waiting for pod %s to become ready", pod.Name)), false + } + } + return Decision{}, true +} + +// outdated are the role's pods not carrying its current revision, which is the +// work left to do. +// +// A StatefulSet without a status yet has no revision to compare against, and a +// pod without the label cannot be classified; both read as up to date. Guessing +// the other way would delete pods on the strength of a missing value. +func outdated(role Role) []Pod { + if role.UpdateRevision == "" { + return nil + } + var pending []Pod + for _, pod := range role.Pods { + if pod.RevisionHash != "" && pod.RevisionHash != role.UpdateRevision { + pending = append(pending, pod) + } + } + return pending +} + +// updated are the role's pods already carrying its current revision — the ones +// this roll has restarted, once it is under way. +func updated(role Role) []Pod { + if role.UpdateRevision == "" { + return nil + } + var done []Pod + for _, pod := range role.Pods { + if pod.RevisionHash == role.UpdateRevision { + done = append(done, pod) + } + } + return done +} + +// highestOrdinalExcept is the outstanding pod with the highest ordinal that does +// not run the named instance. +// +// The exclusion is the point: it is how "the MAIN last" and "the Raft leader +// last" are expressed. So is reporting false — that says the named instance is +// the only pod left to restart, which is the step both callers guard with +// preconditions the earlier ones do not need. +// +// Taking the highest ordinal is only a convention. Any deterministic order would +// be correct; this is the one a StatefulSet's own rolling update uses, so the +// restart sequence looks familiar and the tests can assert on it. +func highestOrdinalExcept(pending []Pod, instance string) (Pod, bool) { + var next Pod + found := false + for _, pod := range pending { + if pod.Instance == instance { + continue + } + if !found || pod.Ordinal > next.Ordinal { + next, found = pod, true + } + } + return next, found +} + +// mainInstance is the data instance Raft reports as MAIN, regardless of whether +// the coordinator leader can currently reach it, or empty when none is reported. +func mainInstance(instances map[string]memgraph.Instance) string { + for name, instance := range instances { + if instance.IsMain() { + return name + } + } + return "" +} + +// leaderInstance is the coordinator reported as Raft leader, or empty when none +// is. +func leaderInstance(instances map[string]memgraph.Instance) string { + for name, instance := range instances { + if instance.IsLeader() { + return name + } + } + return "" +} + +// create a map: instanceName -> instance +func index(observed []memgraph.Instance) map[string]memgraph.Instance { + instances := make(map[string]memgraph.Instance, len(observed)) + for _, instance := range observed { + instances[instance.Name] = instance + } + return instances +} + +// indexLag keys replication lag by instance name. A name the view does not cover +// reads back as the zero value, which reports itself as not caught up — the safe +// answer for an instance nothing is known about, and the answer for every +// instance when there is no MAIN to measure against. +func indexLag(lag []memgraph.ReplicationLag) map[string]memgraph.ReplicationLag { + lags := make(map[string]memgraph.ReplicationLag, len(lag)) + for _, instance := range lag { + lags[instance.Instance] = instance + } + return lags +} + +func waiting(reason, message string) Decision { + return Decision{Action: Wait, Reason: reason, Message: message} +} + +func restarting(pod Pod, message string) Decision { + return Decision{ + Action: Delete, + Pod: pod, + Reason: memgraphcomv1alpha1.ReasonRollingRestartInProgress, + Message: message, + } +} diff --git a/internal/rollout/rollout_test.go b/internal/rollout/rollout_test.go new file mode 100644 index 0000000..949deca --- /dev/null +++ b/internal/rollout/rollout_test.go @@ -0,0 +1,504 @@ +/* +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 rollout + +import ( + "fmt" + "testing" + + memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/memgraph" +) + +const ( + oldRevision = "cluster-data-6c9f8b7d5" + newRevision = "cluster-data-77b4c8f9d" + + // The pods the cases below expect to be restarted, named once so a changed + // expectation cannot silently pass against a typo. + dataPod0 = "cluster-data-0" + dataPod1 = "cluster-data-1" + dataPod2 = "cluster-data-2" + coordinatorPod0 = "cluster-coordinator-0" + coordinatorPod1 = "cluster-coordinator-1" + coordinatorPod2 = "cluster-coordinator-2" +) + +// dataRole builds a data role of the given size whose pods carry the given +// revisions, one per ordinal, all ready. A revision equal to newRevision is a pod +// this roll has already restarted. +func dataRole(revisions ...string) Role { + role := Role{Replicas: int32(len(revisions)), UpdateRevision: newRevision} + for ordinal, revision := range revisions { + role.Pods = append(role.Pods, Pod{ + Name: fmt.Sprintf("cluster-data-%d", ordinal), + UID: fmt.Sprintf("uid-data-%d", ordinal), + Instance: fmt.Sprintf("instance_%d", ordinal), + Ordinal: int32(ordinal), + RevisionHash: revision, + Ready: true, + }) + } + return role +} + +// coordinatorRole is dataRole for the coordinator StatefulSet, whose instances +// are named from a 1-based Raft ID. +func coordinatorRole(revisions ...string) Role { + role := Role{Replicas: int32(len(revisions)), UpdateRevision: newRevision} + for ordinal, revision := range revisions { + role.Pods = append(role.Pods, Pod{ + Name: fmt.Sprintf("cluster-coordinator-%d", ordinal), + UID: fmt.Sprintf("uid-coordinator-%d", ordinal), + Instance: fmt.Sprintf("coordinator_%d", ordinal+1), + Ordinal: int32(ordinal), + RevisionHash: revision, + Ready: true, + }) + } + return role +} + +// converged is a role whose every pod already runs the current revision. +func converged(role Role) Role { + for i := range role.Pods { + role.Pods[i].RevisionHash = role.UpdateRevision + } + return role +} + +// cluster is a SHOW INSTANCES view: the named data instance is MAIN, every other +// declared one is a replica, the first coordinator leads, and everything is up. +func cluster(dataInstances, coordinators int, main string) []memgraph.Instance { + view := make([]memgraph.Instance, 0, dataInstances+coordinators) + for ordinal := range dataInstances { + name := fmt.Sprintf("instance_%d", ordinal) + role := memgraph.RoleReplica + if name == main { + role = memgraph.RoleMain + } + view = append(view, memgraph.Instance{Name: name, Health: memgraph.HealthUp, Role: role}) + } + for ordinal := range coordinators { + role := memgraph.RoleFollower + if ordinal == 0 { + role = memgraph.RoleLeader + } + view = append(view, memgraph.Instance{ + Name: fmt.Sprintf("coordinator_%d", ordinal+1), + BoltServer: "coordinator:7687", + Health: memgraph.HealthUp, + Role: role, + }) + } + return view +} + +// down marks the named instance as one the coordinator leader cannot reach, +// leaving the role it is registered with intact — which is what Memgraph reports +// for an instance whose pod is gone. +func down(view []memgraph.Instance, name string) []memgraph.Instance { + out := make([]memgraph.Instance, len(view)) + copy(out, view) + for i := range out { + if out[i].Name == name { + out[i].Health = "down" + } + } + return out +} + +// caughtUp is the replication lag view with every named instance holding all of +// the MAIN's transactions. +func caughtUp(names ...string) []memgraph.ReplicationLag { + lag := make([]memgraph.ReplicationLag, 0, len(names)) + for _, name := range names { + lag = append(lag, memgraph.ReplicationLag{ + Instance: name, + Databases: []memgraph.DatabaseLag{{Database: "memgraph", CommittedTxns: 42, TxnsBehindMain: 0}}, + }) + } + return lag +} + +// behind is caughtUp for an instance that is missing transactions. +func behind(name string) memgraph.ReplicationLag { + return memgraph.ReplicationLag{ + Instance: name, + Databases: []memgraph.DatabaseLag{{Database: "memgraph", CommittedTxns: 40, TxnsBehindMain: 2}}, + } +} + +func TestNothingToRestart(t *testing.T) { + data := converged(dataRole(newRevision, newRevision, newRevision)) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, + cluster(3, 3, "instance_0"), caughtUp("instance_0", "instance_1", "instance_2")) + + if decision.Action != Done { + t.Fatalf("expected Done for a cluster already on the current revision, got %+v", decision) + } +} + +// A StatefulSet with no status yet has no revision to measure pods against. +// Reading that as "every pod is outdated" would delete pods on the strength of a +// missing value. +func TestRoleWithoutRevisionIsLeftAlone(t *testing.T) { + data := dataRole(oldRevision, oldRevision) + data.UpdateRevision = "" + coordinators := coordinatorRole(oldRevision, oldRevision, oldRevision) + coordinators.UpdateRevision = "" + + if decision := Next(data, coordinators, cluster(2, 3, "instance_0"), nil); decision.Action != Done { + t.Fatalf("expected Done while no revision is known, got %+v", decision) + } + if InProgress(data) { + t.Error("a role without an update revision has no restart in progress") + } +} + +func TestDataInstancesRestartHighestOrdinalFirstAndSkipMain(t *testing.T) { + // MAIN sits in the middle on purpose: a StatefulSet's own rolling update + // would take instance_2, then instance_1 — the MAIN — then instance_0. + data := dataRole(oldRevision, oldRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + view := cluster(3, 3, "instance_1") + lag := caughtUp("instance_0", "instance_1", "instance_2") + + decision := Next(data, coordinators, view, lag) + if decision.Action != Delete || decision.Pod.Name != dataPod2 { + t.Fatalf("expected the highest non-MAIN ordinal first, got %+v", decision) + } + + // instance_2 restarted and caught up; instance_1 is MAIN, so instance_0 is next. + data.Pods[2].RevisionHash = newRevision + decision = Next(data, coordinators, view, lag) + if decision.Action != Delete || decision.Pod.Name != dataPod0 { + t.Fatalf("expected MAIN to be skipped for the lower ordinal, got %+v", decision) + } + + // Only the MAIN is left. + data.Pods[0].RevisionHash = newRevision + decision = Next(data, coordinators, view, lag) + if decision.Action != Delete || decision.Pod.Name != dataPod1 { + t.Fatalf("expected the MAIN's pod last, got %+v", decision) + } + if decision.Pod.UID != "uid-data-1" { + t.Errorf("expected the observed UID to be carried for a conditional delete, got %q", decision.Pod.UID) + } +} + +func TestRestartedInstanceMustBeReadyBeforeTheNextGoes(t *testing.T) { + data := dataRole(oldRevision, oldRevision, newRevision) + data.Pods[2].Ready = false + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, cluster(3, 3, "instance_0"), + caughtUp("instance_0", "instance_1", "instance_2")) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWorkloadsNotReady { + t.Fatalf("expected to wait on the unready pod, got %+v", decision) + } +} + +// A pod deleted and not yet recreated is absent from the role, and the restart +// waits for it rather than treating one fewer pod as one fewer thing to check. +func TestMissingPodStopsTheRoll(t *testing.T) { + data := dataRole(oldRevision, oldRevision, newRevision) + data.Pods = data.Pods[:2] + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, cluster(3, 3, "instance_0"), caughtUp("instance_0", "instance_1")) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWorkloadsNotReady { + t.Fatalf("expected to wait for the deleted pod to be recreated, got %+v", decision) + } +} + +func TestRestartedInstanceMustBeReachableAndCaughtUp(t *testing.T) { + data := dataRole(oldRevision, oldRevision, newRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + view := cluster(3, 3, "instance_0") + + // Ready, but the coordinator leader does not reach it yet. + decision := Next(data, coordinators, down(view, "instance_2"), caughtUp("instance_0", "instance_2")) + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWorkloadsNotReady { + t.Fatalf("expected to wait for the restarted instance to be reachable, got %+v", decision) + } + + // Reachable, but still draining its backlog — the fresh-volume resync case. + decision = Next(data, coordinators, view, append(caughtUp("instance_0"), behind("instance_2"))) + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWaitingForCatchUp { + t.Fatalf("expected to wait for the restarted instance to catch up, got %+v", decision) + } + + // An empty lag view means nothing is known, which is not permission to proceed. + decision = Next(data, coordinators, view, nil) + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWaitingForCatchUp { + t.Fatalf("expected an unknown lag to read as not caught up, got %+v", decision) + } +} + +// An instance that was already lagging before the roll began must not block it: +// it has not been restarted, so it is held to readiness alone. The one step that +// genuinely needs a caught-up instance asks for one directly. +func TestNotYetRestartedInstanceIsNotHeldToLag(t *testing.T) { + data := dataRole(oldRevision, oldRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, cluster(3, 3, "instance_0"), + append(caughtUp("instance_0"), behind("instance_1"))) + + if decision.Action != Delete || decision.Pod.Name != dataPod2 { + t.Fatalf("expected a lagging untouched instance not to block the roll, got %+v", decision) + } +} + +func TestMainIsNotRestartedWithoutACaughtUpSurvivor(t *testing.T) { + data := dataRole(newRevision, newRevision, oldRevision) + data.Pods[2].Instance = "instance_2" + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + view := cluster(3, 3, "instance_2") + + // Both survivors are behind: the cluster keeps serving and the roll parks. + decision := Next(data, coordinators, view, + []memgraph.ReplicationLag{behind("instance_0"), behind("instance_1")}) + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonNoCaughtUpSurvivor { + t.Fatalf("expected to park at the MAIN without a caught-up survivor, got %+v", decision) + } + + // One caught-up survivor is enough: the coordinators promote the most + // up-to-date instance they can reach, so whoever wins is at least as current. + decision = Next(data, coordinators, view, append(caughtUp("instance_1"), behind("instance_0"))) + if decision.Action != Delete || decision.Pod.Name != dataPod2 { + t.Fatalf("expected one caught-up survivor to permit the MAIN's restart, got %+v", decision) + } +} + +// A caught-up instance the leader cannot reach is not a promotion candidate: Raft +// cannot promote what it cannot see. Its registration outliving its reachability +// is exactly why health and lag are both asked, and why lag alone is never enough. +func TestUnreachableSurvivorIsNoSurvivor(t *testing.T) { + data := dataRole(newRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + view := down(cluster(2, 3, "instance_1"), "instance_0") + + decision := Next(data, coordinators, view, caughtUp("instance_0", "instance_1")) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonNoCaughtUpSurvivor { + t.Fatalf("expected the MAIN to park without a reachable survivor, got %+v", decision) + } +} + +// One chronically lagging replica must not be able to freeze the cluster's pod +// template. The MAIN's restart needs one survivor Raft could promote, not every +// survivor, so a replica that never catches up does not block it. +func TestOneLaggingReplicaDoesNotBlockTheMain(t *testing.T) { + data := dataRole(newRevision, newRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, cluster(3, 3, "instance_2"), + append(caughtUp("instance_1"), behind("instance_0"))) + + if decision.Action != Delete || decision.Pod.Name != dataPod2 { + t.Fatalf("expected one caught-up survivor to be enough despite a lagging one, got %+v", decision) + } +} + +// A single data instance has no replica and never will, so the MAIN's +// precondition can never be met. Refusing would freeze its pod template forever +// and protect nothing. +func TestSingleDataInstanceIsRestartedWithAcknowledgedDowntime(t *testing.T) { + data := dataRole(oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, cluster(1, 3, "instance_0"), caughtUp("instance_0")) + + if decision.Action != Delete || decision.Pod.Name != dataPod0 { + t.Fatalf("expected the only data instance to be restarted, got %+v", decision) + } + if !containsAll(decision.Message, "only data instance", "interrupts") { + t.Errorf("expected the message to name the interruption, got %q", decision.Message) + } +} + +// Without a MAIN there is nothing to measure lag against and no telling what the +// cluster is doing, so no data pod is taken down. +func TestNoMainStopsTheDataRoll(t *testing.T) { + data := dataRole(oldRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + view := cluster(2, 3, "") + + decision := Next(data, coordinators, view, nil) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonNoMainElected { + t.Fatalf("expected to wait for a MAIN before restarting data pods, got %+v", decision) + } +} + +// An unreachable MAIN keeps its role in Raft, so it is still found — but it is not +// restarted while the cluster cannot serve from it. +func TestUnreachableMainIsNotRestarted(t *testing.T) { + data := dataRole(newRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + view := down(cluster(2, 3, "instance_1"), "instance_1") + + decision := Next(data, coordinators, view, caughtUp("instance_0")) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWorkloadsNotReady { + t.Fatalf("expected an unreachable MAIN not to be restarted, got %+v", decision) + } +} + +func TestCoordinatorsWaitForEveryDataPod(t *testing.T) { + data := converged(dataRole(newRevision, newRevision)) + coordinators := coordinatorRole(oldRevision, oldRevision, oldRevision) + lag := caughtUp("instance_0", "instance_1") + + // A data pod is still coming back. One pod of the cluster is down at a time + // across both roles, so no coordinator goes on top of it. + unready := converged(dataRole(newRevision, newRevision)) + unready.Pods[1].Ready = false + decision := Next(unready, coordinators, cluster(2, 3, "instance_0"), lag) + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWorkloadsNotReady { + t.Fatalf("expected coordinators to wait for every data pod to be ready, got %+v", decision) + } + + // Replication lag, by contrast, does not gate a coordinator restart: it neither + // reduces the instances holding recent writes nor forces a promotion. Making it + // a gate would let one lagging replica freeze the coordinators' template. + decision = Next(data, coordinators, cluster(2, 3, "instance_0"), + append(caughtUp("instance_0"), behind("instance_1"))) + if decision.Action != Delete || decision.Pod.Name != coordinatorPod2 { + t.Fatalf("expected a lagging replica not to block the coordinator roll, got %+v", decision) + } + + // Healthy: the coordinator roll starts, highest ordinal first, leader excluded. + decision = Next(data, coordinators, cluster(2, 3, "instance_0"), lag) + if decision.Action != Delete || decision.Pod.Name != coordinatorPod2 { + t.Fatalf("expected the highest non-leader coordinator first, got %+v", decision) + } +} + +func TestCoordinatorLeaderIsRestartedLast(t *testing.T) { + data := converged(dataRole(newRevision, newRevision)) + coordinators := coordinatorRole(oldRevision, oldRevision, oldRevision) + view := cluster(2, 3, "instance_0") + lag := caughtUp("instance_0", "instance_1") + + // coordinator_1, on ordinal 0, is the leader. + coordinators.Pods[2].RevisionHash = newRevision + decision := Next(data, coordinators, view, lag) + if decision.Action != Delete || decision.Pod.Name != coordinatorPod1 { + t.Fatalf("expected the leader to be skipped, got %+v", decision) + } + + coordinators.Pods[1].RevisionHash = newRevision + decision = Next(data, coordinators, view, lag) + if decision.Action != Delete || decision.Pod.Name != coordinatorPod0 { + t.Fatalf("expected the leader's pod last, got %+v", decision) + } +} + +// Raft membership survives a pod restart untouched, so reachability is what proves +// a coordinator is back — and it is asked before the next one goes. +func TestRestartedCoordinatorMustBeReachable(t *testing.T) { + data := converged(dataRole(newRevision, newRevision)) + coordinators := coordinatorRole(oldRevision, oldRevision, newRevision) + view := down(cluster(2, 3, "instance_0"), "coordinator_3") + + decision := Next(data, coordinators, view, caughtUp("instance_0", "instance_1")) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonWorkloadsNotReady { + t.Fatalf("expected to wait for the restarted coordinator, got %+v", decision) + } +} + +func TestNoCoordinatorLeaderStopsTheCoordinatorRoll(t *testing.T) { + data := converged(dataRole(newRevision, newRevision)) + coordinators := coordinatorRole(oldRevision, oldRevision, oldRevision) + view := cluster(2, 0, "instance_0") + for ordinal := range 3 { + view = append(view, memgraph.Instance{ + Name: fmt.Sprintf("coordinator_%d", ordinal+1), + BoltServer: "coordinator:7687", + Health: memgraph.HealthUp, + Role: memgraph.RoleFollower, + }) + } + + decision := Next(data, coordinators, view, caughtUp("instance_0", "instance_1")) + + if decision.Action != Wait || decision.Reason != memgraphcomv1alpha1.ReasonNoCoordinatorLeader { + t.Fatalf("expected to wait for a Raft leader, got %+v", decision) + } +} + +// A spec reverted halfway through inverts which pods are outdated, and the roll +// walks back with nothing to unwind — the point of deriving the state every pass +// rather than tracking it. +func TestRevertedSpecRollsBack(t *testing.T) { + data := dataRole(newRevision, newRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + // The spec goes back: the old revision is now the current one, so the two pods + // already restarted are the outdated ones. + data.UpdateRevision = oldRevision + coordinators.UpdateRevision = oldRevision + for i := range coordinators.Pods { + coordinators.Pods[i].RevisionHash = oldRevision + } + + decision := Next(data, coordinators, cluster(3, 3, "instance_0"), + caughtUp("instance_0", "instance_1", "instance_2")) + + if decision.Action != Delete || decision.Pod.Name != dataPod1 { + t.Fatalf("expected the roll to reverse onto the highest re-outdated non-MAIN pod, got %+v", decision) + } +} + +// A failover unrelated to the roll can move MAIN onto a pod already restarted. +// Demanding role=replica of the restarted pods would deadlock against a perfectly +// healthy cluster, so reachable and caught up is what is asked. +func TestMainMovingOntoARestartedPodDoesNotDeadlock(t *testing.T) { + data := dataRole(newRevision, oldRevision, oldRevision) + coordinators := converged(coordinatorRole(newRevision, newRevision, newRevision)) + + decision := Next(data, coordinators, cluster(3, 3, "instance_0"), + caughtUp("instance_0", "instance_1", "instance_2")) + + if decision.Action != Delete || decision.Pod.Name != dataPod2 { + t.Fatalf("expected the roll to continue with MAIN on a restarted pod, got %+v", decision) + } +} + +func containsAll(s string, substrings ...string) bool { + for _, substring := range substrings { + found := false + for i := 0; i+len(substring) <= len(s); i++ { + if s[i:i+len(substring)] == substring { + found = true + break + } + } + if !found { + return false + } + } + return true +} diff --git a/specs/operator-mvp/issues/17-sequenced-rolling-restart.md b/specs/operator-mvp/issues/17-sequenced-rolling-restart.md new file mode 100644 index 0000000..57590dc --- /dev/null +++ b/specs/operator-mvp/issues/17-sequenced-rolling-restart.md @@ -0,0 +1,60 @@ +# Sequenced rolling restart: no-downtime upgrades + +**Type**: AFK + +## Parent + +`specs/operator-mvp/PRD.md` + +## What to build + +Replace Kubernetes' own pod replacement for both roles with a sequence the operator drives, so that a changed pod template — a new Memgraph image above all, but equally a resource limit, an env var or a probe timing — is rolled through the cluster without ever taking down an instance the cluster still needs. The trigger is deliberately not "the image changed": it is `pod.metadata.labels["controller-revision-hash"] != sts.Status.UpdateRevision`, because `updateStrategy: OnDelete` means nothing but the operator will ever restart a pod again, and a template change nobody rolls is a cluster frozen on an old spec with nothing reporting it. What this issue builds is therefore a *sequenced rolling restart*; upgrades are its reason to exist, not its scope. + +The default `RollingUpdate` cannot express the required order and no amount of `partition` fixes it. Kubernetes sweeps highest ordinal to lowest, and `partition` is a descending cutoff rather than a set, so a MAIN sitting on any ordinal but 0 is restarted somewhere in the middle of the sweep. Each such restart costs a failover, and the instance Raft promotes is arbitrary with respect to the sweep, so a five-instance cluster can pay up to four write outages for one upgrade. `PodManagementPolicy: ParallelPodManagement` (`statefulset.go:393`) does not enter into it — it governs creation and scaling, never updates. So both role StatefulSets move to `OnDelete` and the operator owns every pod deletion from then on, permanently. That is the real cost of this issue and it is accepted knowingly: if the operator is down, no pod template ever takes effect, and nothing in Kubernetes will say so. + +**This feature is unsafe on any Memgraph release that reports `role=unknown` for a data instance the coordinator leader cannot reach.** Deleting the MAIN's pod vacates its `main` row, `hasMain` (`planner.go:315-324`) goes false, and `Plan` takes the branch at `planner.go:354` to emit `SetInstanceToMain{promotionTarget(...)}` — the operator racing the coordinators' own failover on every single upgrade, which is exactly what the PRD forbids in user story 6 and the promotion rule at line 84. The prerequisite is the core change that makes a dead MAIN report `role=main, health=down`, keeping `hasMain` true so the planner stays out of the way. The PRD rules out detecting this at runtime ("no version parsing, gating, or branching"), and no `DefaultImageTag` floor is introduced here: which operator version is safe against which Memgraph release is documented outside this repository. + +The order is data plane first, then coordinators, and it is re-derived from observation on every pass rather than tracked in status. Among data instances: delete any outdated pod that is not the observed MAIN, highest ordinal first to match the StatefulSet convention, and only when the MAIN is the last outdated pod left is it deleted. Among coordinators: delete any outdated pod that is not the observed leader, and the leader last. Deriving both rules each pass is what makes leadership moving spontaneously mid-roll, or a spec reverted halfway through, self-correcting with no state to unwind — a revert simply inverts which pods are outdated and the roll walks back. Killing the coordinator leader costs the data plane nothing: per `15-data-instance-scale-down.md`, failover needs a leadership change *with zero MAINs* or a MAIN ping failure, and the data plane is whole throughout the coordinator roll. The operator's own connection dies with the pod it deletes, which is a requeue, not an error. + +Exactly one pod is down at a time, and what allows the next one to go is not pod readiness. Probes are TCP-socket checks against the bolt port (`statefulset.go:464`), so `Ready` says the port is open and nothing about the instance having rejoined replication. A restarted data instance needs no `REGISTER INSTANCE` — registration lives in the coordinators' Raft log, which is why `wipeInstanceRegistration` (`memgraphcluster_test.go:826`) simulates its loss with `UNREGISTER INSTANCE` rather than by touching a volume — but it does need the coordinator to re-attach it, and on a fresh volume it needs a full snapshot resync. So the per-step gate is: pod `Ready`, plus the leader observing it `up` with `role=replica`, plus `SHOW REPLICATION LAG` reporting it caught up through the existing `ReplicationLag.IsCaughtUp` (`client.go:126`). Waiting out a resync is the point, not a regression: it is what keeps recent writes on more than one machine for the whole roll. For coordinators the gate is leader-reported `health=up`, which with `coordinators >= 3` and one pod down at a time *is* the quorum question. `coordinatorRegistered` (`planner.go:484`) cannot serve here — it tests Raft membership, which a pod restart leaves untouched, so it never goes false. + +The MAIN's pod is deleted with no write fence, and the coordinators promote. The precondition is one fresh check: at least one data instance the leader observes `up`, other than the MAIN, is reported caught up. That is sufficient only because core promotes the most up-to-date alive instance — whoever wins is then at least as current as the instance that was verified, which also closes the case of a stale replica returning inside the failover window. `SET COORDINATOR SETTING "global_read_only"` was considered and rejected: the residual exposure without it is transactions committed after a caught-up SYNC replica silently fell behind, inside the few hundred milliseconds between the check and the deletion — the exposure an ordinary MAIN crash already carries, which registering replicas SYNC rather than `STRICT_SYNC` already accepts, and this path at least gets a caught-up check that a crash never does. The latch, by contrast, is cluster-wide and persisted in Raft, and the promoted MAIN inherits it, so writes would resume not when Raft promotes but when the operator next reconciles and clears it — trading an outage bounded by core for a longer one bounded by the operator, which strands the cluster read-only indefinitely if the operator dies in that window. `DEMOTE INSTANCE` as a self-clearing fence was rejected for the same reason it cannot be improved: `SHOW REPLICATION LAG` is served by the MAIN, so lag cannot be measured after a demote (`planner.go:124-126`), and the gain over doing nothing is one query's width. + +The decision itself lives in a new pure package, `internal/rollout` — an eighth module alongside the resource builders and the registration planner. It takes both roles' pods reduced to `{name, revisionHash, ready}`, each StatefulSet's `UpdateRevision`, the observed `[]memgraph.Instance` and `[]memgraph.ReplicationLag`, and returns exactly **one** action: `Done`, `Wait(reason)`, or `Delete(pod)`. Both roles go in together so the data-before-coordinators ordering and the MAIN-last and leader-last rules are unit-testable rather than living in controller flow. One action per pass and never a list, because every step must be re-gated on fresh observation and a precomputed list would act on stale lag; after a `Delete` the controller returns and requeues rather than deciding again against a cache that still shows the pod `Ready`. + +Wiring it needs the readiness gate loosened by exactly the absence the operator causes and no more. `reconcileRegistration` returns at `controller.go:327-338` before ever connecting to a coordinator, so a naive wiring deletes one pod and then never observes anything again — while the decision to proceed needs a live `SHOW INSTANCES` and `SHOW REPLICATION LAG` with one data pod deliberately down. `workloadsReady` (`controller.go:641`) therefore tolerates one *existing but unready* pod in a role, and only when that role is mid-roll (some pod's revision differs from `UpdateRevision`) **and** all of its pods exist (`sts.Status.Replicas == role.applied`). Both conditions are load-bearing: a blanket "tolerate one unready pod" would reintroduce precisely the informer-lag bug that function's doc comment warns about, where a 3→4 scale-up's stale `readyReplicas=3` against `applied=4` reads as ready and registration runs against a pod the API server has not been asked to create. The registration planner keeps running throughout a roll, on the same leader connection and the same observation; it should return empty on every pass, and if it ever does not, that is real drift — a coordinator that lost its Raft state, say — which the cluster wants repaired now rather than after a resync that can take hours. Scale converges first and pauses the roll: no roll step while the plan is non-empty, `planner.Retired` is false, or applied counts differ from spec, so the retirement's own MAIN handover (`DEMOTE INSTANCE` plus `SetInstanceToMain`) is never moving MAIN at the same time as this is. + +Observability follows the house style: one new steady-state condition `Updated`, True when every pod of both roles sits at its StatefulSet's `UpdateRevision` and False carrying the `rollout.Wait` reason otherwise, with new reasons `RollingRestartInProgress` and `WaitingForCatchUp` and reuse of `NoCaughtUpSurvivor` (`types.go:199`) for the stall and `WorkloadsNotReady` for a pod on its way back. `Converged` keeps meaning what it means; folding a roll into it would leave a user unable to tell registration drift from an upgrade merely walking. No progress counters are mirrored into the CR — `status.updatedReplicas`, `currentRevision` and `updateRevision` already carry them on the StatefulSets — and no Events, since this operator has no recorder and communicates through conditions alone. The core prerequisite forces one correction here: `observedMain` (`controller.go:547`) must require `IsMain() && IsUp()`, or a dead MAIN keeps `Ready=True` through the whole failover window while `readyOrNot`'s comment claims the cluster serves writes. `Ready` consequently goes False briefly during every upgrade, which is correct and is the one thing the docs must say plainly. + +Two pieces of collateral. Pod-level reads are unavoidable — `sts.Status.updatedReplicas` gives a count, never which pods — so the manager gains a Pod informer scoped by label selector to `app.kubernetes.io/managed-by=memgraph-operator` (`resources.go:73`, present on every pod the builders emit) in `cmd/main.go:158`, bounding its cache to this operator's own pods instead of every pod in the cluster; and a new `+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;delete` marker joins `controller.go:89-93`, widening the ClusterRole users install with **delete on pods** — worth naming in the chart's release notes. `Owns(&appsv1.StatefulSet{})` already wakes the controller when `readyReplicas` and `updatedReplicas` move, so a roll gets event-driven wakeups with `requeueWhilePending` as the backstop. Separately, neither role sets `terminationGracePeriodSeconds` today, so every pod gets the Kubernetes default of 30 seconds before SIGKILL — harmless when nothing routinely deleted pods, wrong once the operator deletes every one of them on every template change, because an instance killed mid-shutdown recovers from WAL on startup and lengthens exactly the catch-up wait the roll blocks on. Both roles get `terminationGracePeriodSeconds: 300`, a ceiling and not a delay, with no new spec knob until someone asks for one. + +Two edges to get right rather than discover. `dataInstances: 1` is legal, has no replica, and can therefore never satisfy the MAIN-step precondition — so under `OnDelete` its pod would never be restarted by anything, ever. It gets a plain restart with a brief full outage, reported as such, because there is no high availability there to protect and freezing the template permanently protects nothing. The `dataInstances >= 2` stall is the opposite call: if no replica is ever caught up, the roll parks at the MAIN indefinitely and reports why, and never escalates to deleting the MAIN's pod anyway — the cluster is still fully serving, so waiting costs only the upgrade. Under data-first ordering that also freezes the coordinator roll for as long as the stall lasts, which is the right trade (one upgrade half-applied across two roles is worse) but must be stated in the condition and the docs, because one sick replica holding up the whole cluster's template looks like a hang otherwise. + +Deliberately out of scope. A `PodDisruptionBudget` is not built here: a node drain can still take the MAIN and a replica together, but that gap predates this issue and is unrelated to operator-driven rolls, so it belongs to its own. Proving the durability property — that no acked write is lost across a roll — is not attempted in CI either: it needs a write workload running through the whole sequence, and the chaos-testing project the PRD names as the proving ground is where that belongs. What CI asserts is the sequencing the operator actually owns. + +## Acceptance criteria + +- [ ] Both role StatefulSets use `updateStrategy: OnDelete`; a changed pod template restarts no pod until the operator deletes it +- [ ] Any pod-template change triggers a roll, not only an image change; builder tests cover a resource, env and probe edit producing the same outcome as an image edit +- [ ] Data instances roll before coordinators; at most one pod of the cluster is down at any point in a roll +- [ ] Among data instances, the observed MAIN is deleted last; among coordinators, the observed leader is deleted last +- [ ] The next pod is deleted only after the previous one is `Ready`, observed `up` with `role=replica`, and reported caught up by `SHOW REPLICATION LAG`; a coordinator only after it is observed `health=up` +- [ ] The MAIN's pod is deleted only while at least one other data instance is observed `up` and caught up, re-checked on the same pass as the deletion +- [ ] No `SET INSTANCE TO MAIN` is ever issued as part of a roll; planner unit test asserts `Plan` emits no promotion for a non-retiring MAIN observed `role=main, health=down` +- [ ] `dataInstances: 1` gets a plain restart, and the condition names the downtime while it happens +- [ ] A cluster with no caught-up replica parks at the MAIN indefinitely, reports `NoCaughtUpSurvivor`, and never deletes the MAIN's pod +- [ ] `internal/rollout` is pure — pods, revisions, instances and lag in, one `Done`/`Wait`/`Delete` action out — with unit tests for role ordering, MAIN-last, leader-last, mid-roll spec revert, and every wait reason +- [ ] `workloadsReady` tolerates one existing-but-unready pod only while that role is mid-roll and all its pods exist; a scale-up mid-roll still reads as not ready +- [ ] No roll step is taken while the registration plan is non-empty, `planner.Retired` is false, or applied counts differ from spec +- [ ] The registration planner runs on every pass of a roll, against the same observation and leader connection +- [ ] `Updated` is True only when every pod of both roles is at `UpdateRevision`, and False with the reason the rollout decision returned +- [ ] `observedMain` requires `IsMain() && IsUp()`, so `Ready` is False while the MAIN is down; envtest covers a down MAIN +- [ ] Both roles set `terminationGracePeriodSeconds: 300`; builder tests cover it +- [ ] The manager's Pod cache is restricted to `app.kubernetes.io/managed-by=memgraph-operator`; the pods RBAC marker is added and `make manifests chart-sync` regenerated with `make chart-verify` green +- [ ] E2E: a benign pod-template change on the existing cluster rolls every data pod then every coordinator pod, deleting the MAIN last and the leader last, with never more than one pod down, ending at `Updated=True` and one MAIN — ordering captured by polling pod UIDs during the roll +- [ ] Docs state the honest contract: `Ready` goes False for the failover window on every upgrade, single-instance clusters take downtime, and a roll waits on replica catch-up + +## Blocked by + +- `16-coordinator-scale-down.md` +- Core: `SHOW INSTANCES` reporting `role=main, health=down` for an unreachable MAIN instead of `role=unknown` (memgraph/memgraph) — this issue must not merge ahead of a Memgraph release carrying it diff --git a/test/e2e/memgraphcluster_test.go b/test/e2e/memgraphcluster_test.go index 08e5813..0368471 100644 --- a/test/e2e/memgraphcluster_test.go +++ b/test/e2e/memgraphcluster_test.go @@ -29,6 +29,7 @@ import ( "os/exec" "path/filepath" "slices" + "sort" "strings" "time" @@ -37,6 +38,7 @@ import ( "sigs.k8s.io/yaml" memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/resources" "github.com/memgraph/kubernetes-operator/test/utils" ) @@ -147,10 +149,10 @@ func loadExample() *memgraphcomv1alpha1.MemgraphCluster { func (c clusterUnderTest) declaredInstances() []string { names := make([]string, 0, c.coordinators+c.dataInstances) for ordinal := range c.coordinators { - names = append(names, utils.CoordinatorName(ordinal)) + names = append(names, resources.CoordinatorInstanceName(ordinal)) } for ordinal := range c.dataInstances { - names = append(names, utils.DataInstanceName(ordinal)) + names = append(names, resources.DataInstanceName(ordinal)) } return names } @@ -252,6 +254,78 @@ var _ = Describe("MemgraphCluster", Ordered, func() { // Storage survives the cluster under the default retention policy: an // accidental `kubectl delete mgc` must not take a production database with // it. This deletes the CR, so it runs last in this Ordered container. + // The sequenced rolling restart, on the converged cluster the preceding specs + // left behind (Ordered). Both StatefulSets use updateStrategy OnDelete, so + // Kubernetes replaces nothing on its own: every pod here moves because the + // operator deleted it, in an order it chose. + // + // The trigger is a benign pod-template edit rather than an image bump. The + // operator has no version logic at all, so what is under test is the ordering, + // and an extra environment variable exercises the identical revision change + // without a second image pull or Memgraph version skew in the way. + // + // What is deliberately not asserted here: that no acknowledged write is lost + // across the roll. That needs a write workload running through the whole + // sequence, and it belongs to the chaos-testing project rather than a spec that + // gates every pull request. + It("rolls a changed pod template through the data instances before the coordinators", func() { + By("confirming the cluster is converged before changing the pod template") + Eventually(quickstartCluster.verifyRegistered, 10*time.Minute, 10*time.Second).Should(Succeed()) + quickstartCluster.awaitConverged(2 * time.Minute) + + By("recording which pods exist and which instance is MAIN") + before, err := quickstartCluster.podUIDs() + Expect(err).NotTo(HaveOccurred()) + Expect(before).To(HaveLen(int(quickstartCluster.coordinators + quickstartCluster.dataInstances))) + + view, err := quickstartCluster.leaderView() + Expect(err).NotTo(HaveOccurred()) + main := mainOf(view) + Expect(main).NotTo(BeEmpty(), "the roll's order is defined against the MAIN") + mainOrdinal, err := resources.DataInstanceOrdinal(main) + Expect(err).NotTo(HaveOccurred()) + mainPod := fmt.Sprintf("%s-data-%d", quickstartCluster.name, mainOrdinal) + + By("adding an environment variable to both roles, which changes the pod template") + cmd := exec.Command("kubectl", "patch", "memgraphcluster", quickstartCluster.name, + "-n", quickstartCluster.namespace, "--type=merge", "-p", + `{"spec":{"extraEnv":{`+ + `"coordinators":[{"name":"E2E_ROLLING_RESTART","value":"1"}],`+ + `"data":[{"name":"E2E_ROLLING_RESTART","value":"1"}]}}}`) + _, err = utils.Run(cmd) + Expect(err).NotTo(HaveOccurred(), "the operator must accept an extra environment variable") + + By("watching the operator replace every pod, one at a time") + order, maxDown := quickstartCluster.watchRoll(before, 25*time.Minute) + + Expect(order).To(HaveLen(len(before)), "every pod must be replaced exactly once") + Expect(maxDown).To(BeNumerically("<=", 1), + "at most one pod of the cluster may be unready at a time; observed %d", maxDown) + + By("confirming data instances went before coordinators, and MAIN last of its role") + var dataOrder, coordinatorOrder []string + for _, pod := range order { + if strings.Contains(pod, "-data-") { + dataOrder = append(dataOrder, pod) + continue + } + coordinatorOrder = append(coordinatorOrder, pod) + Expect(dataOrder).To(HaveLen(int(quickstartCluster.dataInstances)), + "a coordinator pod (%s) was restarted before the data plane finished: %v", pod, order) + } + Expect(dataOrder).To(HaveLen(int(quickstartCluster.dataInstances))) + Expect(coordinatorOrder).To(HaveLen(int(quickstartCluster.coordinators))) + Expect(dataOrder[len(dataOrder)-1]).To(Equal(mainPod), + "the MAIN's pod must be the last data pod restarted, order was %v", dataOrder) + + By("confirming the cluster converges with the new template and one MAIN") + Eventually(quickstartCluster.verifyRegistered, 10*time.Minute, 10*time.Second).Should(Succeed()) + cmd = exec.Command("kubectl", "wait", "--for=condition=Updated", + "memgraphcluster/"+quickstartCluster.name, "-n", quickstartCluster.namespace, "--timeout=5m") + _, err = utils.Run(cmd) + Expect(err).NotTo(HaveOccurred(), "the MemgraphCluster never reported Updated") + }) + It("leaves the PVCs behind when the default-retention CR is deleted", func() { By("confirming the cluster is converged before deleting it") Eventually(quickstartCluster.verifyRegistered, 10*time.Minute, 10*time.Second).Should(Succeed()) @@ -456,7 +530,7 @@ spec: AfterAll(func() { By("removing the cluster namespace and waiting for its pods to go") cmd := exec.Command("kubectl", "delete", "ns", scalingNamespace, - "--ignore-not-found", "--wait=true", "--timeout=5m") + "--ignore-not-found", "--wait=true", "--timeout=10m") _, _ = utils.Run(cmd) }) @@ -548,10 +622,13 @@ spec: }, 10*time.Minute, 5*time.Second).Should(Succeed()) By("waiting for its pod to be shed") + // Comfortably above terminationGracePeriodSeconds: a pod that needs its full + // shutdown budget takes five minutes to go, so a five-minute timeout here + // would be a coin flip rather than an assertion. Eventually(func(g Gomega) { g.Expect(shrunk.replicas("data")).To(Equal("2")) g.Expect(shrunk.podExists("data", 2)).To(BeFalse()) - }, 5*time.Minute, 5*time.Second).Should(Succeed()) + }, 8*time.Minute, 5*time.Second).Should(Succeed()) By("confirming the shrunk cluster is registered, converged, and led by a survivor") Eventually(shrunk.verifyRegistered, 10*time.Minute, 10*time.Second).Should(Succeed()) @@ -632,11 +709,13 @@ spec: }, 10*time.Minute, 5*time.Second).Should(Succeed()) By("waiting for their pods to be shed") + // Above terminationGracePeriodSeconds, for the reason the data-instance + // shrink's own shed assertion is. Eventually(func(g Gomega) { g.Expect(shrunk.replicas("coordinator")).To(Equal("3")) g.Expect(shrunk.podExists("coordinator", 3)).To(BeFalse()) g.Expect(shrunk.podExists("coordinator", 4)).To(BeFalse()) - }, 5*time.Minute, 5*time.Second).Should(Succeed()) + }, 8*time.Minute, 5*time.Second).Should(Succeed()) By("confirming the shrunk cluster is registered, converged, and led by a survivor") Eventually(shrunk.verifyRegistered, 10*time.Minute, 10*time.Second).Should(Succeed()) @@ -823,6 +902,98 @@ func (c clusterUnderTest) podExists(component string, ordinal int32) (bool, erro return strings.TrimSpace(output) != "", nil } +// podUIDs maps every workload pod of the cluster to the UID it currently has. +// A pod that has been replaced carries a different one, which is how a rolling +// restart is observed from the outside: the name is stable across a restart and +// says nothing, the UID changes exactly once per replacement. +func (c clusterUnderTest) podUIDs() (map[string]string, error) { + cmd := exec.Command("kubectl", "get", "pods", "-n", c.namespace, + "-l", "app.kubernetes.io/instance="+c.name, + "-o", `jsonpath={range .items[*]}{.metadata.name}{" "}{.metadata.uid}{" "}`+ + `{range .status.conditions[?(@.type=="Ready")]}{.status}{end}{"\n"}{end}`) + output, err := utils.Run(cmd) + if err != nil { + return nil, fmt.Errorf("listing pod UIDs: %w", err) + } + uids := map[string]string{} + for _, line := range strings.Split(strings.TrimSpace(output), "\n") { + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + uids[fields[0]] = fields[1] + } + return uids, nil +} + +// notReadyPods is how many of the cluster's pods are not ready right now, +// counting a pod that has gone away entirely. +func (c clusterUnderTest) notReadyPods(expected int) (int, error) { + cmd := exec.Command("kubectl", "get", "pods", "-n", c.namespace, + "-l", "app.kubernetes.io/instance="+c.name, + "-o", `jsonpath={range .items[*]}{range .status.conditions[?(@.type=="Ready")]}{.status}{end}{"\n"}{end}`) + output, err := utils.Run(cmd) + if err != nil { + return 0, fmt.Errorf("counting unready pods: %w", err) + } + ready := 0 + for _, line := range strings.Split(strings.TrimSpace(output), "\n") { + if strings.TrimSpace(line) == "" { + continue + } + if strings.TrimSpace(line) == "True" { + ready++ + } + } + // Counted against the pods the cluster should have, not the ones that happen to + // exist: a pod the operator deleted is missing rather than unready, and the + // invariant is about the cluster. + return max(0, expected-ready), nil +} + +// watchRoll polls until every pod's UID has changed from the given baseline, +// returning the order the replacements were first observed in and the highest +// number of pods seen unready at once. +// +// The order is what the whole feature is about, and a UID is the only evidence of +// it that survives: by the time a roll finishes, nothing on the cluster says which +// pod went first. The concurrency count is sampled rather than watched, so it can +// only ever under-report — it is evidence that one pod at a time held, not proof. +func (c clusterUnderTest) watchRoll(before map[string]string, timeout time.Duration) ([]string, int) { + GinkgoHelper() + + var order []string + replaced := map[string]bool{} + maxDown := 0 + + Eventually(func(g Gomega) { + if down, err := c.notReadyPods(len(before)); err == nil && down > maxDown { + maxDown = down + } + now, err := c.podUIDs() + g.Expect(err).NotTo(HaveOccurred()) + // Ordinal order within a poll, so a sample that catches two replacements at + // once is at least deterministic. One pod at a time is asserted separately; + // this only keeps the recorded order stable. + names := make([]string, 0, len(now)) + for name := range now { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + if replaced[name] || now[name] == "" || now[name] == before[name] { + continue + } + replaced[name] = true + order = append(order, name) + } + g.Expect(order).To(HaveLen(len(before)), + "still waiting for every pod to be replaced; so far %v", order) + }, timeout, 2*time.Second).Should(Succeed()) + + return order, maxDown +} + // wipeInstanceRegistration unregisters the named data instance on the // coordinator leader, simulating registration state a pod loses when it is // rescheduled onto a fresh node. UNREGISTER INSTANCE must run on the leader — @@ -854,10 +1025,10 @@ func removeCoordinatorRegistration() (string, error) { return "", fmt.Errorf("no coordinator leader found to remove a coordinator: %w", err) } ordinal := int32(0) - if coordinatorLeaderOf(view) == utils.CoordinatorName(ordinal) { + if coordinatorLeaderOf(view) == resources.CoordinatorInstanceName(ordinal) { ordinal = 1 } - name := utils.CoordinatorName(ordinal) + name := resources.CoordinatorInstanceName(ordinal) cmd := exec.Command("kubectl", "exec", pod, "-n", clusterNamespace, "-c", "memgraph", "--", "bash", "-c", fmt.Sprintf("echo 'REMOVE COORDINATOR %d;' | mgconsole", ordinal+1)) if _, err := utils.Run(cmd); err != nil { @@ -1012,7 +1183,7 @@ func (c clusterUnderTest) leaderPod() (string, []instanceRow, error) { pod, len(view))) continue } - leaderOrdinal, err := utils.CoordinatorOrdinal(leader) + leaderOrdinal, err := resources.CoordinatorOrdinal(leader) if err != nil { errs = append(errs, fmt.Errorf("%s named %s as leader: %w", pod, leader, err)) continue diff --git a/test/utils/names.go b/test/utils/names.go deleted file mode 100644 index c0bd73b..0000000 --- a/test/utils/names.go +++ /dev/null @@ -1,47 +0,0 @@ -/* -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 utils - -import "fmt" - -// The names a cluster member appears under in SHOW INSTANCES, derived from its -// StatefulSet pod ordinal. They are spelled out here rather than taken from the -// operator's own packages on purpose: a test that asked production code what it -// named an instance could never catch it renaming one. - -// CoordinatorName is the SHOW INSTANCES name of the coordinator in the pod with -// the given ordinal: ordinal N registers as coordinator_N+1, because Raft -// coordinator IDs start at one. -func CoordinatorName(ordinal int32) string { - return fmt.Sprintf("coordinator_%d", ordinal+1) -} - -// CoordinatorOrdinal is the inverse of CoordinatorName: the ordinal of the pod -// running the coordinator a view names. -func CoordinatorOrdinal(name string) (int32, error) { - var id int32 - if _, err := fmt.Sscanf(name, "coordinator_%d", &id); err != nil { - return 0, fmt.Errorf("parsing coordinator name %q: %w", name, err) - } - return id - 1, nil -} - -// DataInstanceName is the SHOW INSTANCES name of the data instance in the pod -// with the given ordinal: ordinal N registers as instance_N. -func DataInstanceName(ordinal int32) string { - return fmt.Sprintf("instance_%d", ordinal) -}