diff --git a/pkg/data-handler/topo/pooler.go b/pkg/data-handler/topo/pooler.go index 252a343b..ce88c84c 100644 --- a/pkg/data-handler/topo/pooler.go +++ b/pkg/data-handler/topo/pooler.go @@ -22,7 +22,10 @@ const deadPoolerReason = "operator: no backing pod for pooler" // PoolerStatusResult holds the result of querying pooler roles from the topology. type PoolerStatusResult struct { // Roles maps hostname to its operator-facing role - // (PRIMARY, REPLICA, DRAINED). + // (PRIMARY, REPLICA, QUARANTINED). Shutdown poolers are omitted. QUARANTINED + // poolers are surfaced for visibility but are not routed and do not drive the + // stand-in-replica path; they are replaced via quarantine remediation (see + // GetQuarantinedPods). Roles map[string]string // QuerySuccess indicates whether all topology queries succeeded. QuerySuccess bool @@ -52,10 +55,18 @@ func GetPoolerStatus( if isLifecycleShutdown(p.Multipooler) { continue } + // 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). + // The operator replaces them via quarantine remediation (delete pod + // + wipe data PVC + re-bootstrap from backup); GetQuarantinedPods + // carries the reason for that. roleName := "REPLICA" - if isLifecycleQuarantined(p.Multipooler) { - roleName = "DRAINED" - } else if IsPrimaryPooler(p.Multipooler) { + switch { + case isLifecycleQuarantined(p.Multipooler): + roleName = "QUARANTINED" + case IsPrimaryPooler(p.Multipooler): roleName = "PRIMARY" } // Match the topology entry to an actual managed pod. @@ -72,6 +83,59 @@ func GetPoolerStatus( return result } +// QuarantinedPod identifies a managed pod whose backing pooler has +// self-quarantined, along with the human-readable reason the pooler recorded in +// its topology lifecycle entry (e.g. "postgres failed to recover for 5m0s +// across 60 attempts (last error: ...)"). +type QuarantinedPod struct { + PodName string + Reason string +} + +// GetQuarantinedPods returns the managed pods whose backing pooler has +// self-quarantined (LIFECYCLE_QUARANTINED) in topology — postgres is +// unrecoverably failing to start, so the node needs replacement and data +// remediation — each with the reason recorded on its lifecycle entry. Only pods +// present in managedPodNames are returned; the result is sorted by pod name for +// deterministic, one-at-a-time remediation. A cell whose topology is +// temporarily unavailable is skipped rather than failing the whole call. +func GetQuarantinedPods( + ctx context.Context, + store topoclient.Store, + shard *multigresv1alpha1.Shard, + managedPodNames []string, +) ([]QuarantinedPod, error) { + var quarantined []QuarantinedPod + for _, cell := range CollectCells(shard) { + poolers, err := store.GetMultipoolersByCell(ctx, cell, ShardFilter(shard)) + if err != nil { + if IsTopoUnavailable(err) { + continue + } + return nil, fmt.Errorf( + "listing poolers in cell %q for quarantine detection: %w", + cell, + err, + ) + } + for _, p := range poolers { + if !isLifecycleQuarantined(p.Multipooler) { + continue + } + if podName := matchPoolerToPod(p, managedPodNames); podName != "" { + quarantined = append(quarantined, QuarantinedPod{ + PodName: podName, + Reason: p.Multipooler.GetLifecycleStatus().GetReason(), + }) + } + } + } + slices.SortFunc(quarantined, func(a, b QuarantinedPod) int { + return strings.Compare(a.PodName, b.PodName) + }) + return quarantined, nil +} + // matchPoolerToPod finds the managed pod name that matches a topology pooler // entry, using the FQDN-aware PodMatchesPooler comparison. func matchPoolerToPod(p *topoclient.MultipoolerInfo, podNames []string) string { diff --git a/pkg/data-handler/topo/pooler_test.go b/pkg/data-handler/topo/pooler_test.go index 2fa50d87..d6ac210e 100644 --- a/pkg/data-handler/topo/pooler_test.go +++ b/pkg/data-handler/topo/pooler_test.go @@ -861,8 +861,10 @@ func TestGetPoolerStatus(t *testing.T) { if result.Roles["unknown"] != "REPLICA" { t.Errorf("expected REPLICA fallback, got %s", result.Roles["unknown"]) } - if result.Roles["quarantined"] != "DRAINED" { - t.Errorf("expected DRAINED, got %s", result.Roles["quarantined"]) + // Quarantined poolers get a distinct QUARANTINED role (visible in status) + // but are handled by quarantine remediation, not routed. + if result.Roles["quarantined"] != "QUARANTINED" { + t.Errorf("expected QUARANTINED, got %s", result.Roles["quarantined"]) } }) diff --git a/pkg/resource-handler/controller/shard/reconcile_data_plane.go b/pkg/resource-handler/controller/shard/reconcile_data_plane.go index cce9db8d..4c07db23 100644 --- a/pkg/resource-handler/controller/shard/reconcile_data_plane.go +++ b/pkg/resource-handler/controller/shard/reconcile_data_plane.go @@ -62,6 +62,24 @@ func (r *ShardReconciler) reconcileDataPlane( childSpan.End() } + // Phase: Remediate quarantined (unrecoverable) poolers by replacing the pod + // and wiping its data PVC so it re-bootstraps from backup. Runs before the + // drain state machine: a quarantined node is already down, so replacing it is + // the priority disruptive action this cycle. + { + _, childSpan := monitoring.StartChildSpan(ctx, "Shard.ReconcileQuarantineRemediation") + acted, err := r.reconcileQuarantineRemediation(ctx, store, shard) + if err != nil { + monitoring.RecordSpanError(childSpan, err) + childSpan.End() + return ctrl.Result{}, err + } + childSpan.End() + if acted { + return ctrl.Result{RequeueAfter: quarantineRemediationRequeue}, nil + } + } + // Phase: Execute drain state machine for pods with drain annotations { _, childSpan := monitoring.StartChildSpan(ctx, "Shard.ReconcileDrainState") diff --git a/pkg/resource-handler/controller/shard/reconcile_pool_pods.go b/pkg/resource-handler/controller/shard/reconcile_pool_pods.go index 366904c5..5897dfad 100644 --- a/pkg/resource-handler/controller/shard/reconcile_pool_pods.go +++ b/pkg/resource-handler/controller/shard/reconcile_pool_pods.go @@ -333,7 +333,10 @@ func isPoolHealthy( if idx, ok := resolvePodIndex(pod.Name); !ok || idx >= int(effectiveReplicas) { continue } - if resolvePodRole(shard, pod.Name) == "DRAINED" { + // 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" { continue } if !isPodReady(pod) { diff --git a/pkg/resource-handler/controller/shard/reconcile_quarantine.go b/pkg/resource-handler/controller/shard/reconcile_quarantine.go new file mode 100644 index 00000000..d14fc6eb --- /dev/null +++ b/pkg/resource-handler/controller/shard/reconcile_quarantine.go @@ -0,0 +1,272 @@ +package shard + +import ( + "context" + "fmt" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" + + "github.com/multigres/multigres/go/common/topoclient" + + multigresv1alpha1 "github.com/multigres/multigres-operator/api/v1alpha1" + "github.com/multigres/multigres-operator/pkg/data-handler/backuphealth" + "github.com/multigres/multigres-operator/pkg/data-handler/topo" + "github.com/multigres/multigres-operator/pkg/util/metadata" + "github.com/multigres/multigres-operator/pkg/util/status" +) + +const ( + // quarantineRemediationRequeue is how soon to requeue after taking a + // quarantine-remediation action, so the recreate converges and the result is + // observed promptly (the operator does not watch topology). + quarantineRemediationRequeue = 5 * time.Second + + // quarantineRemediationMinPodAge is how long a pod must have existed before + // it is eligible for quarantine remediation. A freshly (re)created pod + // re-registers its topology record within seconds, so requiring a minimum age + // avoids acting on a stale QUARANTINED record left by the prior process and + // wiping a pod that is actually mid-bootstrap. + // + // There is deliberately no quarantine-specific attempt-cap or terminal + // "unrecoverable" state: a reconstructed pod is just a new standby, so if it + // repeatedly fails to come up (e.g. a corrupt backup) that is a general + // standby bring-up / restore failure — surfaced by backup health and + // standby-provisioning monitoring — independent of whether the pod was ever + // quarantined. This age-based pacing only avoids tight re-wipe loops. + quarantineRemediationMinPodAge = 3 * time.Minute +) + +// reconcileQuarantineRemediation replaces poolers that have self-quarantined +// (LIFECYCLE_QUARANTINED: postgres is unrecoverably failing to start, e.g. a +// genuinely diverged standby) by deleting the backing pod AND hard-deleting its +// data PVC. The next reconcile's createMissingResources recreates both; the +// fresh, empty data volume makes pgctld re-bootstrap from backup — the only +// remediation that heals genuine on-disk divergence, since a same-PVC restart +// FATAL-loops identically. +// +// Gating is conservative: +// - at most one pod per reconcile (destructive action), +// - never the primary (leave primary loss to failover), +// - only when the rest of the pool is otherwise healthy (don't pile on during +// a broader outage), and +// - only for pods old enough to rule out a stale topology record. +// +// 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, + shard *multigresv1alpha1.Shard, +) (bool, error) { + logger := log.FromContext(ctx) + + lbls := map[string]string{ + metadata.LabelMultigresCluster: shard.Labels[metadata.LabelMultigresCluster], + metadata.LabelMultigresDatabase: string(shard.Spec.DatabaseName), + metadata.LabelMultigresTableGroup: string(shard.Spec.TableGroupName), + metadata.LabelMultigresShard: string(shard.Spec.ShardName), + } + + podList := &corev1.PodList{} + inNamespace := client.InNamespace(shard.Namespace) + matchingLabels := client.MatchingLabels(lbls) + if err := r.List(ctx, podList, inNamespace, matchingLabels); err != nil { + return false, fmt.Errorf("failed to list pods for quarantine remediation: %w", err) + } + + pods := make(map[string]*corev1.Pod, len(podList.Items)) + podNames := make([]string, 0, len(podList.Items)) + for i := range podList.Items { + p := &podList.Items[i] + pods[p.Name] = p + podNames = append(podNames, p.Name) + } + + quarantined, err := topo.GetQuarantinedPods(ctx, store, shard, podNames) + if err != nil { + // Topology hiccup: skip this cycle and retry on the next reconcile rather + // than failing the whole data-plane phase. + logger.Error(err, "Failed to list quarantined poolers; skipping remediation this cycle") + return false, nil + } + if len(quarantined) == 0 { + return false, nil + } + + // Safety gate: never wipe a node's data PVC unless a healthy backup exists to + // restore from — the hard-delete is irreversible, so without a good backup it + // would destroy the last copy. Uses the backup-health condition computed on + // the previous reconcile (this phase runs before backuphealth.Evaluate); an + // absent/false condition conservatively blocks remediation. + if !status.IsConditionTrue(shard.Status.Conditions, backuphealth.ConditionHealthy) { + logger.Info( + "Deferring quarantine remediation: no healthy backup to restore from", + "quarantinedPods", len(quarantined), + ) + r.Recorder.Eventf( + shard, "Warning", "QuarantineRemediationBlocked", + "Deferring replacement of %d quarantined pod(s): no healthy backup to restore from", + len(quarantined), + ) + return false, nil + } + + quarantinedSet := make(map[string]bool, len(quarantined)) + for _, q := range quarantined { + quarantinedSet[q.PodName] = true + } + + // quarantined is sorted by pod name, so remediation is deterministic and + // one-at-a-time. + for _, q := range quarantined { + podName := q.PodName + pod, ok := pods[podName] + if !ok || !pod.DeletionTimestamp.IsZero() { + continue // already gone or being deleted + } + + // Never wipe the primary; a quarantined node's postgres is down so it + // should not be primary, but guard defensively. + if shard.Status.PodRoles[podName] == "PRIMARY" { + logger.Info( + "Skipping quarantine remediation for pod currently marked PRIMARY", + "pod", podName, + ) + continue + } + + // Guard against a stale QUARANTINED record on a freshly recreated pod. + if age := time.Since(pod.CreationTimestamp.Time); age < quarantineRemediationMinPodAge { + logger.V(1).Info( + "Deferring quarantine remediation: pod too young to rule out a stale record", + "pod", podName, "age", age.Round(time.Second), + ) + continue + } + + // Don't pile on during a broader outage: require the rest of the pool + // (excluding other quarantined pods, which are themselves remediated) to + // be healthy before removing this one. + if !poolHealthyExcluding(pods, pod, quarantinedSet) { + logger.Info( + "Deferring quarantine remediation: pool not otherwise healthy", + "pod", podName, + ) + r.Recorder.Eventf(shard, "Warning", "QuarantineRemediationDeferred", + "Deferring replacement of quarantined pod %s: pool has other non-ready pods", + podName) + continue + } + + if err := r.wipeQuarantinedPod(ctx, shard, pod, q.Reason); err != nil { + return false, err + } + return true, nil // one destructive action per reconcile + } + + return false, nil +} + +// wipeQuarantinedPod deletes a quarantined pod and hard-deletes its data PVC so +// the recreated pod bootstraps from backup on an empty volume. Unlike +// scale-down cleanup (which may orphan a PVC for later reuse), the data here is +// known-bad and must be discarded. +func (r *ShardReconciler) wipeQuarantinedPod( + ctx context.Context, + shard *multigresv1alpha1.Shard, + pod *corev1.Pod, + reason string, +) error { + logger := log.FromContext(ctx) + + poolName := pod.Labels[metadata.LabelMultigresPool] + cellName := pod.Labels[metadata.LabelMultigresCell] + idx, ok := resolvePodIndex(pod.Name) + if !ok { + logger.Info( + "Skipping quarantine remediation for pod with unparseable index", + "pod", pod.Name, + ) + return nil + } + + if reason == "" { + reason = "unspecified" + } + logger.Info( + "Quarantine remediation: replacing unrecoverable pooler (delete pod + wipe PVC)", + "pod", pod.Name, "pool", poolName, "cell", cellName, "reason", reason, + ) + r.Recorder.Eventf(shard, "Warning", "QuarantineRemediation", + "Replacing quarantined pod %s (reason: %s): wiping data PVC to re-bootstrap from backup", + pod.Name, reason) + + // Delete the pod first so the data PVC can be released. + if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("failed to delete quarantined pod %s: %w", pod.Name, err) + } + + pvcName := BuildPoolDataPVCName(shard, poolName, cellName, idx) + pvc := &corev1.PersistentVolumeClaim{} + key := client.ObjectKey{Namespace: pod.Namespace, Name: pvcName} + if err := r.Get(ctx, key, pvc); err != nil { + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf( + "failed to fetch data PVC %s for quarantined pod %s: %w", + pvcName, pod.Name, err, + ) + } + if err := r.Delete(ctx, pvc); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf( + "failed to delete data PVC %s for quarantined pod %s: %w", + pvcName, pod.Name, err, + ) + } + logger.Info( + "Quarantine remediation: deleted pod and data PVC", + "pod", pod.Name, "pvc", pvcName, + ) + return nil +} + +// poolHealthyExcluding reports whether every other pod in the target's pool+cell +// is Ready, excluding the target itself, other quarantined pods (which are +// separately remediated), and pods already draining or terminating. It is the +// "don't remediate during a broader outage" gate. +func poolHealthyExcluding( + pods map[string]*corev1.Pod, + target *corev1.Pod, + quarantinedSet map[string]bool, +) bool { + pool := target.Labels[metadata.LabelMultigresPool] + cell := target.Labels[metadata.LabelMultigresCell] + for name, pod := range pods { + if name == target.Name || quarantinedSet[name] { + continue + } + if pod.Labels[metadata.LabelMultigresPool] != pool || + pod.Labels[metadata.LabelMultigresCell] != cell { + continue + } + if !pod.DeletionTimestamp.IsZero() || + pod.Annotations[metadata.AnnotationDrainState] != "" { + continue + } + if !isPodReady(pod) { + return false + } + } + return true +} diff --git a/pkg/resource-handler/controller/shard/reconcile_quarantine_internal_test.go b/pkg/resource-handler/controller/shard/reconcile_quarantine_internal_test.go new file mode 100644 index 00000000..2237b368 --- /dev/null +++ b/pkg/resource-handler/controller/shard/reconcile_quarantine_internal_test.go @@ -0,0 +1,325 @@ +package shard + +import ( + "context" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/multigres/multigres/go/common/topoclient" + "github.com/multigres/multigres/go/common/topoclient/memorytopo" + clustermetadata "github.com/multigres/multigres/go/pb/clustermetadata" + + multigresv1alpha1 "github.com/multigres/multigres-operator/api/v1alpha1" + "github.com/multigres/multigres-operator/pkg/data-handler/backuphealth" + "github.com/multigres/multigres-operator/pkg/util/metadata" +) + +// markBackupHealthy sets the shard's backup-health condition to True so +// quarantine remediation's backup-health gate is satisfied. +func markBackupHealthy(shard *multigresv1alpha1.Shard) { + shard.Status.Conditions = []metav1.Condition{{ + Type: backuphealth.ConditionHealthy, + Status: metav1.ConditionTrue, + Reason: "Healthy", + LastTransitionTime: metav1.Now(), + }} +} + +const ( + qrCell = "cell1" + qrPool = "primary" + qrNS = "default" + qrReason = "postgres failed to recover for 5m0s across 60 attempts" +) + +func qrShard() *multigresv1alpha1.Shard { + return &multigresv1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-shard", + Namespace: qrNS, + Labels: map[string]string{metadata.LabelMultigresCluster: "test-cluster"}, + }, + Spec: multigresv1alpha1.ShardSpec{ + DatabaseName: "testdb", + TableGroupName: "default", + ShardName: "shard0", + Pools: map[multigresv1alpha1.PoolName]multigresv1alpha1.PoolSpec{ + multigresv1alpha1.PoolName(qrPool): {Cells: []multigresv1alpha1.CellName{qrCell}}, + }, + }, + } +} + +func qrPod(shard *multigresv1alpha1.Shard, idx int, created time.Time, ready bool) *corev1.Pod { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: BuildPoolPodName(shard, qrPool, qrCell, idx), + Namespace: qrNS, + Labels: buildPoolLabelsWithCell(shard, qrPool, qrCell), + CreationTimestamp: metav1.NewTime(created), + }, + } + if ready { + pod.Status.Conditions = []corev1.PodCondition{ + {Type: corev1.PodReady, Status: corev1.ConditionTrue}, + } + } + return pod +} + +func qrPVC(shard *multigresv1alpha1.Shard, idx int) *corev1.PersistentVolumeClaim { + return &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: BuildPoolDataPVCName(shard, qrPool, qrCell, idx), + Namespace: qrNS, + Labels: buildPoolLabelsWithCell(shard, qrPool, qrCell), + }, + } +} + +// qrStoreWithQuarantined returns a memory topo store with the pooler backing the +// given pod indexes registered as LIFECYCLE_QUARANTINED (reason=qrReason), plus a +// healthy primary. +func qrStoreWithQuarantined( + t *testing.T, + shard *multigresv1alpha1.Shard, + quarantinedIdxs ...int, +) topoclient.Store { + return qrStoreWithQuarantinedReason(t, shard, qrReason, quarantinedIdxs...) +} + +// qrStoreWithQuarantinedReason is qrStoreWithQuarantined with an explicit reason +// on the quarantined records (use "" to exercise the empty-reason fallback). +func qrStoreWithQuarantinedReason( + t *testing.T, + shard *multigresv1alpha1.Shard, + reason string, + quarantinedIdxs ...int, +) topoclient.Store { + t.Helper() + _, factory := memorytopo.NewServerAndFactory(context.Background(), qrCell) + store := topoclient.NewWithFactory(factory, "", []string{""}, topoclient.NewDefaultTopoConfig()) + t.Cleanup(func() { _ = store.Close() }) + + q := map[int]bool{} + for _, i := range quarantinedIdxs { + q[i] = true + } + for idx := 0; idx < 2; idx++ { + name := BuildPoolPodName(shard, qrPool, qrCell, idx) + mp := &clustermetadata.Multipooler{ + Id: &clustermetadata.ID{Cell: qrCell, Name: name}, + Hostname: name, + RoutingState: &clustermetadata.RoutingState{ + Role: clustermetadata.RoutingRole_ROUTING_ROLE_REPLICA, + }, + ShardKey: &clustermetadata.ShardKey{ + Database: "testdb", TableGroup: "default", Shard: "shard0", + }, + } + if q[idx] { + mp.LifecycleStatus = &clustermetadata.PoolerLifecycle{ + Status: clustermetadata.PoolerLifecycleStatus_LIFECYCLE_QUARANTINED, + Reason: reason, + } + } + if err := store.RegisterMultipooler(context.Background(), mp, false); err != nil { + t.Fatalf("register pooler %s: %v", name, err) + } + } + return store +} + +func qrReconciler(t *testing.T, objs ...client.Object) *ShardReconciler { + t.Helper() + scheme := runtime.NewScheme() + _ = multigresv1alpha1.AddToScheme(scheme) + _ = corev1.AddToScheme(scheme) + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(&multigresv1alpha1.Shard{}). + WithObjects(objs...). + Build() + return &ShardReconciler{Client: c, Scheme: scheme, Recorder: record.NewFakeRecorder(20)} +} + +func exists(t *testing.T, r *ShardReconciler, obj client.Object, key client.ObjectKey) bool { + t.Helper() + err := r.Get(context.Background(), key, obj) + if err == nil { + return true + } + if apierrors.IsNotFound(err) { + return false + } + t.Fatalf("unexpected get error for %s: %v", key, err) + return false +} + +func TestReconcileQuarantineRemediation(t *testing.T) { + old := time.Now().Add(-1 * time.Hour) + + t.Run("wipes a quarantined replica when pool healthy", func(t *testing.T) { + shard := qrShard() + markBackupHealthy(shard) + badPod := qrPod(shard, 1, old, false) // quarantined replica, old, not ready + goodPod := qrPod(shard, 0, old, true) // healthy sibling + badPVC := qrPVC(shard, 1) + r := qrReconciler(t, shard, badPod, goodPod, badPVC) + store := qrStoreWithQuarantined(t, shard, 1) + + acted, err := r.reconcileQuarantineRemediation(context.Background(), store, shard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !acted { + t.Fatal("expected remediation to act on the quarantined pod") + } + if exists(t, r, &corev1.Pod{}, client.ObjectKeyFromObject(badPod)) { + t.Error("expected quarantined pod to be deleted") + } + if exists(t, r, &corev1.PersistentVolumeClaim{}, client.ObjectKeyFromObject(badPVC)) { + t.Error("expected quarantined pod's data PVC to be deleted (wiped)") + } + if !exists(t, r, &corev1.Pod{}, client.ObjectKeyFromObject(goodPod)) { + t.Error("healthy sibling pod should be untouched") + } + + // The remediation event should carry the topology quarantine reason. + rec := r.Recorder.(*record.FakeRecorder) + foundReason := false + for len(rec.Events) > 0 { + if ev := <-rec.Events; strings.Contains(ev, qrReason) { + foundReason = true + } + } + if !foundReason { + t.Errorf("expected a remediation event containing the quarantine reason %q", qrReason) + } + }) + + t.Run("wipes with an empty reason -> event falls back to 'unspecified'", func(t *testing.T) { + shard := qrShard() + markBackupHealthy(shard) + badPod := qrPod(shard, 1, old, false) + goodPod := qrPod(shard, 0, old, true) + badPVC := qrPVC(shard, 1) + r := qrReconciler(t, shard, badPod, goodPod, badPVC) + store := qrStoreWithQuarantinedReason(t, shard, "", 1) // no reason recorded + + acted, err := r.reconcileQuarantineRemediation(context.Background(), store, shard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !acted { + t.Fatal("expected remediation to act even without a recorded reason") + } + if exists(t, r, &corev1.Pod{}, client.ObjectKeyFromObject(badPod)) { + t.Error("expected quarantined pod to be deleted") + } + + rec := r.Recorder.(*record.FakeRecorder) + foundUnspecified := false + for len(rec.Events) > 0 { + if ev := <-rec.Events; strings.Contains(ev, "reason: unspecified") { + foundUnspecified = true + } + } + if !foundUnspecified { + t.Error("expected the event to fall back to 'unspecified' on empty reason") + } + }) + + t.Run("defers when the pod is too young (stale-record guard)", func(t *testing.T) { + shard := qrShard() + markBackupHealthy(shard) + badPod := qrPod(shard, 1, time.Now(), false) // just created + goodPod := qrPod(shard, 0, old, true) + badPVC := qrPVC(shard, 1) + r := qrReconciler(t, shard, badPod, goodPod, badPVC) + store := qrStoreWithQuarantined(t, shard, 1) + + acted, err := r.reconcileQuarantineRemediation(context.Background(), store, shard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if acted { + t.Error("expected no action for a too-young pod") + } + if !exists(t, r, &corev1.Pod{}, client.ObjectKeyFromObject(badPod)) { + t.Error("young quarantined pod must not be deleted yet") + } + if !exists(t, r, &corev1.PersistentVolumeClaim{}, client.ObjectKeyFromObject(badPVC)) { + t.Error("young quarantined pod's PVC must not be deleted yet") + } + }) + + t.Run("defers when another pod in the pool is unhealthy", func(t *testing.T) { + shard := qrShard() + markBackupHealthy(shard) + badPod := qrPod(shard, 1, old, false) // quarantined + sickPod := qrPod(shard, 0, old, false) // non-quarantined but not ready + badPVC := qrPVC(shard, 1) + r := qrReconciler(t, shard, badPod, sickPod, badPVC) + store := qrStoreWithQuarantined(t, shard, 1) // only idx 1 quarantined + + acted, err := r.reconcileQuarantineRemediation(context.Background(), store, shard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if acted { + t.Error("expected no action while another pool pod is unhealthy") + } + if !exists(t, r, &corev1.Pod{}, client.ObjectKeyFromObject(badPod)) { + t.Error("quarantined pod must not be wiped during a broader outage") + } + }) + + t.Run("defers when no healthy backup exists (never wipe the last copy)", func(t *testing.T) { + shard := qrShard() // no backup-health condition => not healthy + badPod := qrPod(shard, 1, old, false) + goodPod := qrPod(shard, 0, old, true) + badPVC := qrPVC(shard, 1) + r := qrReconciler(t, shard, badPod, goodPod, badPVC) + store := qrStoreWithQuarantined(t, shard, 1) + + acted, err := r.reconcileQuarantineRemediation(context.Background(), store, shard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if acted { + t.Error("expected no action when there is no healthy backup to restore from") + } + if !exists(t, r, &corev1.Pod{}, client.ObjectKeyFromObject(badPod)) { + t.Error("must not delete the pod without a healthy backup") + } + if !exists(t, r, &corev1.PersistentVolumeClaim{}, client.ObjectKeyFromObject(badPVC)) { + t.Error("must not wipe the data PVC without a healthy backup") + } + }) + + t.Run("no quarantined poolers -> no action", func(t *testing.T) { + shard := qrShard() + p0 := qrPod(shard, 0, old, true) + p1 := qrPod(shard, 1, old, true) + r := qrReconciler(t, shard, p0, p1) + store := qrStoreWithQuarantined(t, shard) // none quarantined + + acted, err := r.reconcileQuarantineRemediation(context.Background(), store, shard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if acted { + t.Error("expected no action when nothing is quarantined") + } + }) +}