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()) + }) +})