Skip to content

feat: stage pod template rolls behind cluster permit #338

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
daanvinken wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat: stage pod template rolls behind cluster permit #338

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5b4bb29
feat: stage pod template rolls behind cluster permit
daanvinken Jul 27, 2026
8371e8c
fix: gofmt import order in workload_roll files
daanvinken Jul 27, 2026
cc83a43
fix: satisfy lint and heal non-template workload drift
daanvinken Jul 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions api/v1alpha1/valkeynode_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,13 +181,22 @@ const (
// of Spec.Config has been successfully applied via CONFIG SET. The cluster
// controller blocks one-at-a-time progress until this condition is True.
ValkeyNodeConditionLiveConfigApplied = "LiveConfigApplied"
// ValkeyNodeConditionWorkloadDrift indicates the desired pod template built
// by the operator differs from the live StatefulSet/Deployment template, and
// the ValkeyNode controller is waiting for a ValkeyCluster roll permit before
// applying the change. Status True means a rolling update is deferred.
ValkeyNodeConditionWorkloadDrift = "WorkloadDrift"
)

const (
// ValkeyNodeReasonPodRunning indicates the pod is running.
ValkeyNodeReasonPodRunning = "PodRunning"
// ValkeyNodeReasonPodNotReady indicates the pod is not ready.
ValkeyNodeReasonPodNotReady = "PodNotReady"
// ValkeyNodeReasonAwaitingRollPermit indicates a pod-template change is ready
// but blocked until the ValkeyCluster controller grants
// valkey.io/allow-workload-revision.
ValkeyNodeReasonAwaitingRollPermit = "AwaitingRollPermit"
// ValkeyNodeReasonPersistentVolumeClaimPending indicates the managed PVC is not ready yet.
ValkeyNodeReasonPersistentVolumeClaimPending = "PersistentVolumeClaimPending"
// ValkeyNodeReasonPersistentVolumeClaimBound indicates the managed PVC is bound and ready to use.
Expand Down
212 changes: 159 additions & 53 deletions internal/controller/valkeycluster_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -541,7 +541,9 @@ func (r *ValkeyClusterReconciler) reconcileValkeyNodes(ctx context.Context, clus
// primary last within each shard, and after an update we requeue
// immediately, re-scraping fresh state before any further rolls.
var clusterState *valkey.ClusterState
if anyNodeRequiresRoll(cluster, nodes, configHash) {
// Scrape topology for Spec rolls and for workload-template rolls (operator
// builder drift) so proactive failover still runs before killing a primary.
if anyNodeRequiresRoll(cluster, nodes, configHash) || anyNodeRequiresWorkloadRoll(nodes) {
operatorPassword, err := fetchSystemUserPassword(ctx, operatorUser, r.Client, cluster.Name, cluster.Namespace)
if err != nil {
return false, fmt.Errorf("failed to fetch operator password for proactive failover: %w", err)
Expand Down Expand Up @@ -616,45 +618,46 @@ func (r *ValkeyClusterReconciler) reconcileValkeyNode(ctx context.Context, clust
},
}

// Check if proactive failover is needed before updating.
if clusterState != nil {
current := &valkeyiov1alpha1.ValkeyNode{}
if err := r.Get(ctx, client.ObjectKeyFromObject(node), current); err != nil {
if !apierrors.IsNotFound(err) {
log.V(1).Info("could not fetch current ValkeyNode for failover check, skipping",
"name", node.Name, "err", err)
}
} else if nodeRequiresRoll(current, desired) {
shard, replicas := findFailoverShard(clusterState, current.Status.PodIP)
if shard != nil {
log.Info("proactive failover before rolling primary",
"name", node.Name, "address", current.Status.PodIP,
"syncedReplicas", len(replicas))
if err := proactiveFailover(ctx, r.Recorder, cluster, shard, replicas); err != nil {
log.Info("proactive failover did not complete, proceeding with roll",
"name", node.Name, "err", err)
}
} else if cluster.Spec.Replicas > 0 {
// findFailoverShard returned nil for one of three reasons:
// 1. Node is the shard primary but has no synced replicas: wait for replica to rejoin
// 2. Node is in a shard but is a replica: safe to roll
// 3. Node isn't in any shard (isolated): safe to roll
// Only case 1 requires waiting; identify it's the actual primary of its shard.
shardInState := clusterState.FindShardForAddress(current.Status.PodIP)
if shardInState != nil && shardInState.GetPrimaryNode() != nil && shardInState.GetPrimaryNode().Address == current.Status.PodIP {
log.Info("primary has no synced replicas, deferring roll",
"name", node.Name, "address", current.Status.PodIP,
"shardNodes", len(shardInState.Nodes),
"shardId", shardInState.Id)
return nodeDeferred, nil
}
}
// Load current node (if any) for failover + workload-drift decisions.
current := &valkeyiov1alpha1.ValkeyNode{}
currentExists := true
if err := r.Get(ctx, client.ObjectKeyFromObject(node), current); err != nil {
if !apierrors.IsNotFound(err) {
return nodeUnchanged, err
}
currentExists = false
}

needsSpecRoll := currentExists && nodeRequiresRoll(current, desired)
needsWorkloadPermit := currentExists && nodeNeedsWorkloadPermit(current)
inFlightWorkloadRoll := currentExists && nodeHasInFlightWorkloadRoll(current)
// Failover / ordered roll applies to Spec changes and workload template rolls.
needsWorkloadRoll := needsWorkloadPermit || inFlightWorkloadRoll

if deferred := r.maybeProactiveFailoverBeforeRoll(ctx, cluster, clusterState, current, needsSpecRoll || needsWorkloadRoll); deferred {
return nodeDeferred, nil
}

// One-at-a-time for Spec and workload rolls: never grant a new permit or
// Spec update while another node still holds an in-flight allow annotation.
if needsSpecRoll || needsWorkloadPermit {
if other, err := r.otherNodeHasInFlightWorkloadRoll(ctx, cluster, node.Name); err != nil {
return nodeUnchanged, err
} else if other {
log.V(1).Info("another ValkeyNode already holds a workload roll permit, waiting", "name", node.Name)
return nodeRequeued, nil
}
}

result, err := controllerutil.CreateOrUpdate(ctx, r.Client, node, func() error {
node.Labels = desired.Labels
node.Spec = desired.Spec
// Spec-driven rolls: grant a one-shot permit so the node controller may
// apply the new pod template. Workload-only drift uses the desired hash
// path below (CreateOrUpdate result None).
if needsSpecRoll {
setNodeAnnotation(node, allowWorkloadRevisionAnnotation, allowWorkloadRevisionAny)
}
return controllerutil.SetControllerReference(cluster, node, r.Scheme)
})
if err != nil {
Expand All @@ -673,31 +676,134 @@ func (r *ValkeyClusterReconciler) reconcileValkeyNode(ctx context.Context, clust
r.Recorder.Eventf(cluster, node, corev1.EventTypeNormal, "ValkeyNodeUpdated", "UpdateValkeyNode", "Updated ValkeyNode %s", node.Name)
return nodeRequeued, nil
case controllerutil.OperationResultNone:
if node.Status.ObservedGeneration > 0 && node.Generation != node.Status.ObservedGeneration {
log.V(1).Info("ValkeyNode spec not yet observed by controller, waiting",
"name", node.Name,
"generation", node.Generation,
"observedGeneration", node.Status.ObservedGeneration)
return nodeRequeued, nil
}
if !node.Status.Ready {
// No spec change, but the node hasn't reached Ready yet (e.g.
// still starting after a prior update). Unlike Updated above, we
// only wait when not-ready; a ready unchanged node is safe to
// advance past.
log.V(1).Info("ValkeyNode not yet ready, waiting", "name", node.Name)
return nodeRequeued, nil
}
if c := meta.FindStatusCondition(node.Status.Conditions, valkeyiov1alpha1.ValkeyNodeConditionLiveConfigApplied); c != nil && c.Status == metav1.ConditionFalse {
log.V(1).Info("ValkeyNode live config not yet applied, waiting", "name", node.Name)
return nodeRequeued, nil
}
return r.handleUnchangedValkeyNode(ctx, cluster, node, inFlightWorkloadRoll, needsWorkloadPermit)
default:
log.V(1).Info("unexpected CreateOrUpdate result", "result", result, "name", node.Name)
}
return nodeUnchanged, nil
}

// maybeProactiveFailoverBeforeRoll runs proactive failover when rolling a
// primary. Returns true when the primary has no synced replica yet (defer).
func (r *ValkeyClusterReconciler) maybeProactiveFailoverBeforeRoll(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, clusterState *valkey.ClusterState, current *valkeyiov1alpha1.ValkeyNode, rolling bool) bool {
log := logf.FromContext(ctx)
if clusterState == nil || !rolling {
return false
}
shard, replicas := findFailoverShard(clusterState, current.Status.PodIP)
if shard != nil {
log.Info("proactive failover before rolling primary",
"name", current.Name, "address", current.Status.PodIP,
"syncedReplicas", len(replicas))
if err := proactiveFailover(ctx, r.Recorder, cluster, shard, replicas); err != nil {
log.Info("proactive failover did not complete, proceeding with roll",
"name", current.Name, "err", err)
}
return false
}
if cluster.Spec.Replicas == 0 {
return false
}
// findFailoverShard returned nil for one of three reasons:
// 1. Node is the shard primary but has no synced replicas: wait for replica to rejoin
// 2. Node is in a shard but is a replica: safe to roll
// 3. Node isn't in any shard (isolated): safe to roll
// Only case 1 requires waiting; identify it's the actual primary of its shard.
shardInState := clusterState.FindShardForAddress(current.Status.PodIP)
if shardInState != nil && shardInState.GetPrimaryNode() != nil && shardInState.GetPrimaryNode().Address == current.Status.PodIP {
log.Info("primary has no synced replicas, deferring roll",
"name", current.Name, "address", current.Status.PodIP,
"shardNodes", len(shardInState.Nodes),
"shardId", shardInState.Id)
return true
}
return false
}

// handleUnchangedValkeyNode handles CreateOrUpdate OperationResultNone: grant
// a workload permit if needed, or wait until the node is settled and Ready.
func (r *ValkeyClusterReconciler) handleUnchangedValkeyNode(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, node *valkeyiov1alpha1.ValkeyNode, inFlightWorkloadRoll, needsWorkloadPermit bool) (nodeResult, error) {
log := logf.FromContext(ctx)

// In-flight permitted roll: wait until the node clears the permit
// (after the workload revision is live) and is Ready.
if inFlightWorkloadRoll {
log.V(1).Info("ValkeyNode workload roll in flight, waiting", "name", node.Name)
return nodeRequeued, nil
}
// Builder/template drift without Spec change: grant permit for this node.
// Sibling in-flight was already checked by the caller.
if needsWorkloadPermit {
return r.grantWorkloadRollPermit(ctx, cluster, node)
}
if node.Status.ObservedGeneration > 0 && node.Generation != node.Status.ObservedGeneration {
log.V(1).Info("ValkeyNode spec not yet observed by controller, waiting",
"name", node.Name,
"generation", node.Generation,
"observedGeneration", node.Status.ObservedGeneration)
return nodeRequeued, nil
}
if !node.Status.Ready {
// No spec change, but the node hasn't reached Ready yet (e.g.
// still starting after a prior update). Unlike Updated above, we
// only wait when not-ready; a ready unchanged node is safe to
// advance past.
log.V(1).Info("ValkeyNode not yet ready, waiting", "name", node.Name)
return nodeRequeued, nil
}
if c := meta.FindStatusCondition(node.Status.Conditions, valkeyiov1alpha1.ValkeyNodeConditionLiveConfigApplied); c != nil && c.Status == metav1.ConditionFalse {
log.V(1).Info("ValkeyNode live config not yet applied, waiting", "name", node.Name)
return nodeRequeued, nil
}
return nodeUnchanged, nil
}

// grantWorkloadRollPermit writes allow-workload-revision for a node waiting on
// builder/template drift. Always requeues so the node controller can apply.
func (r *ValkeyClusterReconciler) grantWorkloadRollPermit(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, node *valkeyiov1alpha1.ValkeyNode) (nodeResult, error) {
log := logf.FromContext(ctx)
desiredHash := ""
if node.Annotations != nil {
desiredHash = node.Annotations[desiredWorkloadRevisionAnnotation]
}
if desiredHash == "" {
// Condition True but annotation not yet set; wait for node controller.
log.V(1).Info("ValkeyNode workload drift pending without desired hash, waiting", "name", node.Name)
return nodeRequeued, nil
}
if !workloadPermitAllows(node, desiredHash) {
patchBase := node.DeepCopy()
if setNodeAnnotation(node, allowWorkloadRevisionAnnotation, desiredHash) {
if err := r.Patch(ctx, node, client.MergeFrom(patchBase)); err != nil {
return nodeUnchanged, err
}
log.Info("granted workload roll permit", "name", node.Name, "hash", desiredHash)
r.Recorder.Eventf(cluster, node, corev1.EventTypeNormal, "WorkloadRollPermitted", "PermitWorkloadRoll",
"Granted workload roll permit for %s (hash %s)", node.Name, desiredHash)
}
}
return nodeRequeued, nil
}

// otherNodeHasInFlightWorkloadRoll reports whether any ValkeyNode in this cluster
// other than excludeName currently holds a workload roll permit.
func (r *ValkeyClusterReconciler) otherNodeHasInFlightWorkloadRoll(ctx context.Context, cluster *valkeyiov1alpha1.ValkeyCluster, excludeName string) (bool, error) {
list := &valkeyiov1alpha1.ValkeyNodeList{}
if err := r.List(ctx, list, client.InNamespace(cluster.Namespace), client.MatchingLabels{LabelCluster: cluster.Name}); err != nil {
return false, err
}
for i := range list.Items {
n := &list.Items[i]
if n.Name == excludeName {
continue
}
if nodeHasInFlightWorkloadRoll(n) {
return true, nil
}
}
return false, nil
}

const (
// gracePeriodBufferSeconds is added on top of cluster-manual-failover-timeout
// so the SIGTERM-triggered failover has headroom to finish before SIGKILL.
Expand Down
Loading