From 2704ec2f839e673038d74c92a35d7ddaff7a7000 Mon Sep 17 00:00:00 2001 From: Austin Barrington Date: Tue, 20 Oct 2026 09:31:02 +0100 Subject: [PATCH] Restart each pod once per rolling update (v0.3.3) A one-node pool restarted its pod twice in one rolling update. With the proxy enabled the StatefulSet kept RollingUpdate with partition 0, so the StatefulSet controller replaced each pod as soon as the template changed, before the proxy had excluded it; the operator then excluded, drained and deleted the replacement again. Larger pools had the same race: most pods were replaced without the exclude, some twice. - With the proxy enabled the partition is pinned at the replica count, so a template change replaces nothing by itself. - The operator walks ordinals from the highest down. After the exclude and drain it lowers the partition to that ordinal and the StatefulSet controller replaces the pod, once; it waits for the pod on the update revision, Ready, then includes it. - A pod already on the update revision is not restarted. - An exclude or include a running proxy keeps refusing is given up after 2 minutes (plus 10 for the restart, for include) with a Warning event, so a broken proxy cannot hold an update. - Without the proxy nothing changes. Chart 0.3.3. --- README.md | 2 +- dist/chart/Chart.yaml | 4 +- docs/compatibility.md | 15 +- .../hyperbytedbcluster_controller.go | 183 +++++++++++----- internal/controller/rollout_partition_test.go | 207 ++++++++++++++++++ 5 files changed, 352 insertions(+), 59 deletions(-) create mode 100644 internal/controller/rollout_partition_test.go diff --git a/README.md b/README.md index 0838991..1832660 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ configurable values. ```sh helm upgrade hyperbytedb-operator \ oci://ghcr.io/hyperbyte-cloud/charts/hyperbytedb-operator \ - --version 0.3.2 \ + --version 0.3.3 \ --namespace hyperbytedb-operator-system \ --reuse-values ``` diff --git a/dist/chart/Chart.yaml b/dist/chart/Chart.yaml index 64d42d4..5c65d5f 100644 --- a/dist/chart/Chart.yaml +++ b/dist/chart/Chart.yaml @@ -7,8 +7,8 @@ type: application # release workflow (.github/workflows/release.yml) stamps them with the git # tag (or a `0.0.0-sha.` snapshot for main branch builds) when the # chart is packaged and pushed to ghcr.io. -version: 0.3.2 -appVersion: "0.3.2" +version: 0.3.3 +appVersion: "0.3.3" home: https://hyperbyte-cloud.github.io/hyperbytedb/ sources: diff --git a/docs/compatibility.md b/docs/compatibility.md index d3ae8a8..dd90e84 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -32,13 +32,14 @@ Reasons: `Supported`, `Assumed` (no version found), `NewerThanKnown` | Operator | Renders for HyperbyteDB | Newest known | |---|---|---| +| v0.3.3 | 0.8.0 and later | 0.9.1-beta | | v0.3.2 | 0.8.0 and later | 0.9.1-beta | | v0.3.1 | 0.8.0 and later | 0.9.1-beta | | v0.3.0 | 0.8.0 and later | 0.9.1-beta | | v0.2.0 | 0.8.0 and later | 0.9.0-beta | | v0.1.x | not version-aware: renders the newest keys for every cluster (0.8.x ignores the ones it does not know) | — | -What v0.3.2 renders per version: +What v0.3.3 renders per version: | Capability | HyperbyteDB | For older versions the operator | |---|---|---| @@ -56,6 +57,18 @@ The source of truth is `CompatibilityMatrix` in ## Upgrading +- v0.3.2 → v0.3.3: each pod restarts once per rolling update. With the + proxy enabled the operator pins the StatefulSet's `RollingUpdate` + partition, so a pod template change no longer replaces pods by itself; + the operator releases one ordinal at a time, highest first, after + excluding that pod from the proxies and waiting `drainWaitSecs`. Before, + the StatefulSet controller replaced each pod straight away, without the + exclude, and the operator then deleted the replacement again. A pod + already on the new revision is not restarted. If a running proxy keeps + refusing an exclude for 2 minutes the pod is restarted anyway (event + `BackendExcludeFailed`), so a broken proxy cannot hold an update. With + `spec.proxy.enabled: false` nothing changes. The partition is applied on + the first reconcile after the upgrade and restarts nothing. - v0.3.1 → v0.3.2: a pool stuck in a rolling restart recovers by itself. The proxy Deployment is applied before the rolling restart, so a broken proxy is fixed even while a restart waits on it; proxies are excluded diff --git a/internal/controller/hyperbytedbcluster_controller.go b/internal/controller/hyperbytedbcluster_controller.go index c528233..a76cc63 100644 --- a/internal/controller/hyperbytedbcluster_controller.go +++ b/internal/controller/hyperbytedbcluster_controller.go @@ -218,8 +218,9 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R // 6b. Operator-driven rolling restart coordination. When the StatefulSet // has a pending rolling update and the proxy is enabled, the operator - // orchestrates pod-by-pod exclusion from the proxy, drain wait, pod - // deletion, and inclusion after the replacement pod is healthy. + // orchestrates pod-by-pod, highest ordinal first: exclusion from the + // proxy, drain wait, release through the partition (rolloutPartition), + // and inclusion once the replacement pod is Ready. if hyperbytedb.ProxyEnabled(cluster) { if result, err := r.reconcileRollingRestart(ctx, cluster, stsResult, replicas); err != nil { return r.setFailedStatus(ctx, cluster, "RollingRestartFailed", err) @@ -476,6 +477,11 @@ type stsReconcileResult struct { func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, configHash string) (stsReconcileResult, error) { desired := hyperbytedb.BuildStatefulSet(cluster, configHash) + if hyperbytedb.ProxyEnabled(cluster) { + desired.Spec.UpdateStrategy.RollingUpdate = &appsv1.RollingUpdateStatefulSetStrategy{ + Partition: ptr.To(rolloutPartition(cluster, *desired.Spec.Replicas)), + } + } if err := controllerutil.SetControllerReference(cluster, desired, r.Scheme); err != nil { return stsReconcileResult{}, err } @@ -517,21 +523,11 @@ func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, existing.Spec.Template.Spec.TopologySpreadConstraints = desired.Spec.Template.Spec.TopologySpreadConstraints existing.Spec.Template.Annotations = desired.Spec.Template.Annotations existing.Spec.Template.Labels = desired.Spec.Template.Labels + // With the proxy enabled this carries the partition from + // rolloutPartition; without it, partition 0 and the StatefulSet + // controller rolls pods itself. existing.Spec.UpdateStrategy = desired.Spec.UpdateStrategy - // When the proxy is enabled the operator drives pod lifecycle itself - // (exclude → drain → delete → include) so there is no need to pin - // the StatefulSet partition. Clear any leftover partition from a - // previous operator version so the STS controller can reconcile - // normally after the operator coordination finishes. - if existing.Spec.UpdateStrategy.RollingUpdate != nil && - existing.Spec.UpdateStrategy.RollingUpdate.Partition != nil && - *existing.Spec.UpdateStrategy.RollingUpdate.Partition != 0 { - existing.Spec.UpdateStrategy.RollingUpdate = &appsv1.RollingUpdateStatefulSetStrategy{ - Partition: ptr.To(int32(0)), - } - } - if err := r.Update(ctx, existing); err != nil { return stsReconcileResult{}, err } @@ -543,6 +539,50 @@ func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, }, nil } +// rolloutPartition is the StatefulSet partition while the operator drives +// rollouts (proxy enabled). With partition 0 the StatefulSet controller +// replaced each pod as soon as the template changed, before the proxy had +// excluded it, and the operator then deleted the replacement again: every +// rollout restarted pods twice. Pinned, a pod only moves to the new revision +// once the operator releases its ordinal, highest first, after the exclude +// and drain. +func rolloutPartition(cluster *hyperbytedbv1alpha1.HyperbytedbCluster, replicas int32) int32 { + state := cluster.Status.RollingRestart + if state == nil { + return replicas + } + switch state.Phase { + case hyperbytedbv1alpha1.RollingRestartCompleted: + return 0 + case hyperbytedbv1alpha1.RollingRestartWaitingReady: + return state.CurrentOrdinal + default: + return state.CurrentOrdinal + 1 + } +} + +// releaseOrdinal lowers the StatefulSet partition to ordinal, so the +// StatefulSet controller replaces that pod with the update revision. It +// never raises it: the pods above are already updated. +func (r *HyperbytedbClusterReconciler) releaseOrdinal(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, ordinal int32) error { + sts := &appsv1.StatefulSet{} + if err := r.Get(ctx, types.NamespacedName{ + Name: hyperbytedb.StatefulSetName(cluster), Namespace: cluster.Namespace, + }, sts); err != nil { + return err + } + ru := sts.Spec.UpdateStrategy.RollingUpdate + if ru != nil && ru.Partition != nil && *ru.Partition <= ordinal { + return nil + } + patch := client.MergeFrom(sts.DeepCopy()) + if ru == nil { + sts.Spec.UpdateStrategy.RollingUpdate = &appsv1.RollingUpdateStatefulSetStrategy{} + } + sts.Spec.UpdateStrategy.RollingUpdate.Partition = ptr.To(ordinal) + return r.Patch(ctx, sts, patch) +} + // ---------- Rolling Restart Coordination ---------- // reconcileRollingRestart orchestrates pod-by-pod proxy exclusion during a @@ -601,13 +641,14 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( podName := fmt.Sprintf("%s-%d", stsName, state.CurrentOrdinal) requeue3s := ctrl.Result{RequeueAfter: 3 * time.Second} + updateRevision := curSTS.Status.UpdateRevision switch state.Phase { case hyperbytedbv1alpha1.RollingRestartExcluding: - return r.rollingRestartExcluding(ctx, cluster, state, podName, proxyPort, requeue3s) + return r.rollingRestartExcluding(ctx, cluster, state, podName, updateRevision, proxyPort, requeue3s) case hyperbytedbv1alpha1.RollingRestartDraining: return r.rollingRestartDraining(ctx, cluster, state, podName, requeue3s) case hyperbytedbv1alpha1.RollingRestartWaitingReady: - return r.rollingRestartWaitingReady(ctx, cluster, state, podName, proxyPort, requeue3s) + return r.rollingRestartWaitingReady(ctx, cluster, state, podName, updateRevision, proxyPort, requeue3s) case hyperbytedbv1alpha1.RollingRestartCompleted: return &requeue3s, nil default: @@ -646,9 +687,12 @@ func (r *HyperbytedbClusterReconciler) ensureRollingRestartState( replicas int32, ) (*hyperbytedbv1alpha1.RollingRestartState, error) { log := logf.FromContext(ctx) + // Highest ordinal first: the StatefulSet partition releases every + // ordinal at or above it. + first := max(replicas-1, 0) if cluster.Status.RollingRestart == nil { cluster.Status.RollingRestart = &hyperbytedbv1alpha1.RollingRestartState{ - CurrentOrdinal: 0, + CurrentOrdinal: first, TotalOrdinals: replicas, Phase: hyperbytedbv1alpha1.RollingRestartExcluding, PhaseStartedAt: metav1.Now(), @@ -662,7 +706,7 @@ func (r *HyperbytedbClusterReconciler) ensureRollingRestartState( state := cluster.Status.RollingRestart if state == nil { state = &hyperbytedbv1alpha1.RollingRestartState{ - CurrentOrdinal: 0, + CurrentOrdinal: first, TotalOrdinals: replicas, Phase: hyperbytedbv1alpha1.RollingRestartExcluding, PhaseStartedAt: metav1.Now(), @@ -680,6 +724,7 @@ func (r *HyperbytedbClusterReconciler) rollingRestartExcluding( cluster *hyperbytedbv1alpha1.HyperbytedbCluster, state *hyperbytedbv1alpha1.RollingRestartState, podName string, + updateRevision string, proxyPort int32, requeue ctrl.Result, ) (*ctrl.Result, error) { @@ -687,15 +732,17 @@ func (r *HyperbytedbClusterReconciler) rollingRestartExcluding( pod := &corev1.Pod{} if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { if apierrors.IsNotFound(err) { - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue, nil + return r.rollingRestartRelease(ctx, cluster, state, requeue) } return nil, err } + // Already on the new revision (it crashed and came back, or a rollout + // begun by an older operator replaced it): restarting it again is the + // double restart this coordination exists to prevent. + if updateRevision != "" && pod.Labels[appsv1.StatefulSetRevisionLabel] == updateRevision { + log.Info("Pod already runs the update revision, not restarting it", "pod", podName) + return r.rollingRestartRelease(ctx, cluster, state, requeue) + } podIP := pod.Status.PodIP if podIP == "" { @@ -705,13 +752,20 @@ func (r *HyperbytedbClusterReconciler) rollingRestartExcluding( state.OldPodIP = podIP if err := r.excludeFromAllProxies(ctx, cluster, podIP, proxyPort); err != nil { - log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) - return &requeue, nil + // The partition holds the rollout until the operator releases each + // pod, so a proxy that refuses the exclude must not hold it forever. + if time.Since(state.PhaseStartedAt.Time) < proxyCallTimeout { + log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) + return &requeue, nil + } + r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "BackendExcludeFailed", + "Restarting pod %s without excluding it from the proxy: %v", podName, err) + } else { + log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", + "Excluded pod %s (%s) from proxy routing", podName, podIP) + state.ExcludeConfirmed = true } - log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", - "Excluded pod %s (%s) from proxy routing", podName, podIP) - state.ExcludeConfirmed = true state.Phase = hyperbytedbv1alpha1.RollingRestartDraining state.PhaseStartedAt = metav1.Now() if err := r.Status().Update(ctx, cluster); err != nil { @@ -739,25 +793,23 @@ func (r *HyperbytedbClusterReconciler) rollingRestartDraining( return &ctrl.Result{RequeueAfter: drainWait - elapsed + time.Second}, nil } - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue, nil - } - return nil, err - } + log.Info("Releasing pod to the update revision", "pod", podName) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", + "Released pod %s for rolling upgrade", podName) + return r.rollingRestartRelease(ctx, cluster, state, requeue) +} - if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { - return nil, fmt.Errorf("deleting pod %s for rolling restart: %w", podName, err) +// rollingRestartRelease lowers the partition to the current ordinal (the +// StatefulSet controller then replaces that pod, once) and waits for it. +func (r *HyperbytedbClusterReconciler) rollingRestartRelease( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + requeue ctrl.Result, +) (*ctrl.Result, error) { + if err := r.releaseOrdinal(ctx, cluster, state.CurrentOrdinal); err != nil { + return nil, fmt.Errorf("releasing ordinal %d for rolling restart: %w", state.CurrentOrdinal, err) } - log.Info("Deleted pod for rolling restart", "pod", podName) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", - "Deleted pod %s for rolling upgrade", podName) state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady state.PhaseStartedAt = metav1.Now() if err := r.Status().Update(ctx, cluster); err != nil { @@ -771,6 +823,7 @@ func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( cluster *hyperbytedbv1alpha1.HyperbytedbCluster, state *hyperbytedbv1alpha1.RollingRestartState, podName string, + updateRevision string, proxyPort int32, requeue ctrl.Result, ) (*ctrl.Result, error) { @@ -784,6 +837,12 @@ func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( return nil, err } + // Until the StatefulSet controller acts on the partition the old pod is + // still here, Ready. + if updateRevision != "" && pod.Labels[appsv1.StatefulSetRevisionLabel] != updateRevision { + log.Info("Waiting for the StatefulSet to replace the pod", "pod", podName) + return &requeue, nil + } if !isPodReady(pod) || pod.DeletionTimestamp != nil { if pod.DeletionTimestamp != nil { log.Info("Old pod still terminating, waiting for replacement", "pod", podName) @@ -800,15 +859,20 @@ func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( } if err := r.includeFromAllProxies(ctx, cluster, newIP, proxyPort); err != nil { - log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) - return &requeue, nil + // PhaseStartedAt includes the pod's own restart, so allow for that. + if time.Since(state.PhaseStartedAt.Time) < proxyCallTimeout+podRestartAllowance { + log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) + return &requeue, nil + } + r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "BackendIncludeFailed", + "Moving on without including pod %s (%s) in the proxy: %v", podName, newIP, err) + } else { + log.Info("Included backend in proxy", "pod", podName, "ip", newIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", + "Included pod %s (%s) in proxy routing", podName, newIP) } - log.Info("Included backend in proxy", "pod", podName, "ip", newIP) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", - "Included pod %s (%s) in proxy routing", podName, newIP) - state.CurrentOrdinal++ - if state.CurrentOrdinal >= state.TotalOrdinals { + if state.CurrentOrdinal <= 0 { state.Phase = hyperbytedbv1alpha1.RollingRestartCompleted state.PhaseStartedAt = metav1.Now() if err := r.Status().Update(ctx, cluster); err != nil { @@ -818,6 +882,7 @@ func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( return &ctrl.Result{RequeueAfter: 3 * time.Second}, nil } + state.CurrentOrdinal-- state.Phase = hyperbytedbv1alpha1.RollingRestartExcluding state.PhaseStartedAt = metav1.Now() state.ExcludeConfirmed = false @@ -828,6 +893,14 @@ func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( return &requeue, nil } +// How long a rolling restart retries an exclude or include on a running +// proxy before it moves on without it, and the extra allowance for the +// restart itself on the include. +const ( + proxyCallTimeout = 2 * time.Minute + podRestartAllowance = 10 * time.Minute +) + // isPodReady returns true if the pod has a Ready condition with Status=True. func isPodReady(pod *corev1.Pod) bool { for _, cond := range pod.Status.Conditions { diff --git a/internal/controller/rollout_partition_test.go b/internal/controller/rollout_partition_test.go new file mode 100644 index 0000000..28ce4e8 --- /dev/null +++ b/internal/controller/rollout_partition_test.go @@ -0,0 +1,207 @@ +package controller + +import ( + "context" + "fmt" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + hyperbytedbv1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" + "github.com/hyperbyte-cloud/hyperbytedb-operator/internal/hyperbytedb" +) + +// A one-node pool restarted its pod twice in one rolling update: with +// partition 0 the StatefulSet controller replaced the pod as soon as the +// template changed, then the operator's own drain step deleted the +// replacement. Envtest runs no StatefulSet controller, so playStatefulSet +// below does its part: replace, highest first, the pods the partition +// releases. +var _ = Describe("Rolling update through the StatefulSet partition", func() { + ctx := context.Background() + var name string + var key types.NamespacedName + pools := 0 + + reconcileOnce := func() { + r := &HyperbytedbClusterReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Recorder: record.NewFakeRecorder(256), + Members: hyperbytedb.NewMemberManager(), + } + _, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: key}) + Expect(err).NotTo(HaveOccurred()) + } + + getSTS := func() *appsv1.StatefulSet { + sts := &appsv1.StatefulSet{} + Expect(k8sClient.Get(ctx, key, sts)).To(Succeed()) + return sts + } + partition := func() int32 { + ru := getSTS().Spec.UpdateStrategy.RollingUpdate + Expect(ru).NotTo(BeNil()) + Expect(ru.Partition).NotTo(BeNil()) + return *ru.Partition + } + getState := func() *hyperbytedbv1alpha1.RollingRestartState { + c := &hyperbytedbv1alpha1.HyperbytedbCluster{} + Expect(k8sClient.Get(ctx, key, c)).To(Succeed()) + return c.Status.RollingRestart + } + + podName := func(ordinal int) string { return fmt.Sprintf("%s-%d", name, ordinal) } + createPod := func(ordinal int, revision string) { + pod := testPod(podName(ordinal), map[string]string{ + "app.kubernetes.io/name": "hyperbytedb", + appsv1.StatefulSetRevisionLabel: revision, + }, fmt.Sprintf("127.0.0.%d", ordinal+2), "hyperbytedb", true) + Expect(createWithStatus(ctx, pod)).To(Succeed()) + } + podRevision := func(ordinal int) string { + pod := &corev1.Pod{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: podName(ordinal), Namespace: "default"}, pod)).To(Succeed()) + return pod.Labels[appsv1.StatefulSetRevisionLabel] + } + + // setStatus writes what the StatefulSet controller would report for + // these pod revisions, completing the update once every pod has it. + setStatus := func(replicas int, current, update string) { + updated := int32(0) + for i := range replicas { + if podRevision(i) == update { + updated++ + } + } + if updated == int32(replicas) { + current = update + } + sts := getSTS() + sts.Status.Replicas = int32(replicas) + sts.Status.ReadyReplicas = int32(replicas) + sts.Status.UpdatedReplicas = updated + sts.Status.CurrentRevision = current + sts.Status.UpdateRevision = update + Expect(k8sClient.Status().Update(ctx, sts)).To(Succeed()) + } + + // playStatefulSet replaces the highest released pod not yet on the + // update revision, as the StatefulSet controller does: one at a time. + playStatefulSet := func(replicas int, update string) (int, bool) { + p := int(partition()) + for i := replicas - 1; i >= p; i-- { + if podRevision(i) == update { + continue + } + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName(i), Namespace: "default"}} + Expect(k8sClient.Delete(ctx, pod)).To(Succeed()) + createPod(i, update) + return i, true + } + return 0, false + } + + // backdate makes a drain wait already over. + backdate := func() { + c := &hyperbytedbv1alpha1.HyperbytedbCluster{} + Expect(k8sClient.Get(ctx, key, c)).To(Succeed()) + if c.Status.RollingRestart != nil && + c.Status.RollingRestart.Phase == hyperbytedbv1alpha1.RollingRestartDraining { + c.Status.RollingRestart.PhaseStartedAt = metav1.NewTime(time.Now().Add(-time.Hour)) + Expect(k8sClient.Status().Update(ctx, c)).To(Succeed()) + } + } + + newPool := func(replicas int) { + pools++ + name = fmt.Sprintf("rollout-pool-%d", pools) + key = types.NamespacedName{Name: name, Namespace: "default"} + Expect(k8sClient.Create(ctx, &hyperbytedbv1alpha1.HyperbytedbCluster{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: hyperbytedbv1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(replicas)), + Version: "0.9.1-beta", + Proxy: &hyperbytedbv1alpha1.ProxySpec{Enabled: true}, + }, + })).To(Succeed()) + reconcileOnce() + } + + AfterEach(func() { + _ = k8sClient.DeleteAllOf(ctx, &corev1.Pod{}, client.InNamespace("default")) + }) + + // rollOut changes the template and drives the update to the end. It + // returns the ordinals in the order they were replaced, and the restart + // phase the operator was in for each. + rollOut := func(replicas int) ([]int, []hyperbytedbv1alpha1.RollingRestartPhase) { + c := &hyperbytedbv1alpha1.HyperbytedbCluster{} + Expect(k8sClient.Get(ctx, key, c)).To(Succeed()) + c.Spec.Version = "0.9.2-beta" + Expect(k8sClient.Update(ctx, c)).To(Succeed()) + setStatus(replicas, "old", "new") + + var order []int + var phases []hyperbytedbv1alpha1.RollingRestartPhase + for range 20 * replicas { + reconcileOnce() + if i, ok := playStatefulSet(replicas, "new"); ok { + order = append(order, i) + phases = append(phases, getState().Phase) + } + setStatus(replicas, "old", "new") + backdate() + if getState() == nil && podRevision(0) == "new" { + break + } + } + return order, phases + } + + for _, replicas := range []int{1, 3} { + It(fmt.Sprintf("replaces each of %d pods once, highest first, after its drain", replicas), func() { + newPool(replicas) + Expect(partition()).To(Equal(int32(replicas)), "pinned before any update") + for i := range replicas { + createPod(i, "old") + } + setStatus(replicas, "old", "old") + + order, phases := rollOut(replicas) + + want := make([]int, 0, replicas) + for i := replicas - 1; i >= 0; i-- { + want = append(want, i) + } + Expect(order).To(Equal(want), "each pod replaced once, highest ordinal first") + for _, ph := range phases { + Expect(ph).To(Equal(hyperbytedbv1alpha1.RollingRestartWaitingReady), + "a pod is replaced only once the operator has excluded and drained it") + } + Expect(getState()).To(BeNil(), "the restart completes") + reconcileOnce() + Expect(partition()).To(Equal(int32(replicas)), "pinned again for the next update") + }) + } + + It("leaves a pod that already runs the update revision alone", func() { + newPool(2) + createPod(0, "old") + createPod(1, "new") + setStatus(2, "old", "new") + + order, _ := rollOut(2) + Expect(order).To(Equal([]int{0})) + Expect(getState()).To(BeNil()) + }) +})