Skip to content

refactor(shard): remove dormant DRAINED stand-in machinery #570

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

Draft
mkindahl wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

refactor(shard): remove dormant DRAINED stand-in machinery #570

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
3 changes: 1 addition & 2 deletions pkg/data-handler/topo/pooler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,7 @@ func GetPoolerStatus(
}
// Quarantined poolers are unrecoverable (postgres cannot start).
// They are surfaced with a distinct QUARANTINED role so they are
// visible in Shard.Status.PodRoles, but they are not routed and do
// not drive the stand-in-replica path (which keyed on DRAINED).
// visible in Shard.Status.PodRoles, but they are not routed.
// The operator replaces them via quarantine remediation (delete pod
// + wipe data PVC + re-bootstrap from backup); GetQuarantinedPods
// carries the reason for that.
Expand Down
15 changes: 2 additions & 13 deletions pkg/resource-handler/controller/shard/drain_helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ import (
"github.com/multigres/multigres-operator/pkg/util/metadata"
)

// resolvePodRole returns the role (e.g. "PRIMARY", "REPLICA", "DRAINED") for a
// pod by checking shard.Status.PodRoles. It checks both the exact pod name and
// resolvePodRole returns the role (e.g. "PRIMARY", "REPLICA", "QUARANTINED") for
// a pod by checking shard.Status.PodRoles. It checks both the exact pod name and
// FQDN prefix (podName.subdomain...) since the data-handler may store either.
func resolvePodRole(shard *multigresv1alpha1.Shard, podName string) string {
if shard.Status.PodRoles == nil {
Expand All @@ -31,17 +31,6 @@ func resolvePodRole(shard *multigresv1alpha1.Shard, podName string) string {
return ""
}

// countDrainedPods returns the number of pods whose topology role is DRAINED.
func countDrainedPods(shard *multigresv1alpha1.Shard, existingPods map[string]*corev1.Pod) int32 {
var count int32
for _, pod := range existingPods {
if resolvePodRole(shard, pod.Name) == "DRAINED" {
count++
}
}
return count
}

// clearDrainAnnotations removes all drain annotations from a pod via merge patch,
// cancelling a drain that is no longer needed (e.g. scale-down reversed).
func clearDrainAnnotations(ctx context.Context, k8sClient client.Client, pod *corev1.Pod) error {
Expand Down
8 changes: 0 additions & 8 deletions pkg/resource-handler/controller/shard/reconcile_data_plane.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,14 +283,6 @@ func (r *ShardReconciler) isDrainStale(
return false
}

// A drain on a DRAINED pod comes from external deletion (kubectl delete),
// which also sets DeletionTimestamp (handled above). If we somehow reach
// here with a DRAINED pod in requested state without a DeletionTimestamp,
// the drain should still complete — it should never be cancelled.
if resolvePodRole(shard, pod.Name) == "DRAINED" {
return false
}

poolName := pod.Labels[metadata.LabelMultigresPool]
cellName := pod.Labels[metadata.LabelMultigresCell]
if poolName == "" || cellName == "" {
Expand Down
104 changes: 16 additions & 88 deletions pkg/resource-handler/controller/shard/reconcile_pool_pods.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,26 +77,17 @@ func (r *ShardReconciler) reconcilePoolPods(
existingPVCs[pvc.Name] = pvc
}

// Phase 0: Sync DRAINED labels and compute effective replicas.
// DRAINED pods stay alive for investigation; stand-in replicas compensate.
drainedCount := countDrainedPods(shard, existingPods)
effectiveReplicas := replicas + drainedCount

if err := r.syncDrainedLabels(ctx, shard, existingPods); err != nil {
return err
}

// Phase 1: Create missing resources and handle terminal/deleted pods
driftedCount, actionTaken, err := r.createMissingResources(
ctx, shard, poolName, cellName, poolSpec, existingPods, existingPVCs, effectiveReplicas,
ctx, shard, poolName, cellName, poolSpec, existingPods, existingPVCs, replicas,
)
if err != nil {
return err
}

// Phase 2: Handle scale-down (extra pod draining, ready-for-deletion cleanup)
actionTaken, inProgress, err := r.handleScaleDown(
ctx, shard, poolName, poolSpec, existingPods, replicas, effectiveReplicas, actionTaken,
ctx, shard, poolName, poolSpec, existingPods, replicas, actionTaken,
)
if err != nil {
return err
Expand Down Expand Up @@ -133,7 +124,6 @@ func (r *ShardReconciler) reconcilePoolPods(

// createMissingResources creates PVCs and Pods that should exist but don't.
// It also handles terminal pods (Failed/Succeeded) and externally-deleted pods.
// effectiveReplicas includes stand-in pods for DRAINED pods (replicas + drainedCount).
// Returns the number of drifted pods and whether an action was taken this reconcile.
func (r *ShardReconciler) createMissingResources(
ctx context.Context,
Expand All @@ -142,11 +132,11 @@ func (r *ShardReconciler) createMissingResources(
poolSpec multigresv1alpha1.PoolSpec,
existingPods map[string]*corev1.Pod,
existingPVCs map[string]*corev1.PersistentVolumeClaim,
effectiveReplicas int32,
replicas int32,
) (driftedCount int, actionTaken bool, err error) {
logger := log.FromContext(ctx)

for i := int32(0); i < effectiveReplicas; i++ {
for i := int32(0); i < replicas; i++ {
podName := BuildPoolPodName(shard, poolName, cellName, int(i))
pvcName := BuildPoolDataPVCName(shard, poolName, cellName, int(i))

Expand Down Expand Up @@ -316,27 +306,24 @@ func isPodReady(pod *corev1.Pod) bool {
return false
}

// isPoolHealthy returns true if all non-draining, non-terminating, non-DRAINED
// isPoolHealthy returns true if all non-draining, non-terminating, non-QUARANTINED
// pods that will remain after scale-down are Ready. Extra pods (index >=
// effectiveReplicas) are excluded so an unhealthy extra pod does not block its
// own removal. DRAINED pods are excluded because they are expected to be
// unhealthy and should not block scale-down of stand-in pods.
// replicas) are excluded so an unhealthy extra pod does not block its own
// removal. QUARANTINED pods are excluded because they are expected to be
// unhealthy and are being replaced by quarantine remediation.
func isPoolHealthy(
existingPods map[string]*corev1.Pod,
effectiveReplicas int32,
replicas int32,
shard *multigresv1alpha1.Shard,
) bool {
for _, pod := range existingPods {
if pod.Annotations[metadata.AnnotationDrainState] != "" || !pod.DeletionTimestamp.IsZero() {
continue
}
if idx, ok := resolvePodIndex(pod.Name); !ok || idx >= int(effectiveReplicas) {
if idx, ok := resolvePodIndex(pod.Name); !ok || idx >= int(replicas) {
continue
}
// DRAINED and QUARANTINED pods are expected to be unhealthy (the latter is
// being replaced by quarantine remediation); they must not block
// scale-down of other pods.
if role := resolvePodRole(shard, pod.Name); role == "DRAINED" || role == "QUARANTINED" {
if resolvePodRole(shard, pod.Name) == "QUARANTINED" {
continue
}
if !isPodReady(pod) {
Expand Down Expand Up @@ -386,8 +373,7 @@ func (r *ShardReconciler) handleExternalDeletion(
}

// handleScaleDown processes pods that need removal: ready-for-deletion cleanup
// and draining extra pods beyond the effective replica count.
// replicas is the user-desired count; effectiveReplicas = replicas + drainedCount.
// and draining extra pods beyond the desired replica count.
// Returns whether an action was taken and whether any drain is in progress.
func (r *ShardReconciler) handleScaleDown(
ctx context.Context,
Expand All @@ -396,7 +382,6 @@ func (r *ShardReconciler) handleScaleDown(
poolSpec multigresv1alpha1.PoolSpec,
existingPods map[string]*corev1.Pod,
replicas int32,
effectiveReplicas int32,
actionTaken bool,
) (bool, bool, error) {
logger := log.FromContext(ctx)
Expand Down Expand Up @@ -426,7 +411,7 @@ func (r *ShardReconciler) handleScaleDown(
}

index, ok := resolvePodIndex(pod.Name)
if !ok || index >= int(effectiveReplicas) {
if !ok || index >= int(replicas) {
extraPods = append(extraPods, pod)
}
}
Expand Down Expand Up @@ -469,9 +454,9 @@ func (r *ShardReconciler) handleScaleDown(
// This prevents cascading failures where removing pods from an unhealthy pool
// could cause an outage.
logger.V(1).
Info("Scale-down check", "extraPods", len(extraPods), "actionTaken", actionTaken, "inProgress", inProgress, "desiredReplicas", replicas, "effectiveReplicas", effectiveReplicas)
Info("Scale-down check", "extraPods", len(extraPods), "actionTaken", actionTaken, "inProgress", inProgress, "desiredReplicas", replicas)
if !actionTaken && !inProgress && len(extraPods) > 0 {
if !isPoolHealthy(existingPods, effectiveReplicas, shard) {
if !isPoolHealthy(existingPods, replicas, shard) {
logger.Info(
"Deferring scale-down: pool has non-ready pods",
"extraPods",
Expand Down Expand Up @@ -677,43 +662,6 @@ func (r *ShardReconciler) selectPodToDrain(
return bestPod
}

// syncDrainedLabels ensures pods with topology role DRAINED have the
// multigres.com/role=DRAINED label, and pods no longer DRAINED have it removed.
// The label is the durable signal for DRAINED PVC cleanup — PodRoles may be
// cleared by the data-handler during drain before cleanup runs.
func (r *ShardReconciler) syncDrainedLabels(
ctx context.Context,
shard *multigresv1alpha1.Shard,
existingPods map[string]*corev1.Pod,
) error {
for _, pod := range existingPods {
role := resolvePodRole(shard, pod.Name)
currentLabel := pod.Labels[metadata.LabelPodRole]

if role == "DRAINED" && currentLabel != "DRAINED" {
patch := client.MergeFrom(pod.DeepCopy())
if pod.Labels == nil {
pod.Labels = make(map[string]string)
}
pod.Labels[metadata.LabelPodRole] = "DRAINED"
if err := r.Patch(ctx, pod, patch); err != nil {
return fmt.Errorf("failed to set DRAINED label on pod %s: %w", pod.Name, err)
}
r.Recorder.Eventf(shard, "Warning", "PodDrained",
"Pod %s detected as DRAINED — provisioning stand-in replica", pod.Name)
} else if role != "DRAINED" && currentLabel == "DRAINED" {
patch := client.MergeFrom(pod.DeepCopy())
delete(pod.Labels, metadata.LabelPodRole)
if err := r.Patch(ctx, pod, patch); err != nil {
return fmt.Errorf("failed to remove DRAINED label from pod %s: %w", pod.Name, err)
}
r.Recorder.Eventf(shard, "Normal", "PodRecovered",
"Pod %s is no longer DRAINED", pod.Name)
}
}
return nil
}

func (r *ShardReconciler) cleanupDrainedPod(
ctx context.Context,
shard *multigresv1alpha1.Shard,
Expand All @@ -724,27 +672,7 @@ func (r *ShardReconciler) cleanupDrainedPod(
) error {
logger := log.FromContext(ctx)

// DRAINED pods always get their PVC marked orphan — data is known-bad.
// The multigres-gc CronJob deletes the PVC after the retention window.
// We check the pod label (not PodRoles) because the data-handler clears
// the topology entry during drain before this cleanup point.
if pod.Labels[metadata.LabelPodRole] == "DRAINED" {
if err := r.cleanupPodPVC(
ctx,
shard,
pod,
poolName,
"DRAINED (data known-bad)",
); err != nil {
return err
}
logger.Info("Drained pod cleanup complete", "pod", pod.Name)
r.Recorder.Eventf(shard, "Normal", "DrainCompleted",
"Completed drain for DRAINED pod %s — PVC cleanup queued", pod.Name)
return nil
}

// For non-DRAINED pods, respect WhenScaled policy
// Respect the WhenScaled PVC-deletion policy.
mergedPolicy := multigresv1alpha1.MergePVCDeletionPolicy(
poolSpec.PVCDeletionPolicy,
shard.Spec.PVCDeletionPolicy,
Expand Down
6 changes: 0 additions & 6 deletions pkg/resource-handler/controller/shard/reconcile_quarantine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,6 @@ const (
//
// Returns true when it took a (destructive) action so the caller requeues and
// skips other disruptive work this cycle.
//
// NOTE(review): this supersedes, for quarantined poolers, the older "stand-in
// replica" model (GetPoolerStatus previously mapped quarantined -> DRAINED,
// which provisioned a replacement at a new index and kept the bad pod). That
// DRAINED machinery in reconcile_pool_pods.go is now dormant for the quarantine
// case; a follow-up can remove it if we settle on wipe-in-place.
func (r *ShardReconciler) reconcileQuarantineRemediation(
ctx context.Context,
store topoclient.Store,
Expand Down
Loading
Loading