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
2 changes: 1 addition & 1 deletion pkg/resource-handler/controller/shard/disruption.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ func (r *ShardReconciler) selectShardScaleDownPod(
for poolName, pool := range shard.Spec.Pools {
for _, cell := range pool.Cells {
group := groups[string(poolName)+"/"+string(cell)]
replicas := poolReplicas(pool) + countDrainedPods(shard, group)
replicas := poolReplicas(pool)
for _, pod := range group {
index, ok := resolvePodIndex(pod.Name)
if ok && index >= int(replicas) && !isMaintenanceSurge(pod) {
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
5 changes: 1 addition & 4 deletions pkg/resource-handler/controller/shard/maintenance_surge.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,9 +99,6 @@ func (r *ShardReconciler) reconcileCellMaintenanceSurge(
baseUnsettled = true
continue
}
if resolvePodRole(shard, pod.Name) == "DRAINED" {
continue
}
stable := isAvailablePooler(pod) &&
pod.Annotations[metadata.AnnotationDrainState] == ""
if !stable {
Expand Down Expand Up @@ -220,7 +217,7 @@ func (r *ShardReconciler) createOrAdoptMaintenanceSurge(
replicas int32,
) error {
logger := log.FromContext(ctx)
index := replicas + countDrainedPods(shard, existingPods)
index := replicas
podName := BuildPoolPodName(shard, poolName, cellName, int(index))
pvcName := BuildPoolDataPVCName(shard, poolName, cellName, int(index))

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 @@ -586,14 +586,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
83 changes: 10 additions & 73 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 @@ -78,11 +78,7 @@ func (r *ShardReconciler) reconcilePoolPods(
existingPVCs[pvc.Name] = pvc
}

// Phase 0: Sync DRAINED labels and reconcile temporary maintenance capacity.
if err := r.syncDrainedLabels(ctx, shard, existingPods); err != nil {
return err
}
drainedCount := countDrainedPods(shard, existingPods)
// Phase 0: Reconcile temporary maintenance capacity.
maintenanceSurges, surgeAction, err := r.reconcileCellMaintenanceSurge(
ctx,
shard,
Expand All @@ -101,9 +97,8 @@ func (r *ShardReconciler) reconcilePoolPods(
return nil
}

// DRAINED pods stay alive for investigation; stand-in replicas compensate.
// Active maintenance surges remain desired until the cell has settled.
effectiveReplicas := replicas + drainedCount + maintenanceSurges
effectiveReplicas := replicas + maintenanceSurges

// Phase 1: Create missing resources and handle terminal/deleted pods
driftedCount, actionTaken, err := r.createMissingResources(
Expand Down Expand Up @@ -155,7 +150,7 @@ 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).
// effectiveReplicas includes temporary maintenance-surge pods (replicas + maintenanceSurges).
// Returns the number of drifted pods and whether an action was taken this reconcile.
func (r *ShardReconciler) createMissingResources(
ctx context.Context,
Expand Down Expand Up @@ -370,10 +365,10 @@ func isPodReady(pod *corev1.Pod) bool {

// isPoolHealthy returns true if the pool/cell has at least effectiveReplicas
// pods, and all of them — except extras (index >= effectiveReplicas) and
// DRAINED/QUARANTINED pods — are Ready, with none draining or terminating.
// QUARANTINED pods — are Ready, with none draining or terminating.
// Extra pods are excluded so an unhealthy extra pod does not block its own
// removal. DRAINED and QUARANTINED pods are excluded because they are
// expected to be unhealthy and should not block scale-down of stand-in pods.
// removal. QUARANTINED pods are excluded because they are expected to be
// unhealthy and are being replaced by quarantine remediation.
//
// The count check matters because a pod drained all the way to deletion
// disappears from existingPods entirely — there is nothing left for the
Expand All @@ -393,7 +388,7 @@ func isPoolHealthy(
if idx, ok := resolvePodIndex(pod.Name); !ok || idx >= int(effectiveReplicas) {
continue
}
if role := resolvePodRole(shard, pod.Name); role == "DRAINED" || role == "QUARANTINED" {
if resolvePodRole(shard, pod.Name) == "QUARANTINED" {
continue
}
if !pod.DeletionTimestamp.IsZero() {
Expand Down Expand Up @@ -443,8 +438,7 @@ func (r *ShardReconciler) isShardHealthy(
replicas = *pool.ReplicasPerCell
}
group := podsByPoolCell[string(poolName)+"/"+string(cell)]
effectiveReplicas := replicas + countDrainedPods(shard, group)
if !isPoolHealthy(group, effectiveReplicas, shard) {
if !isPoolHealthy(group, replicas, shard) {
return false, nil
}
}
Expand Down Expand Up @@ -494,7 +488,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.
// replicas is the user-desired count; effectiveReplicas = replicas + maintenanceSurges.
// Returns whether an action was taken and whether any drain is in progress.
func (r *ShardReconciler) handleScaleDown(
ctx context.Context,
Expand Down Expand Up @@ -936,43 +930,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 @@ -983,27 +940,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