diff --git a/pkg/data-handler/topo/pooler.go b/pkg/data-handler/topo/pooler.go index e0b03350..bbea5450 100644 --- a/pkg/data-handler/topo/pooler.go +++ b/pkg/data-handler/topo/pooler.go @@ -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. diff --git a/pkg/resource-handler/controller/shard/disruption.go b/pkg/resource-handler/controller/shard/disruption.go index caefdd53..6db18622 100644 --- a/pkg/resource-handler/controller/shard/disruption.go +++ b/pkg/resource-handler/controller/shard/disruption.go @@ -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) { diff --git a/pkg/resource-handler/controller/shard/drain_helpers.go b/pkg/resource-handler/controller/shard/drain_helpers.go index e3a49d19..246a6646 100644 --- a/pkg/resource-handler/controller/shard/drain_helpers.go +++ b/pkg/resource-handler/controller/shard/drain_helpers.go @@ -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 { @@ -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 { diff --git a/pkg/resource-handler/controller/shard/maintenance_surge.go b/pkg/resource-handler/controller/shard/maintenance_surge.go index 614436b1..98602218 100644 --- a/pkg/resource-handler/controller/shard/maintenance_surge.go +++ b/pkg/resource-handler/controller/shard/maintenance_surge.go @@ -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 { @@ -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)) diff --git a/pkg/resource-handler/controller/shard/reconcile_data_plane.go b/pkg/resource-handler/controller/shard/reconcile_data_plane.go index 735acbd0..09d8fb21 100644 --- a/pkg/resource-handler/controller/shard/reconcile_data_plane.go +++ b/pkg/resource-handler/controller/shard/reconcile_data_plane.go @@ -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 == "" { diff --git a/pkg/resource-handler/controller/shard/reconcile_pool_pods.go b/pkg/resource-handler/controller/shard/reconcile_pool_pods.go index 030fa5c4..1f120fed 100644 --- a/pkg/resource-handler/controller/shard/reconcile_pool_pods.go +++ b/pkg/resource-handler/controller/shard/reconcile_pool_pods.go @@ -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, @@ -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( @@ -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, @@ -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 @@ -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() { @@ -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 } } @@ -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, @@ -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, @@ -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, diff --git a/pkg/resource-handler/controller/shard/reconcile_quarantine.go b/pkg/resource-handler/controller/shard/reconcile_quarantine.go index d14fc6eb..db05103f 100644 --- a/pkg/resource-handler/controller/shard/reconcile_quarantine.go +++ b/pkg/resource-handler/controller/shard/reconcile_quarantine.go @@ -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, diff --git a/pkg/resource-handler/controller/shard/shard_controller_internal_test.go b/pkg/resource-handler/controller/shard/shard_controller_internal_test.go index 94a624ff..4cde9873 100644 --- a/pkg/resource-handler/controller/shard/shard_controller_internal_test.go +++ b/pkg/resource-handler/controller/shard/shard_controller_internal_test.go @@ -883,7 +883,7 @@ func TestUpdateStatus_FieldOwner(t *testing.T) { // TestHandleScaleDown_ConcurrentDrainPrevention verifies that handleScaleDown // respects the inProgress flag: when any pod already has a drain annotation // (DrainStateRequested, DrainStateDraining, or DrainStateAcknowledged), no new -// drains are initiated for either DRAINED replacement or extra-pod scale-down. +// drains are initiated for extra-pod scale-down. func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { scheme := runtime.NewScheme() _ = multigresv1alpha1.AddToScheme(scheme) @@ -933,14 +933,13 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { tests := map[string]struct { replicas int32 pods []*corev1.Pod - podRoles map[string]string actionTaken bool wantAction bool wantInProgress bool wantNoDrains bool wantDrainedPod string }{ - "drain in progress (DrainStateRequested) blocks DRAINED replacement": { + "drain in progress (DrainStateRequested) blocks a new drain": { replicas: 2, pods: []*corev1.Pod{ makePod(podName0, map[string]string{ @@ -948,14 +947,11 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { }), makePod(podName1, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - }, wantAction: false, wantInProgress: true, wantNoDrains: true, }, - "drain in progress (DrainStateDraining) blocks DRAINED replacement": { + "drain in progress (DrainStateDraining) blocks a new drain": { replicas: 2, pods: []*corev1.Pod{ makePod(podName0, map[string]string{ @@ -963,14 +959,11 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { }), makePod(podName1, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - }, wantAction: false, wantInProgress: true, wantNoDrains: true, }, - "drain in progress (DrainStateAcknowledged) blocks DRAINED replacement": { + "drain in progress (DrainStateAcknowledged) blocks a new drain": { replicas: 2, pods: []*corev1.Pod{ makePod(podName0, map[string]string{ @@ -978,9 +971,6 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { }), makePod(podName1, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - }, wantAction: false, wantInProgress: true, wantNoDrains: true, @@ -1009,15 +999,12 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { wantInProgress: true, wantNoDrains: true, }, - "no drain in progress with DRAINED pod does not auto-drain": { + "no drain in progress with matching replicas does not auto-drain": { replicas: 2, pods: []*corev1.Pod{ makePod(podName0, nil), makePod(podName1, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - }, wantAction: false, wantNoDrains: true, }, @@ -1031,15 +1018,12 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { wantInProgress: false, wantDrainedPod: podName1, }, - "actionTaken from earlier phase blocks DRAINED replacement": { + "actionTaken from earlier phase blocks a new drain": { replicas: 2, pods: []*corev1.Pod{ makePod(podName0, nil), makePod(podName1, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - }, actionTaken: true, wantAction: true, wantInProgress: false, @@ -1056,7 +1040,7 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { wantInProgress: false, wantNoDrains: true, }, - "pod with DeletionTimestamp sets inProgress and blocks DRAINED replacement": { + "pod with DeletionTimestamp sets inProgress and blocks a new drain": { replicas: 2, pods: []*corev1.Pod{ func() *corev1.Pod { @@ -1068,14 +1052,11 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { }(), makePod(podName1, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - }, wantAction: false, wantInProgress: true, wantNoDrains: true, }, - "multiple DRAINED pods with drain in progress drains none": { + "multiple pods with drain in progress drains none": { replicas: 3, pods: []*corev1.Pod{ makePod(podName0, map[string]string{ @@ -1084,10 +1065,6 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { makePod(podName1, nil), makePod(podName2, nil), }, - podRoles: map[string]string{ - podName1: "DRAINED", - podName2: "DRAINED", - }, wantAction: false, wantInProgress: true, wantNoDrains: true, @@ -1111,7 +1088,6 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { for testName, tc := range tests { t.Run(testName, func(t *testing.T) { shard := baseShard.DeepCopy() - shard.Status.PodRoles = tc.podRoles objects := make([]client.Object, 0, len(tc.pods)+1) objects = append(objects, shard) @@ -1141,14 +1117,6 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { poolSpec := multigresv1alpha1.PoolSpec{} - var drainedInTest int32 - for _, role := range tc.podRoles { - if role == "DRAINED" { - drainedInTest++ - } - } - effectiveReplicas := tc.replicas + drainedInTest - gotAction, gotInProgress, err := reconciler.handleScaleDown( context.Background(), shard, @@ -1156,7 +1124,7 @@ func TestHandleScaleDown_ConcurrentDrainPrevention(t *testing.T) { poolSpec, existingPods, tc.replicas, - effectiveReplicas, + tc.replicas, tc.actionTaken, ) if err != nil { @@ -1258,9 +1226,8 @@ func TestSetupWithManager(t *testing.T) { } // TestOrphanByRemainingCount verifies that cleanupDrainedPod handles -// PVC deletion correctly for DRAINED replacement pods (idx < deletion threshold), -// scale-down pods (idx >= deletion threshold), and rolling-update pods under different -// PVC deletion policies. +// PVC deletion correctly for scale-down pods (idx >= replicas) and +// rolling-update pods (idx < replicas) under different PVC deletion policies. func TestOrphanByRemainingCount(t *testing.T) { t.Parallel() @@ -1357,60 +1324,38 @@ func TestCleanupDrainedPod_PVCDeletion(t *testing.T) { // pvcName). orphanByRemainingCount uses this: siblingCount-1 >= threshold // deletes, otherwise orphans. siblingCount int - podRoles map[string]string policy *multigresv1alpha1.PVCDeletionPolicy wantPVC bool wantOrphan bool }{ - "DRAINED small pool (3) -> orphan": { + "idx orphan": { + "idx orphan": { podName: podName5, pvcName: pvcName5, siblingCount: 3, - podRoles: map[string]string{}, policy: deletePolicy, wantPVC: true, wantOrphan: true, }, - "DRAINED large pool (4, scaling to 3) -> in-line delete": { - podName: podName0, - pvcName: pvcName0, - siblingCount: 4, - podRoles: map[string]string{podName0: "DRAINED"}, - policy: deletePolicy, - wantPVC: false, - }, "scale-down large pool (4, scaling to 3) -> in-line delete": { podName: podName5, pvcName: pvcName5, siblingCount: 4, - podRoles: map[string]string{}, policy: deletePolicy, wantPVC: false, }, @@ -1419,12 +1364,8 @@ func TestCleanupDrainedPod_PVCDeletion(t *testing.T) { for tn, tc := range tests { t.Run(tn, func(t *testing.T) { shard := baseShard.DeepCopy() - shard.Status.PodRoles = tc.podRoles pod := makePod(tc.podName) - if tc.podRoles[tc.podName] == "DRAINED" { - pod.Labels[metadata.LabelPodRole] = "DRAINED" - } // Create the target PVC plus filler siblings so the pool+cell has // exactly siblingCount PVCs. @@ -2274,7 +2215,7 @@ func TestCreateMissingResources(t *testing.T) { desiredPod0, _ := BuildPoolPod(shard, poolName, cellName, poolSpec, 0, scheme) hash0 := ComputeSpecHash(desiredPod0) - // Pod 0: not ready, has matching spec hash (not drifted, not DRAINED) + // Pod 0: not ready, has matching spec hash (not drifted) pod0 := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: podName0, @@ -2745,9 +2686,9 @@ func TestIsPoolHealthy(t *testing.T) { } }) - t.Run("DRAINED pod that is not ready does not block health check", func(t *testing.T) { - drainedShard := shard.DeepCopy() - drainedShard.Status.PodRoles = map[string]string{"pod-0": "DRAINED"} + t.Run("QUARANTINED pod that is not ready does not block health check", func(t *testing.T) { + quarantinedShard := shard.DeepCopy() + quarantinedShard.Status.PodRoles = map[string]string{"pod-0": "QUARANTINED"} pods := map[string]*corev1.Pod{ "pod-0": { ObjectMeta: metav1.ObjectMeta{Name: "pod-0"}, @@ -2758,8 +2699,8 @@ func TestIsPoolHealthy(t *testing.T) { }, }, } - if !isPoolHealthy(pods, 1, drainedShard) { - t.Error("DRAINED pod should be excluded from health check") + if !isPoolHealthy(pods, 1, quarantinedShard) { + t.Error("QUARANTINED pod should be excluded from health check") } }) } @@ -5026,17 +4967,6 @@ func TestIsDrainStale(t *testing.T) { } }) - t.Run("DoesNotCancelDrainedPodDrain", func(t *testing.T) { - shardWithDrained := shard.DeepCopy() - pod := matchingPod(0, metadata.DrainStateRequested) - shardWithDrained.Status.PodRoles = map[string]string{ - pod.Name: "DRAINED", - } - if r.isDrainStale(shardWithDrained, pod, metadata.DrainStateRequested) { - t.Error("expected drain NOT to be stale (pod role is DRAINED)") - } - }) - t.Run("DoesNotCancelDrainOnDeletingPod", func(t *testing.T) { pod := matchingPod(4, metadata.DrainStateRequested) now := metav1.Now() @@ -6010,43 +5940,6 @@ func TestReconcilePoolPods_AdditionalErrorPaths(t *testing.T) { } poolSpec := shard.Spec.Pools["main"] - t.Run("syncDrainedLabels error", func(t *testing.T) { - pod := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Name: BuildPoolPodName(shard, "main", "z1", 0), - Namespace: "default", - Labels: map[string]string{ - metadata.LabelMultigresCluster: "test-cluster", - metadata.LabelMultigresDatabase: "db", - metadata.LabelMultigresTableGroup: "tg", - metadata.LabelMultigresShard: "test-shard", - metadata.LabelMultigresPool: "main", - metadata.LabelMultigresCell: "z1", - metadata.LabelPodRole: "DRAINED", - }, - }, - } - - baseClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(shard, pod).Build() - fails := testutil.NewFakeClientWithFailures(baseClient, &testutil.FailureConfig{ - OnPatch: testutil.FailOnObjectName(pod.Name, testutil.ErrPermissionError), - }) - - r := &ShardReconciler{Client: fails, Scheme: scheme, Recorder: record.NewFakeRecorder(10)} - - // Temporarily set the topology returned role to NOT DRAINED, which causes syncDrainedLabels to try and strip the label - // which triggers the patch failure - // Wait, the test uses resolvePodRole, which defaults to whatever unless mocked. - // Since topo is not mocked here, resolvePodRole returns "" -> not DRAINED -> tries to patch remove DRAINED -> fails - existingPods := map[string]*corev1.Pod{ - pod.Name: pod, - } - err := r.syncDrainedLabels(context.Background(), shard, existingPods) - if err == nil || !strings.Contains(err.Error(), "failed to remove DRAINED label") { - t.Fatalf("expected syncDrainedLabels error, got %v", err) - } - }) - t.Run("CreatePVC error", func(t *testing.T) { baseClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(shard).Build() pvcName := BuildPoolDataPVCName(shard, "main", "z1", 0) @@ -6069,10 +5962,11 @@ func TestReconcilePoolPods_AdditionalErrorPaths(t *testing.T) { }) t.Run("markPodPVCOrphan network error", func(t *testing.T) { - // Mock a DRAINED pod so markPodPVCOrphan is called during cleanupDrainedPod + // A scaled-down pod (index >= replicas) so cleanupDrainedPod orphans its + // PVC, exercising the markPodPVCOrphan patch-failure path. pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ - Name: BuildPoolPodName(shard, "main", "z1", 0), + Name: BuildPoolPodName(shard, "main", "z1", 1), Namespace: "default", Labels: map[string]string{ metadata.LabelMultigresCluster: "test-cluster", @@ -6081,7 +5975,6 @@ func TestReconcilePoolPods_AdditionalErrorPaths(t *testing.T) { metadata.LabelMultigresShard: "test-shard", metadata.LabelMultigresPool: "main", metadata.LabelMultigresCell: "z1", - metadata.LabelPodRole: "DRAINED", }, Annotations: map[string]string{ metadata.AnnotationDrainState: metadata.DrainStateReadyForDeletion, @@ -6089,7 +5982,7 @@ func TestReconcilePoolPods_AdditionalErrorPaths(t *testing.T) { }, } - pvcName := BuildPoolDataPVCName(shard, "main", "z1", 0) + pvcName := BuildPoolDataPVCName(shard, "main", "z1", 1) pvc := &corev1.PersistentVolumeClaim{ ObjectMeta: metav1.ObjectMeta{ Name: pvcName, diff --git a/pkg/resource-handler/controller/shard/shard_controller_test.go b/pkg/resource-handler/controller/shard/shard_controller_test.go index f07a4b3f..ea822b4a 100644 --- a/pkg/resource-handler/controller/shard/shard_controller_test.go +++ b/pkg/resource-handler/controller/shard/shard_controller_test.go @@ -1804,7 +1804,7 @@ func TestScaleDown_HealthGateBlocksDrain(t *testing.T) { context.Background(), shard, poolName, multigresv1alpha1.PoolSpec{}, existingPods, 2, // replicas: pod-2 is extra - 2, // effectiveReplicas + 2, // effectiveReplicas: no maintenance surge in this test false, ) if err != nil { @@ -1875,7 +1875,7 @@ func TestScaleDown_HealthGateBlocksDrain(t *testing.T) { context.Background(), shard, poolName, multigresv1alpha1.PoolSpec{}, existingPods, 2, // replicas: pod-2 is extra - 2, // effectiveReplicas + 2, // effectiveReplicas: no maintenance surge in this test false, ) if err != nil { @@ -1988,7 +1988,7 @@ func TestScaleDown_HealthGateBlocksDrain(t *testing.T) { context.Background(), shard, poolName, multigresv1alpha1.PoolSpec{}, existingPods, 2, // replicas: pod-2 is extra - 2, // effectiveReplicas + 2, // effectiveReplicas: no maintenance surge in this test false, ) if err != nil { @@ -2252,111 +2252,3 @@ func TestRollingUpdateWaitsForSiblingCell(t *testing.T) { ) } } - -func TestDrainedPodReplacement(t *testing.T) { - t.Parallel() - scheme := runtime.NewScheme() - _ = multigresv1alpha1.AddToScheme(scheme) - _ = corev1.AddToScheme(scheme) - - shardObj := &multigresv1alpha1.Shard{ - ObjectMeta: metav1.ObjectMeta{ - Name: "test-shard", Namespace: "default", - Labels: map[string]string{metadata.LabelMultigresCluster: "test-cluster"}, - }, - Spec: multigresv1alpha1.ShardSpec{ - DatabaseName: "db", - TableGroupName: "tg", - ShardName: "s1", - Pools: map[multigresv1alpha1.PoolName]multigresv1alpha1.PoolSpec{ - "primary": { - ReplicasPerCell: ptr.To(int32(1)), - Storage: multigresv1alpha1.StorageSpec{Size: "10Gi"}, - }, - }, - }, - } - - podName0 := BuildPoolPodName(shardObj, "primary", "zone1", 0) - shardObj.Status.PodRoles = map[string]string{ - podName0: "DRAINED", - } - - pod := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Name: podName0, - Namespace: "default", - Labels: map[string]string{ - "app.kubernetes.io/component": "shard-pool", - "app.kubernetes.io/instance": "test-cluster", - metadata.LabelMultigresCluster: "test-cluster", - metadata.LabelMultigresDatabase: "db", - metadata.LabelMultigresTableGroup: "tg", - metadata.LabelMultigresShard: "s1", - metadata.LabelMultigresPool: "primary", - metadata.LabelMultigresCell: "zone1", - metadata.LabelPodRole: "DRAINED", - }, - Annotations: map[string]string{}, - }, - Status: corev1.PodStatus{ - Conditions: []corev1.PodCondition{ - {Type: corev1.PodReady, Status: corev1.ConditionTrue}, - }, - }, - } - - // Pre-calculate hash so it doesn't get deleted as drifted - poolSpec := shardObj.Spec.Pools["primary"] - desiredPod, _ := BuildPoolPod(shardObj, "primary", "zone1", poolSpec, 0, scheme) - pod.Annotations[metadata.AnnotationSpecHash] = desiredPod.Annotations[metadata.AnnotationSpecHash] - - c := fake.NewClientBuilder().WithScheme(scheme).WithObjects(shardObj).Build() - r := &ShardReconciler{Client: c, Scheme: scheme, Recorder: record.NewFakeRecorder(10)} - - if err := c.Create(context.Background(), pod); err != nil { - t.Fatalf("failed to create pod: %v", err) - } - - // Run reconcile loop - err := r.reconcilePoolPods( - context.Background(), - shardObj, - "primary", - "zone1", - poolSpec, - &shardRolloutTracker{}, - ) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - // The DRAINED pod should NOT have a drain annotation (DRAINED pods are no longer auto-drained) - err = c.Get( - context.Background(), - types.NamespacedName{Name: podName0, Namespace: "default"}, - pod, - ) - if err != nil { - t.Fatalf("expected pod to exist, got %v", err) - } - - if pod.Annotations[metadata.AnnotationDrainState] != "" { - t.Errorf( - "Expected DRAINED pod to have no drain annotation, got %q", - pod.Annotations[metadata.AnnotationDrainState], - ) - } - - // A stand-in pod at index 1 should be created (replicas=1, effectiveReplicas=2) - podName1 := BuildPoolPodName(shardObj, "primary", "zone1", 1) - standInPod := &corev1.Pod{} - err = c.Get( - context.Background(), - types.NamespacedName{Name: podName1, Namespace: "default"}, - standInPod, - ) - if err != nil { - t.Fatalf("expected stand-in pod at index 1 to be created, got %v", err) - } -} diff --git a/pkg/util/metadata/labels.go b/pkg/util/metadata/labels.go index a741b17f..e352962d 100644 --- a/pkg/util/metadata/labels.go +++ b/pkg/util/metadata/labels.go @@ -160,12 +160,6 @@ const ( // cross-cell durability during a rollout or explicit maintenance request. AnnotationMaintenanceSurge = "maintenance.multigres.com/surge" - // LabelPodRole reflects the pod's topology role (e.g. "DRAINED"). - // Set by the resource-handler when the topology store reports a notable role. - // Used as the durable signal for DRAINED PVC cleanup, since PodRoles may be - // cleared by the data-handler during drain before cleanup runs. - LabelPodRole = "multigres.com/role" - // LabelOrphan marks a resource as orphaned (no longer referenced by an owner) // and eligible for garbage collection. Value is the UTC timestamp when the // resource became orphaned, formatted as OrphanTimestampFormat.