diff --git a/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storageclusters.yaml b/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storageclusters.yaml index 321d639e..b5784084 100644 --- a/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storageclusters.yaml +++ b/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storageclusters.yaml @@ -374,14 +374,19 @@ spec: state: type: string subPhase: - description: SubPhase tracks the active drain step within the - remove action. + description: |- + SubPhase tracks the active step within a multi-phase action: the drain steps + for remove, and the non-blocking restart/migration steps + (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + two flows and disambiguated by Action. enum: - Validating - Suspending - Migrating - Verifying - Removing + - Promoting + - ReconcilingWorkers type: string triggered: description: Triggered indicates whether the underlying backend diff --git a/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storagenodesets.yaml b/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storagenodesets.yaml index 297e2a29..73089e4e 100644 --- a/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storagenodesets.yaml +++ b/helm-charts/charts/simplyblock-operator/crds/storage.simplyblock.io_storagenodesets.yaml @@ -259,6 +259,17 @@ spec: x-kubernetes-validations: - message: field is immutable rule: self == oldSelf + newSsdPcie: + description: |- + NewSsdPcie is an action-scoped parameter: it applies ONLY to the node named by nodeUUID + during the current restart action, exactly like workerNode and reattachVolume. It is not + per-set device configuration. It lists additional NVMe PCIe addresses to bind on the + target host and is passed through to the control-plane restart as new_ssd_pcie. Use it when + migrating to a host whose data drives sit at different PCIe addresses than the origin host. + Note: the backend appends these to the node's existing device list rather than replacing it. + items: + type: string + type: array nodeFailureDomains: additionalProperties: format: int32 @@ -443,14 +454,19 @@ spec: state: type: string subPhase: - description: SubPhase tracks the active drain step within the - remove action. + description: |- + SubPhase tracks the active step within a multi-phase action: the drain steps + for remove, and the non-blocking restart/migration steps + (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + two flows and disambiguated by Action. enum: - Validating - Suspending - Migrating - Verifying - Removing + - Promoting + - ReconcilingWorkers type: string triggered: description: Triggered indicates whether the underlying backend diff --git a/operator/api/v1alpha1/storagenodeset_types.go b/operator/api/v1alpha1/storagenodeset_types.go index 1228a8a1..064fefc3 100644 --- a/operator/api/v1alpha1/storagenodeset_types.go +++ b/operator/api/v1alpha1/storagenodeset_types.go @@ -112,6 +112,14 @@ type StorageNodeSetSpec struct { // +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="Reattach Volume" // ReattachVolume reattaches volumes during restart where supported by the backend. ReattachVolume *bool `json:"reattachVolume,omitempty"` + // +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="New SSD PCIe" + // NewSsdPcie is an action-scoped parameter: it applies ONLY to the node named by nodeUUID + // during the current restart action, exactly like workerNode and reattachVolume. It is not + // per-set device configuration. It lists additional NVMe PCIe addresses to bind on the + // target host and is passed through to the control-plane restart as new_ssd_pcie. Use it when + // migrating to a host whose data drives sit at different PCIe addresses than the origin host. + // Note: the backend appends these to the node's existing device list rather than replacing it. + NewSsdPcie []string `json:"newSsdPcie,omitempty"` // +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="OpenShift Cluster" // OpenShiftCluster indicates OpenShift-specific behavior should be enabled. OpenShiftCluster *bool `json:"openShiftCluster,omitempty"` @@ -313,8 +321,11 @@ type ActionStatus struct { ObservedGeneration int64 `json:"observedGeneration,omitempty"` // Triggered indicates whether the underlying backend action has been fired. Triggered bool `json:"triggered,omitempty"` - // SubPhase tracks the active drain step within the remove action. - // +kubebuilder:validation:Enum=Validating;Suspending;Migrating;Verifying;Removing + // SubPhase tracks the active step within a multi-phase action: the drain steps + // for remove, and the non-blocking restart/migration steps + // (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + // two flows and disambiguated by Action. + // +kubebuilder:validation:Enum=Validating;Suspending;Migrating;Verifying;Removing;Promoting;ReconcilingWorkers // +optional SubPhase string `json:"subPhase,omitempty"` // VolumesMigrated is the count of volumes successfully migrated so far. diff --git a/operator/api/v1alpha1/zz_generated.deepcopy.go b/operator/api/v1alpha1/zz_generated.deepcopy.go index d5a6e672..bfecb105 100644 --- a/operator/api/v1alpha1/zz_generated.deepcopy.go +++ b/operator/api/v1alpha1/zz_generated.deepcopy.go @@ -1674,6 +1674,11 @@ func (in *StorageNodeSetSpec) DeepCopyInto(out *StorageNodeSetSpec) { *out = new(bool) **out = **in } + if in.NewSsdPcie != nil { + in, out := &in.NewSsdPcie, &out.NewSsdPcie + *out = make([]string, len(*in)) + copy(*out, *in) + } if in.OpenShiftCluster != nil { in, out := &in.OpenShiftCluster, &out.OpenShiftCluster *out = new(bool) diff --git a/operator/config/crd/bases/storage.simplyblock.io_storageclusters.yaml b/operator/config/crd/bases/storage.simplyblock.io_storageclusters.yaml index 321d639e..b5784084 100644 --- a/operator/config/crd/bases/storage.simplyblock.io_storageclusters.yaml +++ b/operator/config/crd/bases/storage.simplyblock.io_storageclusters.yaml @@ -374,14 +374,19 @@ spec: state: type: string subPhase: - description: SubPhase tracks the active drain step within the - remove action. + description: |- + SubPhase tracks the active step within a multi-phase action: the drain steps + for remove, and the non-blocking restart/migration steps + (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + two flows and disambiguated by Action. enum: - Validating - Suspending - Migrating - Verifying - Removing + - Promoting + - ReconcilingWorkers type: string triggered: description: Triggered indicates whether the underlying backend diff --git a/operator/config/crd/bases/storage.simplyblock.io_storagenodesets.yaml b/operator/config/crd/bases/storage.simplyblock.io_storagenodesets.yaml index 297e2a29..73089e4e 100644 --- a/operator/config/crd/bases/storage.simplyblock.io_storagenodesets.yaml +++ b/operator/config/crd/bases/storage.simplyblock.io_storagenodesets.yaml @@ -259,6 +259,17 @@ spec: x-kubernetes-validations: - message: field is immutable rule: self == oldSelf + newSsdPcie: + description: |- + NewSsdPcie is an action-scoped parameter: it applies ONLY to the node named by nodeUUID + during the current restart action, exactly like workerNode and reattachVolume. It is not + per-set device configuration. It lists additional NVMe PCIe addresses to bind on the + target host and is passed through to the control-plane restart as new_ssd_pcie. Use it when + migrating to a host whose data drives sit at different PCIe addresses than the origin host. + Note: the backend appends these to the node's existing device list rather than replacing it. + items: + type: string + type: array nodeFailureDomains: additionalProperties: format: int32 @@ -443,14 +454,19 @@ spec: state: type: string subPhase: - description: SubPhase tracks the active drain step within the - remove action. + description: |- + SubPhase tracks the active step within a multi-phase action: the drain steps + for remove, and the non-blocking restart/migration steps + (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + two flows and disambiguated by Action. enum: - Validating - Suspending - Migrating - Verifying - Removing + - Promoting + - ReconcilingWorkers type: string triggered: description: Triggered indicates whether the underlying backend diff --git a/operator/dist/install.yaml b/operator/dist/install.yaml index 7a25c9b4..38e3dcf4 100644 --- a/operator/dist/install.yaml +++ b/operator/dist/install.yaml @@ -1588,14 +1588,19 @@ spec: state: type: string subPhase: - description: SubPhase tracks the active drain step within the - remove action. + description: |- + SubPhase tracks the active step within a multi-phase action: the drain steps + for remove, and the non-blocking restart/migration steps + (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + two flows and disambiguated by Action. enum: - Validating - Suspending - Migrating - Verifying - Removing + - Promoting + - ReconcilingWorkers type: string triggered: description: Triggered indicates whether the underlying backend @@ -2015,6 +2020,17 @@ spec: x-kubernetes-validations: - message: field is immutable rule: self == oldSelf + newSsdPcie: + description: |- + NewSsdPcie is an action-scoped parameter: it applies ONLY to the node named by nodeUUID + during the current restart action, exactly like workerNode and reattachVolume. It is not + per-set device configuration. It lists additional NVMe PCIe addresses to bind on the + target host and is passed through to the control-plane restart as new_ssd_pcie. Use it when + migrating to a host whose data drives sit at different PCIe addresses than the origin host. + Note: the backend appends these to the node's existing device list rather than replacing it. + items: + type: string + type: array nodeFailureDomains: additionalProperties: format: int32 @@ -2199,14 +2215,19 @@ spec: state: type: string subPhase: - description: SubPhase tracks the active drain step within the - remove action. + description: |- + SubPhase tracks the active step within a multi-phase action: the drain steps + for remove, and the non-blocking restart/migration steps + (Migrating -> Promoting -> ReconcilingWorkers). Migrating is shared between the + two flows and disambiguated by Action. enum: - Validating - Suspending - Migrating - Verifying - Removing + - Promoting + - ReconcilingWorkers type: string triggered: description: Triggered indicates whether the underlying backend diff --git a/operator/internal/controller/simplyblockstoragenodeset_controller.go b/operator/internal/controller/simplyblockstoragenodeset_controller.go index 59574e09..0f8e3100 100644 --- a/operator/internal/controller/simplyblockstoragenodeset_controller.go +++ b/operator/internal/controller/simplyblockstoragenodeset_controller.go @@ -100,6 +100,12 @@ var ( waitForActionCompletionWaitInterval = 5 * time.Second waitForActionCompletionSleepFn = time.Sleep + // Non-blocking restart/migration phase machine (reconcileRestartAction) tunables. + restartActionRequeue = 5 * time.Second // poll cadence while migrating / awaiting online + restartPromoteRequeue = 2 * time.Second // brief hop between the promote and worker-list phases + restartOnlineDeadline = 250 * time.Second // max wait for the node to reach online (mirrors old 50*5s) + restartPromoteDeadline = 120 * time.Second // max time to keep retrying a failing /promote + syncNodeStatusInterval = 30 * time.Second spdkPodEventDelay = 20 * time.Second @@ -1294,6 +1300,10 @@ func adoptStorageNodeStatus(snCR *simplyblockv1alpha1.StorageNodeSet, nodeName s } } +// checkNodeInfoReachableFn indirects checkNodeInfoReachable so the non-blocking +// restart/migration path can be exercised in tests without a live node-info endpoint. +var checkNodeInfoReachableFn = checkNodeInfoReachable + func checkNodeInfoReachable(ctx context.Context, nodeName, namespace string, tlsEnabled, tlsMutualEnabled bool) error { scheme := "http" httpClient := &http.Client{Timeout: 3 * time.Second} @@ -1743,6 +1753,13 @@ func (r *StorageNodeSetReconciler) reconcileAction( return r.performDrainAndRemove(ctx, apiClient, clusterUUID, snCR) } + // Restart (and workerNode-targeted migration) runs as a non-blocking phase + // machine so the reconcile worker is freed between the fire, await-online, + // promote, and worker-list steps instead of sleeping in-line. + if snCR.Spec.Action == utils.NodeActionRestart { + return r.reconcileRestartAction(ctx, apiClient, clusterUUID, snCR) + } + if err := r.handleNodeAction( ctx, apiClient, @@ -1815,7 +1832,7 @@ func (r *StorageNodeSetReconciler) handleNodeAction( } snCR.Status.ActionStatus.State = utils.ActionStateSuccess - snCR.Status.ActionStatus.Message = "Action executed successfully" + snCR.Status.ActionStatus.Message = actionSuccessMessage snCR.Status.ActionStatus.UpdatedAt = metav1.Now() if err := r.Status().Update(ctx, snCR); err != nil { log.Error(err, "Failed to update action status") @@ -1870,6 +1887,7 @@ func (r *StorageNodeSetReconciler) performNodeAction( payload := map[string]any{ "force": nodeActionForce(snCR, true), "reattach_volume": utils.BoolPtrOrFalse(snCR.Spec.ReattachVolume), + "new_ssd_pcie": nodeActionNewSsdPcie(snCR), } if snCR.Spec.WorkerNode != "" { @@ -1892,6 +1910,7 @@ func (r *StorageNodeSetReconciler) performNodeAction( body = map[string]any{ "force": nodeActionForce(snCR, true), "reattach_volume": utils.BoolPtrOrFalse(snCR.Spec.ReattachVolume), + "new_ssd_pcie": nodeActionNewSsdPcie(snCR), "node_address": utils.StorageNodeSetAPIAddress(snCR.Spec.WorkerNode, snCR.Namespace), } } else { @@ -1979,10 +1998,340 @@ func nodeActionForce(snCR *simplyblockv1alpha1.StorageNodeSet, defaultValue bool return *snCR.Spec.Force } +// nodeActionNewSsdPcie returns the action-scoped NewSsdPcie list normalised to a +// non-nil slice so it serialises as a JSON array ([]) rather than null. +func nodeActionNewSsdPcie(snCR *simplyblockv1alpha1.StorageNodeSet) []string { + if snCR.Spec.NewSsdPcie == nil { + return []string{} + } + return snCR.Spec.NewSsdPcie +} + +const ( + restartSubPhaseMigrating = "Migrating" + restartSubPhasePromoting = "Promoting" + restartSubPhaseReconcilingWorkers = "ReconcilingWorkers" + + actionSuccessMessage = "Action executed successfully" +) + +// reconcileRestartAction drives a restart — and, when spec.workerNode is set, a +// migration to a new host — as a NON-BLOCKING phase machine. Unlike the shared +// handleNodeAction path it never sleeps or polls in-line: each reconcile performs a +// single step and returns a RequeueAfter, freeing the worker to service other objects +// between steps. Phase is persisted in ActionStatus.SubPhase: +// +// (fire) POST /restart (target-worker prep + reachability first for a +// migration), then SubPhase=Migrating +// Migrating poll node status; on "online" -> Promoting (migration) or +// success (plain restart) +// Promoting POST /promote (activate new devices, fail+migrate the removed +// origin-host devices which starts the rebalance, set primary, +// re-home lvols), then SubPhase=ReconcilingWorkers +// ReconcilingWorkers drop the vacated origin host from spec.workerNodes / add the +// target, then success +func (r *StorageNodeSetReconciler) reconcileRestartAction( + ctx context.Context, + apiClient *webapi.Client, + clusterUUID string, + snCR *simplyblockv1alpha1.StorageNodeSet, +) (ctrl.Result, error) { + gen := snCR.Generation + as := snCR.Status.ActionStatus + + // Idempotency gate: this exact request already completed for this generation. + if as != nil && + as.Action == snCR.Spec.Action && + as.NodeUUID == snCR.Spec.NodeUUID && + as.State == utils.ActionStateSuccess && + as.ObservedGeneration == gen { + return ctrl.Result{}, nil + } + + // Fresh request (new or changed spec generation): (re)initialise the running + // status and start from the fire phase. + if as == nil || + as.Action != snCR.Spec.Action || + as.NodeUUID != snCR.Spec.NodeUUID || + as.ObservedGeneration != gen { + as = &simplyblockv1alpha1.ActionStatus{ + Action: snCR.Spec.Action, + NodeUUID: snCR.Spec.NodeUUID, + State: utils.ActionStateRunning, + ObservedGeneration: gen, + UpdatedAt: metav1.Now(), + } + snCR.Status.ActionStatus = as + if err := r.Status().Update(ctx, snCR); err != nil { + return ctrl.Result{}, err + } + // Re-enter to run the fire phase. Keeping one status write per reconcile + // avoids a second same-reconcile update racing the first on resourceVersion, + // and gives the reachability/online deadlines a stable UpdatedAt to measure + // from. + return ctrl.Result{RequeueAfter: restartPromoteRequeue}, nil + } + + switch as.SubPhase { + case "": + return r.restartPhaseFire(ctx, apiClient, clusterUUID, snCR) + case restartSubPhaseMigrating: + return r.restartPhaseAwaitOnline(ctx, apiClient, clusterUUID, snCR) + case restartSubPhasePromoting: + return r.restartPhasePromote(ctx, apiClient, clusterUUID, snCR) + case restartSubPhaseReconcilingWorkers: + return r.restartPhaseReconcileWorkers(ctx, snCR) + default: + return r.failRestart(ctx, snCR, fmt.Sprintf("unexpected restart sub-phase %q", as.SubPhase)) + } +} + +// failRestart records a terminal failure on the restart action. If persisting the +// Failed state fails it returns the error so controller-runtime requeues and writes +// it on a later attempt, rather than leaving the action stuck in Running with no +// further reconciles scheduled. +func (r *StorageNodeSetReconciler) failRestart(ctx context.Context, snCR *simplyblockv1alpha1.StorageNodeSet, msg string) (ctrl.Result, error) { + as := snCR.Status.ActionStatus + as.State = utils.ActionStateFailed + as.Message = msg + as.UpdatedAt = metav1.Now() + logf.FromContext(ctx).Info("Restart action failed", "nodeUUID", snCR.Spec.NodeUUID, "message", msg) + if err := r.Status().Update(ctx, snCR); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{}, nil +} + +// succeedRestart records success and clears the sub-phase. reaffirmGeneration must be +// set when a spec edit happened in this reconcile (worker-list swap), so the success +// record carries the current generation and the idempotency gate keeps matching. +func (r *StorageNodeSetReconciler) succeedRestart(ctx context.Context, snCR *simplyblockv1alpha1.StorageNodeSet, reaffirmGeneration bool) (ctrl.Result, error) { + as := snCR.Status.ActionStatus + as.State = utils.ActionStateSuccess + as.Message = actionSuccessMessage + as.SubPhase = "" + as.UpdatedAt = metav1.Now() + if reaffirmGeneration { + as.ObservedGeneration = snCR.Generation + } + if err := r.Status().Update(ctx, snCR); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{}, nil +} + +// restartPhaseFire prepares the target worker (migration only, non-blocking) and, unless +// the backend already has the restart underway/complete, POSTs /restart. Advances to Migrating. +func (r *StorageNodeSetReconciler) restartPhaseFire(ctx context.Context, apiClient *webapi.Client, clusterUUID string, snCR *simplyblockv1alpha1.StorageNodeSet) (ctrl.Result, error) { + log := logf.FromContext(ctx) + as := snCR.Status.ActionStatus + + if snCR.Spec.WorkerNode != "" { + // Non-blocking target-worker prep: label + EndpointSlice are idempotent, and + // reachability is a single probe — requeue (staying in this phase) until the + // target's node-info endpoint answers, then fire. + if err := r.labelWorkerNode(ctx, snCR); err != nil { + return ctrl.Result{}, fmt.Errorf("failed to label worker node %s: %w", snCR.Spec.WorkerNode, err) + } + if err := r.ensureWorkerInEndpointSlice(ctx, snCR, snCR.Spec.WorkerNode); err != nil { + return ctrl.Result{}, fmt.Errorf("failed to ensure endpoint for worker %s: %w", snCR.Spec.WorkerNode, err) + } + if err := checkNodeInfoReachableFn(ctx, snCR.Spec.WorkerNode, snCR.Namespace, r.TLSEnabled, r.TLSMutualEnabled); err != nil { + if time.Since(as.UpdatedAt.Time) > restartOnlineDeadline { + return r.failRestart(ctx, snCR, fmt.Sprintf("target worker %s never became reachable: %v", snCR.Spec.WorkerNode, err)) + } + log.Info("Target worker not yet reachable; requeuing", "workerNode", snCR.Spec.WorkerNode) + return ctrl.Result{RequeueAfter: restartActionRequeue}, nil + } + } + + // Storm guard: never re-fire a restart the backend already has underway + // (in_restart), or one already completed for this request (online AND we already + // fired — e.g. operator crashed after the POST but before persisting the phase). A + // fresh request against an already-online node still fires, since Triggered is only + // set once we advance past the fire phase. Best-effort read; on error we fire. + if cur, err := getNodeStatus(ctx, apiClient, clusterUUID, snCR.Spec.NodeUUID); err == nil && + (cur == utils.NodeStatusInRestart || (cur == utils.NodeStatusOnline && as.Triggered)) { + log.Info("Restart already underway/complete on backend; not re-firing", + "nodeUUID", snCR.Spec.NodeUUID, "status", cur) + } else if err := r.fireRestart(ctx, apiClient, clusterUUID, snCR); err != nil { + // Transient: leave status running and retry the fire phase. + log.Error(err, "restart POST failed; will retry", "nodeUUID", snCR.Spec.NodeUUID) + return ctrl.Result{RequeueAfter: restartActionRequeue}, nil + } + + as.Triggered = true + as.SubPhase = restartSubPhaseMigrating + as.UpdatedAt = metav1.Now() + if err := r.Status().Update(ctx, snCR); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: restartActionRequeue}, nil +} + +// restartPhaseAwaitOnline does a single node-status poll. On "online" it succeeds a plain +// restart or advances a migration to Promoting; otherwise it requeues until the deadline. +func (r *StorageNodeSetReconciler) restartPhaseAwaitOnline(ctx context.Context, apiClient *webapi.Client, clusterUUID string, snCR *simplyblockv1alpha1.StorageNodeSet) (ctrl.Result, error) { + as := snCR.Status.ActionStatus + cur, err := getNodeStatus(ctx, apiClient, clusterUUID, snCR.Spec.NodeUUID) + if err != nil || cur != utils.NodeStatusOnline { + // UpdatedAt is not rewritten while polling, so it marks when this phase began + // and bounds the total wait. + if time.Since(as.UpdatedAt.Time) > restartOnlineDeadline { + return r.failRestart(ctx, snCR, fmt.Sprintf("node did not reach online after restart (last status %q, err %v)", cur, err)) + } + return ctrl.Result{RequeueAfter: restartActionRequeue}, nil + } + if snCR.Spec.WorkerNode == "" { + return r.succeedRestart(ctx, snCR, false) + } + as.SubPhase = restartSubPhasePromoting + as.UpdatedAt = metav1.Now() + if err := r.Status().Update(ctx, snCR); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: restartPromoteRequeue}, nil +} + +// restartPhasePromote calls /promote (start rebalance + primary switch) and advances to +// ReconcilingWorkers. +func (r *StorageNodeSetReconciler) restartPhasePromote(ctx context.Context, apiClient *webapi.Client, clusterUUID string, snCR *simplyblockv1alpha1.StorageNodeSet) (ctrl.Result, error) { + log := logf.FromContext(ctx) + as := snCR.Status.ActionStatus + if err := r.promoteNode(ctx, apiClient, clusterUUID, snCR.Spec.NodeUUID); err != nil { + // UpdatedAt was stamped on entry to this phase and is not rewritten while + // retrying, so it bounds the total retry window. Fail terminally past the + // deadline instead of looping on a permanent error (e.g. 4xx). + if time.Since(as.UpdatedAt.Time) > restartPromoteDeadline { + return r.failRestart(ctx, snCR, fmt.Sprintf("promote did not succeed within %s: %v", restartPromoteDeadline, err)) + } + log.Error(err, "promote failed; will retry", "nodeUUID", snCR.Spec.NodeUUID) + return ctrl.Result{RequeueAfter: restartActionRequeue}, nil + } + log.Info("Promoted migrated node; rebalance started", + "nodeUUID", snCR.Spec.NodeUUID, "workerNode", snCR.Spec.WorkerNode) + as.SubPhase = restartSubPhaseReconcilingWorkers + as.UpdatedAt = metav1.Now() + if err := r.Status().Update(ctx, snCR); err != nil { + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: restartPromoteRequeue}, nil +} + +// restartPhaseReconcileWorkers swaps the vacated origin host for the target in +// spec.workerNodes, then succeeds the migration. +func (r *StorageNodeSetReconciler) restartPhaseReconcileWorkers(ctx context.Context, snCR *simplyblockv1alpha1.StorageNodeSet) (ctrl.Result, error) { + log := logf.FromContext(ctx) + // status.Nodes is frozen during the action (the action path skips the status + // sync), so it still reports the origin host for this node. + sourceHost := "" + for _, n := range snCR.Status.Nodes { + if n.UUID == snCR.Spec.NodeUUID { + sourceHost = n.Hostname + break + } + } + if sourceHost == "" { + log.Info("Origin host unknown for migrated node; adding target to workerNodes but cannot drop the vacated worker", + "nodeUUID", snCR.Spec.NodeUUID, "workerNode", snCR.Spec.WorkerNode) + } + if reconcileMigratedWorkerList(snCR, sourceHost) { + if err := r.Update(ctx, snCR); err != nil { + log.Error(err, "failed to reconcile worker list after migration; will retry") + return ctrl.Result{RequeueAfter: restartActionRequeue}, nil + } + log.Info("Reconciled worker list after migration", + "removed", sourceHost, "added", snCR.Spec.WorkerNode, "workerNodes", snCR.Spec.WorkerNodes) + } + // reaffirmGeneration: the r.Update above bumped metadata.generation. + res, err := r.succeedRestart(ctx, snCR, true) + if err == nil { + log.Info("Migration completed", "nodeUUID", snCR.Spec.NodeUUID, "workerNode", snCR.Spec.WorkerNode) + } + return res, err +} + +// fireRestart issues the control-plane restart POST for the node. When spec.workerNode +// is set the restart is directed at that host (node_address) for a migration; the +// caller is responsible for target-worker preparation (label/endpoint/reachability). +func (r *StorageNodeSetReconciler) fireRestart( + ctx context.Context, + apiClient *webapi.Client, + clusterUUID string, + snCR *simplyblockv1alpha1.StorageNodeSet, +) error { + body := map[string]any{ + "force": nodeActionForce(snCR, true), + "reattach_volume": utils.BoolPtrOrFalse(snCR.Spec.ReattachVolume), + "new_ssd_pcie": nodeActionNewSsdPcie(snCR), + } + if snCR.Spec.WorkerNode != "" { + body["node_address"] = utils.StorageNodeSetAPIAddress(snCR.Spec.WorkerNode, snCR.Namespace) + } + endpoint := fmt.Sprintf("/api/v2/clusters/%s/storage-nodes/%s/restart", clusterUUID, snCR.Spec.NodeUUID) + respBody, status, err := apiClient.Do(ctx, http.MethodPost, endpoint, body) + if err != nil || status >= 300 { + if err == nil { + err = fmt.Errorf("unexpected status %d", status) + } + return fmt.Errorf("restart API failed: status=%d err=%v resp=%s", status, err, string(respBody)) + } + return nil +} + +// promoteNode calls the control-plane /promote endpoint (make_sec_new_primary) for a +// migrated node: it activates the new-host devices, fails and migrates the removed +// origin-host devices (which starts the rebalance), sets the node primary, and re-homes +// its lvols. +func (r *StorageNodeSetReconciler) promoteNode( + ctx context.Context, + apiClient *webapi.Client, + clusterUUID string, + nodeUUID string, +) error { + endpoint := fmt.Sprintf("/api/v2/clusters/%s/storage-nodes/%s/promote", clusterUUID, nodeUUID) + respBody, status, err := apiClient.Do(ctx, http.MethodPost, endpoint, nil) + if err != nil || status >= 300 { + if err == nil { + err = fmt.Errorf("unexpected status %d", status) + } + return fmt.Errorf("promote API failed: status=%d err=%v resp=%s", status, err, string(respBody)) + } + return nil +} + +// reconcileMigratedWorkerList mutates snCR.Spec.WorkerNodes in place to reflect a +// completed migration: the target workerNode is added (if absent) and the vacated +// sourceHost is removed (when known and distinct from the target). Returns true when +// the list changed. +func reconcileMigratedWorkerList(snCR *simplyblockv1alpha1.StorageNodeSet, sourceHost string) bool { + target := snCR.Spec.WorkerNode + changed := false + + out := make([]string, 0, len(snCR.Spec.WorkerNodes)+1) + for _, w := range snCR.Spec.WorkerNodes { + if sourceHost != "" && sourceHost != target && w == sourceHost { + changed = true + continue + } + out = append(out, w) + } + if !slices.Contains(out, target) { + out = append(out, target) + changed = true + } + if changed { + snCR.Spec.WorkerNodes = out + } + return changed +} + // ensureWorkerInEndpointSlice adds the target worker to the storage-node-api -// EndpointSlice when it is absent. spec.workerNode holds the migration target -// but is never part of spec.workerNodes, so reconcileEndpointSlice would never -// add a DNS hostname entry for it, causing headless-service lookups to fail. +// EndpointSlice when it is absent. During a migration spec.workerNode is not yet +// part of spec.workerNodes (it is only added once the migration succeeds), so +// reconcileEndpointSlice would not add a DNS hostname entry for it, causing +// headless-service lookups to fail. This call bridges that gap; it is a no-op once +// the worker is in spec.workerNodes. func (r *StorageNodeSetReconciler) ensureWorkerInEndpointSlice( ctx context.Context, snCR *simplyblockv1alpha1.StorageNodeSet, diff --git a/operator/internal/controller/simplyblockstoragenodeset_restart_migration_test.go b/operator/internal/controller/simplyblockstoragenodeset_restart_migration_test.go new file mode 100644 index 00000000..c3dfe945 --- /dev/null +++ b/operator/internal/controller/simplyblockstoragenodeset_restart_migration_test.go @@ -0,0 +1,276 @@ +package controller + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + simplyblockv1alpha1 "github.com/simplyblock/simplyblock-operator/api/v1alpha1" + "github.com/simplyblock/simplyblock-operator/internal/utils" + "github.com/simplyblock/simplyblock-operator/internal/webapi" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// restartActionTestServer serves the /restart POST, /promote POST, and GET +// node-status calls that reconcileRestartAction makes, walking through the +// provided statuses (holding on the last). It counts restart and promote POSTs. +func restartActionTestServer(t *testing.T, statuses ...string) (*httptest.Server, *int32, *int32) { + t.Helper() + var restartPosts, promotePosts int32 + var mu sync.Mutex + idx := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/restart"): + atomic.AddInt32(&restartPosts, 1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{}`)) + case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/promote"): + atomic.AddInt32(&promotePosts, 1) + w.WriteHeader(http.StatusNoContent) + default: // GET node status + mu.Lock() + s := statuses[idx] + if idx < len(statuses)-1 { + idx++ + } + mu.Unlock() + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(utils.NodeStatusResponse{UUID: "node-1", Status: s}) + } + })) + return srv, &restartPosts, &promotePosts +} + +// drive runs reconcileRestartAction to completion (success/failed) or until the +// iteration budget is exhausted, re-reading the CR each step like the real loop. +// It returns the number of steps and the terminal action state. +func drive(t *testing.T, r *StorageNodeSetReconciler, apiClient *webapi.Client, key client.ObjectKey) (int, string) { + t.Helper() + ctx := context.Background() + for i := 1; i <= 20; i++ { + cur := &simplyblockv1alpha1.StorageNodeSet{} + if err := r.Get(ctx, key, cur); err != nil { + t.Fatalf("get CR: %v", err) + } + res, err := r.reconcileRestartAction(ctx, apiClient, "cluster", cur) + if err != nil { + t.Fatalf("reconcileRestartAction step %d: %v", i, err) + } + if cur.Status.ActionStatus != nil && + (cur.Status.ActionStatus.State == utils.ActionStateSuccess || + cur.Status.ActionStatus.State == utils.ActionStateFailed) { + return i, cur.Status.ActionStatus.State + } + // Every non-terminal step must free the loop with a delayed requeue. + if res.RequeueAfter <= 0 { + t.Fatalf("step %d returned no requeue while still running (blocking?): %+v", i, res) + } + } + t.Fatalf("phase machine did not terminate") + return 0, "" +} + +// A plain restart (no workerNode) drives fire -> Migrating -> success without +// blocking, firing exactly one /restart and no /promote. +func TestReconcileRestartAction_PlainRestartNonBlocking(t *testing.T) { + srv, restartPosts, promotePosts := restartActionTestServer(t, utils.NodeStatusOnline) + defer srv.Close() + + sn := &simplyblockv1alpha1.StorageNodeSet{ + ObjectMeta: metav1.ObjectMeta{Name: "sn-restart", Namespace: "default"}, + Spec: simplyblockv1alpha1.StorageNodeSetSpec{Action: "restart", NodeUUID: "node-1"}, + } + r := newStorageNodeSetStateTestReconciler(t, sn) + apiClient := webapi.NewClient(srv.URL) + key := client.ObjectKeyFromObject(sn) + + steps, state := drive(t, r, apiClient, key) + if state != utils.ActionStateSuccess { + t.Fatalf("expected success, got %q", state) + } + if steps < 2 { + t.Fatalf("expected the flow to span multiple requeues (non-blocking), got %d step(s)", steps) + } + if got := atomic.LoadInt32(restartPosts); got != 1 { + t.Fatalf("expected exactly one /restart POST, got %d", got) + } + if got := atomic.LoadInt32(promotePosts); got != 0 { + t.Fatalf("expected no /promote for a plain restart, got %d", got) + } + + // Idempotency: once success is recorded for this generation, a further + // reconcile is a no-op with no requeue and no extra POST. + final := &simplyblockv1alpha1.StorageNodeSet{} + if err := r.Get(context.Background(), key, final); err != nil { + t.Fatalf("get CR: %v", err) + } + res, err := r.reconcileRestartAction(context.Background(), apiClient, "cluster", final) + if err != nil { + t.Fatalf("post-success reconcile: %v", err) + } + if res.RequeueAfter != 0 { + t.Fatalf("expected no requeue after success, got %+v", res) + } + if got := atomic.LoadInt32(restartPosts); got != 1 { + t.Fatalf("expected no extra /restart POST after success, got %d", got) + } +} + +// The storm guard must not re-fire a restart the backend already has in progress. +func TestReconcileRestartAction_StormGuard(t *testing.T) { + srv, restartPosts, _ := restartActionTestServer(t, utils.NodeStatusInRestart, utils.NodeStatusOnline) + defer srv.Close() + + sn := &simplyblockv1alpha1.StorageNodeSet{ + ObjectMeta: metav1.ObjectMeta{Name: "sn-inflight", Namespace: "default"}, + Spec: simplyblockv1alpha1.StorageNodeSetSpec{Action: "restart", NodeUUID: "node-1"}, + } + r := newStorageNodeSetStateTestReconciler(t, sn) + apiClient := webapi.NewClient(srv.URL) + key := client.ObjectKeyFromObject(sn) + + _, state := drive(t, r, apiClient, key) + if state != utils.ActionStateSuccess { + t.Fatalf("expected success, got %q", state) + } + if got := atomic.LoadInt32(restartPosts); got != 0 { + t.Fatalf("expected no /restart POST while backend already in_restart, got %d", got) + } +} + +// End-to-end migration (spec.workerNode set, not yet in workerNodes): drives +// fire -> Migrating -> Promoting -> ReconcilingWorkers -> success, asserting exactly +// one /restart POST, one /promote POST, and that workerNodes is reconciled (origin +// dropped, target added). +func TestReconcileRestartAction_MigrationEndToEnd(t *testing.T) { + orig := checkNodeInfoReachableFn + t.Cleanup(func() { checkNodeInfoReachableFn = orig }) + checkNodeInfoReachableFn = func(context.Context, string, string, bool, bool) error { return nil } + + srv, restartPosts, promotePosts := restartActionTestServer(t, utils.NodeStatusOnline) + defer srv.Close() + + node := func(name, ip string) *corev1.Node { + return &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{Type: corev1.NodeInternalIP, Address: ip}}}, + } + } + + sn := &simplyblockv1alpha1.StorageNodeSet{ + ObjectMeta: metav1.ObjectMeta{Name: "sn-mig", Namespace: "default"}, + Spec: simplyblockv1alpha1.StorageNodeSetSpec{ + Action: "restart", + NodeUUID: "node-1", + WorkerNode: "worker-4", + WorkerNodes: []string{"worker-0", "worker-1"}, + }, + Status: simplyblockv1alpha1.StorageNodeSetStatus{ + Nodes: []simplyblockv1alpha1.NodeStatus{{UUID: "node-1", Hostname: "worker-1"}}, + }, + } + r := newStorageNodeSetStateTestReconciler(t, sn, + node("worker-0", "10.0.0.10"), node("worker-1", "10.0.0.11"), node("worker-4", "10.0.0.14")) + apiClient := webapi.NewClient(srv.URL) + key := client.ObjectKeyFromObject(sn) + + _, state := drive(t, r, apiClient, key) + if state != utils.ActionStateSuccess { + t.Fatalf("expected success, got %q", state) + } + if got := atomic.LoadInt32(restartPosts); got != 1 { + t.Fatalf("expected exactly one /restart POST, got %d", got) + } + if got := atomic.LoadInt32(promotePosts); got != 1 { + t.Fatalf("expected exactly one /promote POST, got %d", got) + } + + final := &simplyblockv1alpha1.StorageNodeSet{} + if err := r.Get(context.Background(), key, final); err != nil { + t.Fatalf("get CR: %v", err) + } + if !equalStrings(final.Spec.WorkerNodes, []string{"worker-0", "worker-4"}) { + t.Fatalf("workerNodes = %v, want [worker-0 worker-4] (origin dropped, target added)", final.Spec.WorkerNodes) + } +} + +func TestReconcileMigratedWorkerList(t *testing.T) { + cases := []struct { + name string + workers []string + target string + source string + want []string + changed bool + }{ + { + name: "swap source for target", + workers: []string{"w0", "w1", "w2"}, + target: "w4", + source: "w1", + want: []string{"w0", "w2", "w4"}, + changed: true, + }, + { + name: "source unknown only adds target", + workers: []string{"w0", "w1", "w2"}, + target: "w4", + source: "", + want: []string{"w0", "w1", "w2", "w4"}, + changed: true, + }, + { + name: "already reconciled is a no-op", + workers: []string{"w0", "w2", "w4"}, + target: "w4", + source: "w1", + want: []string{"w0", "w2", "w4"}, + changed: false, + }, + { + name: "target equals source is not dropped", + workers: []string{"w0", "w4"}, + target: "w4", + source: "w4", + want: []string{"w0", "w4"}, + changed: false, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + sn := &simplyblockv1alpha1.StorageNodeSet{ + Spec: simplyblockv1alpha1.StorageNodeSetSpec{ + WorkerNodes: append([]string(nil), tc.workers...), + WorkerNode: tc.target, + }, + } + got := reconcileMigratedWorkerList(sn, tc.source) + if got != tc.changed { + t.Fatalf("changed = %v, want %v", got, tc.changed) + } + if !equalStrings(sn.Spec.WorkerNodes, tc.want) { + t.Fatalf("workerNodes = %v, want %v", sn.Spec.WorkerNodes, tc.want) + } + }) + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +}