diff --git a/pkg/data-handler/posture/cohort.go b/pkg/data-handler/posture/cohort.go new file mode 100644 index 00000000..1cebe0d3 --- /dev/null +++ b/pkg/data-handler/posture/cohort.go @@ -0,0 +1,90 @@ +package posture + +import ( + "context" + "errors" + "fmt" + + "github.com/multigres/multigres/go/common/rpcclient" + "github.com/multigres/multigres/go/common/topoclient" + clustermetadatapb "github.com/multigres/multigres/go/pb/clustermetadata" + multipoolermanagerdatapb "github.com/multigres/multigres/go/pb/multipoolermanagerdata" + "google.golang.org/protobuf/proto" + + multigresv1alpha1 "github.com/multigres/multigres-operator/api/v1alpha1" + "github.com/multigres/multigres-operator/pkg/data-handler/topo" +) + +// ErrConfirmedCohortMember means the authoritative committed rule still +// contains the target pooler. +var ErrConfirmedCohortMember = errors.New("pooler is a confirmed cohort member") + +// CheckCohortAbsence confirms that poolerID is absent from the authoritative +// committed cohort. Missing or conflicting evidence blocks deletion. +func CheckCohortAbsence( + ctx context.Context, + store topoclient.Store, + rpc rpcclient.MultipoolerClient, + shard *multigresv1alpha1.Shard, + poolerID *clustermetadatapb.ID, +) error { + var leader *clustermetadatapb.ID + var rule *clustermetadatapb.ShardRule + for _, cell := range topo.CollectCells(shard) { + poolers, err := store.GetMultipoolersByCell(ctx, cell, topo.ShardFilter(shard)) + if err != nil { + return fmt.Errorf("observe cell %s: %w", cell, err) + } + for _, pooler := range poolers { + if pooler.Id == nil || !topo.IsPrimaryPooler(pooler.Multipooler) { + continue + } + rpcCtx, cancel := context.WithTimeout(ctx, statusRPCTimeout) + resp, err := rpc.Status( + rpcCtx, + pooler.Multipooler, + &multipoolermanagerdatapb.StatusRequest{}, + ) + cancel() + if err != nil { + continue // An unreachable pooler cannot attest to the rule. + } + if !proto.Equal(resp.GetConsensusStatus().GetId(), pooler.Id) { + continue // Do not trust an identity mismatch. + } + if resp.GetStatus().GetPostgresStatus() != + multipoolermanagerdatapb.PostgresStatus_POSTGRES_STATUS_PRIMARY { + continue + } + position := resp.GetConsensusStatus().GetCurrentPosition().GetPosition() + if position.GetProposal() != nil { + return fmt.Errorf("primary has an unsettled consensus rule") + } + candidateRule := position.GetDecision() + if !proto.Equal(candidateRule.GetLeaderId(), pooler.Id) { + continue + } + if leader != nil { + return fmt.Errorf("primary observations disagree") + } + leader = pooler.Id + rule = candidateRule + } + } + if leader == nil || rule.GetRuleNumber() == nil { + return fmt.Errorf("cannot establish a committed rule from reachable poolers") + } + // Match on Cell and Name (the pooler's service ID), Component is ignored so + // a caller-built ID cannot silently miss a real member. + for _, member := range rule.GetCohortMembers() { + if member.GetCell() == poolerID.GetCell() && member.GetName() == poolerID.GetName() { + return fmt.Errorf( + "%w: pooler %s under rule %v", + ErrConfirmedCohortMember, + topoclient.ClusterIDString(poolerID), + rule.GetRuleNumber(), + ) + } + } + return nil +} diff --git a/pkg/data-handler/posture/cohort_test.go b/pkg/data-handler/posture/cohort_test.go new file mode 100644 index 00000000..d994e152 --- /dev/null +++ b/pkg/data-handler/posture/cohort_test.go @@ -0,0 +1,416 @@ +package posture_test + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + "github.com/multigres/multigres/go/common/rpcclient" + "github.com/multigres/multigres/go/common/topoclient" + "github.com/multigres/multigres/go/pb/clustermetadata" + "github.com/multigres/multigres/go/pb/multipoolermanagerdata" + + "github.com/multigres/multigres-operator/pkg/data-handler/posture" +) + +// withCommittedStatus is like withStatus but also sets a RuleNumber and +// LeaderId, which CheckCohortAbsence requires to consider a rule committed +// and its reporter authoritative (withStatus omits them since posture.Evaluate +// does not consult them). leaderID defaults to mp.Id when nil, letting callers +// build a stale/split-brain rule that names a different leader. +func withCommittedStatus( + rpc *rpcclient.FakeClient, + mp *topoclient.MultipoolerInfo, + s multipoolermanagerdata.PostgresStatus, + leaderID *clustermetadata.ID, + cohortMembers []*clustermetadata.ID, +) { + if leaderID == nil { + leaderID = mp.Id + } + rpc.SetStatusResponse( + topoclient.ComponentIDString(mp.Id), + &multipoolermanagerdata.StatusResponse{ + Status: &multipoolermanagerdata.Status{PostgresStatus: s}, + ConsensusStatus: &clustermetadata.ConsensusStatus{ + Id: mp.Id, + CurrentPosition: &clustermetadata.PoolerPosition{ + Position: &clustermetadata.RulePosition{ + Decision: &clustermetadata.ShardRule{ + RuleNumber: &clustermetadata.RuleNumber{CoordinatorTerm: 1}, + LeaderId: leaderID, + CohortMembers: cohortMembers, + }, + }, + }, + }, + }, + ) +} + +func TestCheckCohortAbsence(t *testing.T) { + t.Parallel() + + t.Run("pooler absent from committed rule is safe to delete", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primary}, nil + }, + } + rpc := rpcclient.NewFakeClient() + withCommittedStatus( + rpc, primary, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{primary.Id}, + ) + + gone := &clustermetadata.ID{Cell: "cell1", Name: "already-deleted-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, gone, + ); err != nil { + t.Fatalf("expected nil, got %v", err) + } + }) + + t.Run("pooler still a committed cohort member blocks deletion", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primary}, nil + }, + } + rpc := rpcclient.NewFakeClient() + withCommittedStatus( + rpc, primary, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{primary.Id}, + ) + err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, primary.Id, + ) + if !errors.Is(err, posture.ErrConfirmedCohortMember) { + t.Fatalf("expected confirmed-member error, got %v", err) + } + }) + + t.Run("pending proposal blocks absence confirmation", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primary}, nil + }, + } + rpc := rpcclient.NewFakeClient() + rpc.SetStatusResponse( + topoclient.ComponentIDString(primary.Id), + &multipoolermanagerdata.StatusResponse{ + Status: &multipoolermanagerdata.Status{ + PostgresStatus: multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + }, + ConsensusStatus: &clustermetadata.ConsensusStatus{ + Id: primary.Id, + CurrentPosition: &clustermetadata.PoolerPosition{ + Position: &clustermetadata.RulePosition{ + Decision: &clustermetadata.ShardRule{ + RuleNumber: &clustermetadata.RuleNumber{CoordinatorTerm: 1}, + LeaderId: primary.Id, + CohortMembers: []*clustermetadata.ID{primary.Id}, + }, + Proposal: &clustermetadata.ShardRule{ + RuleNumber: &clustermetadata.RuleNumber{CoordinatorTerm: 2}, + LeaderId: primary.Id, + }, + }, + }, + }, + }, + ) + + gone := &clustermetadata.ID{Cell: "cell1", Name: "already-deleted-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, gone, + ); err == nil { + t.Fatal("expected pending proposal to block absence confirmation") + } + }) + + t.Run("no reachable primary cannot establish a rule", func(t *testing.T) { + t.Parallel() + shard := testShard() + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return nil, nil + }, + } + rpc := rpcclient.NewFakeClient() + + id := &clustermetadata.ID{Cell: "cell1", Name: "some-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, id, + ); err == nil { + t.Fatal("expected error when no committed rule can be established, got nil") + } + }) + + t.Run("unreachable primary cannot establish a rule", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primary}, nil + }, + } + rpc := rpcclient.NewFakeClient() + rpc.Errors[topoclient.ComponentIDString(primary.Id)] = fmt.Errorf("fake rpc failure") + + id := &clustermetadata.ID{Cell: "cell1", Name: "some-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, id, + ); err == nil { + t.Fatal("expected error when the primary is unreachable, got nil") + } + }) + + t.Run("consensus identity mismatch is not trusted as leader", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primary}, nil + }, + } + rpc := rpcclient.NewFakeClient() + // Response claims a different consensus identity than the topology + // entry it was fetched for (e.g. a stale/reused RPC target). + imposter := &clustermetadata.ID{Cell: "cell1", Name: "someone-else"} + rpc.SetStatusResponse( + topoclient.ComponentIDString(primary.Id), + &multipoolermanagerdata.StatusResponse{ + Status: &multipoolermanagerdata.Status{ + PostgresStatus: multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + }, + ConsensusStatus: &clustermetadata.ConsensusStatus{ + Id: imposter, + CurrentPosition: &clustermetadata.PoolerPosition{ + Position: &clustermetadata.RulePosition{ + Decision: &clustermetadata.ShardRule{ + RuleNumber: &clustermetadata.RuleNumber{CoordinatorTerm: 1}, + LeaderId: primary.Id, + CohortMembers: []*clustermetadata.ID{primary.Id}, + }, + }, + }, + }, + }, + ) + + id := &clustermetadata.ID{Cell: "cell1", Name: "some-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, id, + ); err == nil { + t.Fatal("expected error: identity mismatch must not be trusted, got nil") + } + }) + + t.Run("topology role disagreement blocks authority", func(t *testing.T) { + t.Parallel() + shard := testShard() + // Reports PostgreSQL PRIMARY over RPC, but topology still has it as a + // REPLICA — a stale/self-promoted pooler must not authorize deletion. + staleReplica := poolerInfo( + "stale-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_REPLICA, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{staleReplica}, nil + }, + } + rpc := rpcclient.NewFakeClient() + withCommittedStatus( + rpc, staleReplica, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{staleReplica.Id}, + ) + + id := &clustermetadata.ID{Cell: "cell1", Name: "some-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, id, + ); err == nil { + t.Fatal( + "expected error: topology role disagreement must not authorize deletion, got nil", + ) + } + }) + + t.Run("stale rule naming a different leader blocks authority", func(t *testing.T) { + t.Parallel() + shard := testShard() + stale := poolerInfo( + "stale-primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{stale}, nil + }, + } + rpc := rpcclient.NewFakeClient() + otherLeader := &clustermetadata.ID{Cell: "cell1", Name: "actual-leader-pod"} + withCommittedStatus( + rpc, stale, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + otherLeader, []*clustermetadata.ID{otherLeader}, + ) + + id := &clustermetadata.ID{Cell: "cell1", Name: "some-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, id, + ); err == nil { + t.Fatal("expected error: stale rule.LeaderId mismatch must block authority, got nil") + } + }) + + t.Run("split-brain: two authoritative primaries disagree", func(t *testing.T) { + t.Parallel() + shard := testShard() + primaryA := poolerInfo( + "pod-a", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + primaryB := poolerInfo( + "pod-b", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primaryA, primaryB}, nil + }, + } + rpc := rpcclient.NewFakeClient() + withCommittedStatus( + rpc, primaryA, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{primaryA.Id}, + ) + withCommittedStatus( + rpc, primaryB, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{primaryB.Id}, + ) + + id := &clustermetadata.ID{Cell: "cell1", Name: "some-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, id, + ); err == nil { + t.Fatal("expected error on split-brain (two disagreeing primaries), got nil") + } + }) + + t.Run("membership matches real topology IDs on cell and name", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + return []*topoclient.MultipoolerInfo{primary}, nil + }, + } + // Real members carry Component and a service-ID name, a caller rebuilding the ID must still match them. + member := &clustermetadata.ID{ + Component: clustermetadata.ID_MULTIPOOLER, + Cell: "cell1", + Name: "p-0badcafe", + } + rpc := rpcclient.NewFakeClient() + withCommittedStatus( + rpc, primary, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{primary.Id, member}, + ) + + withoutComponent := &clustermetadata.ID{Cell: "cell1", Name: "p-0badcafe"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, withoutComponent, + ); err == nil { + t.Fatal("expected error: member must match regardless of Component, got nil") + } + otherCell := &clustermetadata.ID{Cell: "cell2", Name: "p-0badcafe"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, otherCell, + ); err != nil { + t.Fatalf("expected nil for same name in another cell, got %v", err) + } + }) + + t.Run("non-primary poolers are not queried", func(t *testing.T) { + t.Parallel() + shard := testShard() + primary := poolerInfo( + "primary-pod", + clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + deadReplica := poolerInfo( + "dead-replica", + clustermetadata.RoutingRole_ROUTING_ROLE_REPLICA, + clustermetadata.PoolerLifecycleStatus_LIFECYCLE_UNKNOWN, + ) + store := &mockTopoStore{ + getMultipoolersByCellFunc: func(context.Context, string, *topoclient.GetMultipoolersByCellOptions) ([]*topoclient.MultipoolerInfo, error) { + // Unreachable replica listed first must not consume the deadline. + return []*topoclient.MultipoolerInfo{deadReplica, primary}, nil + }, + } + rpc := rpcclient.NewFakeClient() + rpc.Errors[topoclient.ComponentIDString(deadReplica.Id)] = fmt.Errorf("unreachable") + withCommittedStatus( + rpc, primary, multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + nil, []*clustermetadata.ID{primary.Id}, + ) + + gone := &clustermetadata.ID{Cell: "cell1", Name: "already-deleted-pod"} + if err := posture.CheckCohortAbsence( + context.Background(), store, rpc, shard, gone, + ); err != nil { + t.Fatalf("expected nil, got %v", err) + } + for _, call := range rpc.GetCallLog() { + if strings.Contains(call, string(topoclient.ComponentIDString(deadReplica.Id))) { + t.Fatalf("replica must not be queried, saw call %q", call) + } + } + }) +} diff --git a/pkg/resource-handler/controller/shard/disruption.go b/pkg/resource-handler/controller/shard/disruption.go index caefdd53..91cb9837 100644 --- a/pkg/resource-handler/controller/shard/disruption.go +++ b/pkg/resource-handler/controller/shard/disruption.go @@ -9,12 +9,21 @@ import ( corev1 "k8s.io/api/core/v1" "sigs.k8s.io/controller-runtime/pkg/client" + clustermetadatapb "github.com/multigres/multigres/go/pb/clustermetadata" + multigresv1alpha1 "github.com/multigres/multigres-operator/api/v1alpha1" "github.com/multigres/multigres-operator/pkg/data-handler/posture" "github.com/multigres/multigres-operator/pkg/util/metadata" ) -const disruptionRecoveryRequeue = 5 * time.Second +const ( + disruptionRecoveryRequeue = 5 * time.Second + + // pvcCleanupDeferralTimeout bounds how long an excess PVC's hard delete is + // deferred while cohort absence cannot be confirmed. Past it the PVC is + // orphaned (retention-window GC). + pvcCleanupDeferralTimeout = 10 * time.Minute +) // listDisruptionPods reads from the API server so a new reconcile cannot miss // a drain annotation that a previous reconcile just wrote through the cache. @@ -116,6 +125,35 @@ func (r *ShardReconciler) canStartDisruption( return true, nil } +// confirmPoolerNotInCohort re-observes the committed cohort before PVC deletion. +// Missing infrastructure or evidence blocks deletion. +func (r *ShardReconciler) confirmPoolerNotInCohort( + ctx context.Context, + shard *multigresv1alpha1.Shard, + podName, cellName string, +) error { + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + if r.PoolerClients == nil { + return fmt.Errorf("no pooler client available to confirm cohort membership") + } + rpc, err := r.PoolerClients.ClientFor(ctx, shard) + if err != nil || rpc == nil { + return fmt.Errorf("no pooler client available to confirm cohort membership: %w", err) + } + store, err := r.topoStore(ctx, shard) + if err != nil || store == nil { + return fmt.Errorf("no topology store available to confirm cohort membership: %w", err) + } + defer func() { _ = store.Close() }() + poolerID := &clustermetadatapb.ID{ + Component: clustermetadatapb.ID_MULTIPOOLER, + Cell: cellName, + Name: BuildPoolServiceID(podName), + } + return posture.CheckCohortAbsence(ctx, store, rpc, shard, poolerID) +} + // selectShardScaleDownPod ranks removable pods across pool/cell boundaries. // Active surges in other cells are retained by their maintenance workflow; // released local surges are already present in localExtras. diff --git a/pkg/resource-handler/controller/shard/reconcile_pool_pods.go b/pkg/resource-handler/controller/shard/reconcile_pool_pods.go index 030fa5c4..2782a963 100644 --- a/pkg/resource-handler/controller/shard/reconcile_pool_pods.go +++ b/pkg/resource-handler/controller/shard/reconcile_pool_pods.go @@ -2,6 +2,7 @@ package shard import ( "context" + stderrors "errors" "fmt" "slices" "strconv" @@ -18,6 +19,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/log" multigresv1alpha1 "github.com/multigres/multigres-operator/api/v1alpha1" + "github.com/multigres/multigres-operator/pkg/data-handler/posture" "github.com/multigres/multigres-operator/pkg/monitoring" "github.com/multigres/multigres-operator/pkg/util/metadata" pvcutil "github.com/multigres/multigres-operator/pkg/util/pvc" @@ -557,13 +559,25 @@ func (r *ShardReconciler) handleScaleDown( // Cleanup pods ready for deletion for _, pod := range readyForDeletion { logger.Info("Cleaning up pod in ready-for-deletion state", "pod", pod.Name) - if err := r.cleanupDrainedPod(ctx, shard, pod, poolName, poolSpec, replicas); err != nil { + deferred, err := r.cleanupDrainedPod( + ctx, shard, pod, poolName, poolSpec, replicas, disruptions, + ) + if err != nil { return actionTaken, inProgress, fmt.Errorf( "failed to cleanup drained pod %s: %w", pod.Name, err, ) } + if deferred { + // Retain the pod so a later reconcile can retry PVC cleanup. + logger.Info( + "Deferring pod deletion until PVC cleanup can be confirmed safe", + "pod", pod.Name, + ) + inProgress = true + continue + } if err := r.Delete(ctx, pod); err != nil && !errors.IsNotFound(err) { return actionTaken, inProgress, fmt.Errorf( "failed to delete ready-for-deletion pod %s: %w", @@ -973,6 +987,7 @@ func (r *ShardReconciler) syncDrainedLabels( return nil } +// cleanupDrainedPod reports whether the caller must retain the pod and retry. func (r *ShardReconciler) cleanupDrainedPod( ctx context.Context, shard *multigresv1alpha1.Shard, @@ -980,27 +995,31 @@ func (r *ShardReconciler) cleanupDrainedPod( poolName string, poolSpec multigresv1alpha1.PoolSpec, replicas int32, -) error { + tracker *shardRolloutTracker, +) (deferred bool, err 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. + // Use the pod label because the data-handler clears PodRoles during drain. if pod.Labels[metadata.LabelPodRole] == "DRAINED" { - if err := r.cleanupPodPVC( + deferred, err := r.cleanupPodPVC( ctx, shard, pod, poolName, "DRAINED (data known-bad)", - ); err != nil { - return err + true, + tracker, + ) + if err != nil { + return false, err + } + if deferred { + return true, nil } 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 + return false, nil } // For non-DRAINED pods, respect WhenScaled policy @@ -1020,8 +1039,14 @@ func (r *ShardReconciler) cleanupDrainedPod( if !idxOK { logger.Info("Skipping PVC deletion for pod with unparseable index", "pod", pod.Name) } else if idx >= int(replicas) { - if err := r.cleanupPodPVC(ctx, shard, pod, poolName, "scaled down"); err != nil { - return err + deferred, err := r.cleanupPodPVC( + ctx, shard, pod, poolName, "scaled down", true, tracker, + ) + if err != nil { + return false, err + } + if deferred { + return true, nil } } else { logger.Info( @@ -1040,27 +1065,25 @@ func (r *ShardReconciler) cleanupDrainedPod( logger.Info("Drained pod cleanup complete", "pod", pod.Name) r.Recorder.Eventf(shard, "Normal", "DrainCompleted", "Completed drain for pod %s", pod.Name) - return nil + return false, nil } -// cleanupPodPVC removes a pod's data PVC from the operator's care. The choice -// between orphaning (deferred deletion via multigres-gc) and in-line deletion -// is based on how many sibling PVCs remain in the same pool+cell: if removing -// this one still leaves >= pvcOrphanReplicasThreshold volumes, it is excess and -// is deleted, otherwise it is orphaned so the data can be recovered. See -// orphanByRemainingCount. +// cleanupPodPVC deletes excess PVCs and orphans the minimum retained set. +// A deferred result tells the caller to retain the pod and retry. func (r *ShardReconciler) cleanupPodPVC( ctx context.Context, shard *multigresv1alpha1.Shard, pod *corev1.Pod, poolName, reason string, -) error { + verifyCohort bool, + tracker *shardRolloutTracker, +) (deferred bool, err error) { logger := log.FromContext(ctx) idx, ok := resolvePodIndex(pod.Name) if !ok { logger.Info("Skipping PVC cleanup for pod with unparseable index", "pod", pod.Name) - return nil + return false, nil } cellName := pod.Labels[metadata.LabelMultigresCell] @@ -1072,32 +1095,140 @@ func (r *ShardReconciler) cleanupPodPVC( pvc, ); err != nil { if errors.IsNotFound(err) { - return nil + return false, nil } logger.Error(err, "Failed to fetch PVC for cleanup", "pvc", pvcName) - return fmt.Errorf("failed to fetch PVC %s for cleanup: %w", pvcName, err) + return false, fmt.Errorf("failed to fetch PVC %s for cleanup: %w", pvcName, err) } liveCount, err := r.countPoolCellPVCs(ctx, shard, poolName, cellName) if err != nil { - return err + return false, err } if orphanByRemainingCount(liveCount) { if err := pvcutil.MarkOrphan(ctx, r.Client, pvc, shard.GetUID(), time.Now()); err != nil { logger.Error(err, "Failed to mark PVC orphan for "+reason+" pod", "pvc", pvcName) - return fmt.Errorf("failed to mark PVC %s orphan: %w", pvcName, err) + return false, fmt.Errorf("failed to mark PVC %s orphan: %w", pvcName, err) } logger.Info("Marked PVC orphan for "+reason+" pod", "pvc", pvcName, "liveCount", liveCount) - return nil + return false, nil + } + + // Reconfirm cohort absence immediately before hard deletion. + if verifyCohort { + if err := r.confirmPoolerNotInCohort(ctx, shard, pod.Name, cellName); err != nil { + return r.deferOrFallbackPVCCleanup(ctx, shard, pod, pvc, err, tracker) + } } if err := r.Delete(ctx, pvc); err != nil && !errors.IsNotFound(err) { logger.Error(err, "Failed to delete PVC for "+reason+" pod", "pvc", pvcName) - return fmt.Errorf("failed to delete PVC %s: %w", pvcName, err) + return false, fmt.Errorf("failed to delete PVC %s: %w", pvcName, err) } logger.Info("Deleted PVC for "+reason+" pod", "pvc", pvcName, "liveCount", liveCount) - return nil + return false, nil +} + +// deferOrFallbackPVCCleanup handles a failed pre-delete cohort recheck. The +// first failure stamps the pod so later reconciles can measure how long the +// PVC has been waiting. Within pvcCleanupDeferralTimeout the cleanup is +// deferred (pod kept, reconcile requeued). Past it the operator stops waiting +// and orphans the PVC instead of hard-deleting it. +func (r *ShardReconciler) deferOrFallbackPVCCleanup( + ctx context.Context, + shard *multigresv1alpha1.Shard, + pod *corev1.Pod, + pvc *corev1.PersistentVolumeClaim, + cause error, + tracker *shardRolloutTracker, +) (deferred bool, err error) { + logger := log.FromContext(ctx) + if stderrors.Is(cause, posture.ErrConfirmedCohortMember) { + if _, ok := pod.Annotations[metadata.AnnotationPVCCleanupDeferredSince]; ok { + patch := client.MergeFrom(pod.DeepCopy()) + delete(pod.Annotations, metadata.AnnotationPVCCleanupDeferredSince) + if err := r.Patch(ctx, pod, patch); err != nil { + return false, fmt.Errorf( + "failed to clear PVC cleanup deferral on pod %s: %w", pod.Name, err, + ) + } + } + logger.Info( + "Deferring PVC deletion because the pooler remains a cohort member", + "pvc", pvc.Name, "reason", cause, + ) + r.Recorder.Eventf( + shard, + "Normal", + "PVCDeletionDeferred", + "Deferring PVC %s deletion while its pooler remains a cohort member: %v", + pvc.Name, + cause, + ) + if tracker != nil { + tracker.waitingForRecovery = true + } + return true, nil + } + + now := time.Now() + + since, err := time.Parse( + time.RFC3339, + pod.Annotations[metadata.AnnotationPVCCleanupDeferredSince], + ) + if err != nil { + patch := client.MergeFrom(pod.DeepCopy()) + if pod.Annotations == nil { + pod.Annotations = map[string]string{} + } + pod.Annotations[metadata.AnnotationPVCCleanupDeferredSince] = now.Format(time.RFC3339) + if err := r.Patch(ctx, pod, patch); err != nil { + return false, fmt.Errorf( + "failed to record PVC cleanup deferral on pod %s: %w", pod.Name, err, + ) + } + since = now + } + + if now.Sub(since) >= pvcCleanupDeferralTimeout { + if err := pvcutil.MarkOrphan(ctx, r.Client, pvc, shard.GetUID(), now); err != nil { + return false, fmt.Errorf("failed to mark PVC %s orphan: %w", pvc.Name, err) + } + logger.Info( + "Cohort absence unconfirmed past deadline; orphaned PVC instead of deleting", + "pvc", pvc.Name, "deferredSince", since, "reason", cause, + ) + r.Recorder.Eventf( + shard, + "Warning", + "PVCCleanupFallback", + "Could not confirm cohort absence for PVC %s within %s; "+ + "marked orphan for retention-window GC instead of deleting: %v", + pvc.Name, + pvcCleanupDeferralTimeout, + cause, + ) + return false, nil + } + + logger.Info( + "Deferring PVC deletion pending cohort membership confirmation", + "pvc", pvc.Name, "deferredSince", since, "reason", cause, + ) + r.Recorder.Eventf( + shard, + "Normal", + "PVCDeletionDeferred", + "Deferring PVC %s deletion until cohort membership can be confirmed: %v", + pvc.Name, + cause, + ) + if tracker != nil { + tracker.waitingForRecovery = true + } + return true, nil } // countPoolCellPVCs returns the number of PVCs currently present for the given 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..6c71c6ad 100644 --- a/pkg/resource-handler/controller/shard/shard_controller_internal_test.go +++ b/pkg/resource-handler/controller/shard/shard_controller_internal_test.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" "testing" + "time" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" @@ -28,7 +29,13 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" + "github.com/multigres/multigres/go/common/rpcclient" + "github.com/multigres/multigres/go/common/topoclient" + clustermetadata "github.com/multigres/multigres/go/pb/clustermetadata" + multipoolermanagerdata "github.com/multigres/multigres/go/pb/multipoolermanagerdata" + multigresv1alpha1 "github.com/multigres/multigres-operator/api/v1alpha1" + "github.com/multigres/multigres-operator/pkg/data-handler/poolerclient" "github.com/multigres/multigres-operator/pkg/data-handler/posture" "github.com/multigres/multigres-operator/pkg/testutil" "github.com/multigres/multigres-operator/pkg/util/metadata" @@ -1291,6 +1298,10 @@ func TestCleanupDrainedPod_PVCDeletion(t *testing.T) { _ = multigresv1alpha1.AddToScheme(scheme) _ = corev1.AddToScheme(scheme) + poolName := "primary" + cellName := "zone1" + replicas := int32(3) + baseShard := &multigresv1alpha1.Shard{ ObjectMeta: metav1.ObjectMeta{ Name: "test-shard", @@ -1303,12 +1314,52 @@ func TestCleanupDrainedPod_PVCDeletion(t *testing.T) { DatabaseName: "testdb", TableGroupName: "default", ShardName: "shard0", + Pools: map[multigresv1alpha1.PoolName]multigresv1alpha1.PoolSpec{ + multigresv1alpha1.PoolName(poolName): { + Cells: []multigresv1alpha1.CellName{multigresv1alpha1.CellName(cellName)}, + }, + }, }, } - poolName := "primary" - cellName := "zone1" - replicas := int32(3) + // Use an unrelated committed leader so hard-delete cases pass the cohort check. + leaderID := &clustermetadata.ID{ + Component: clustermetadata.ID_MULTIPOOLER, + Cell: cellName, + Name: "p-synthetic", + } + cohortStore := &disruptionTopo{ + poolers: []*topoclient.MultipoolerInfo{ + {Multipooler: &clustermetadata.Multipooler{ + Id: leaderID, + Hostname: leaderID.Name, + RoutingState: &clustermetadata.RoutingState{ + Role: clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + }, + }}, + }, + } + cohortRPC := rpcclient.NewFakeClient() + cohortRPC.SetStatusResponse( + topoclient.ComponentIDString(leaderID), + &multipoolermanagerdata.StatusResponse{ + Status: &multipoolermanagerdata.Status{ + PostgresStatus: multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + }, + ConsensusStatus: &clustermetadata.ConsensusStatus{ + Id: leaderID, + CurrentPosition: &clustermetadata.PoolerPosition{ + Position: &clustermetadata.RulePosition{ + Decision: &clustermetadata.ShardRule{ + RuleNumber: &clustermetadata.RuleNumber{CoordinatorTerm: 1}, + LeaderId: leaderID, + CohortMembers: []*clustermetadata.ID{leaderID}, + }, + }, + }, + }, + }, + ) podName0 := BuildPoolPodName(baseShard, poolName, cellName, 0) pvcName0 := BuildPoolDataPVCName(baseShard, poolName, cellName, 0) @@ -1439,18 +1490,24 @@ func TestCleanupDrainedPod_PVCDeletion(t *testing.T) { Build() reconciler := &ShardReconciler{ - Client: fakeClient, - Scheme: scheme, - Recorder: record.NewFakeRecorder(100), - APIReader: fakeClient, + Client: fakeClient, + Scheme: scheme, + Recorder: record.NewFakeRecorder(100), + APIReader: fakeClient, + PoolerClients: poolerclient.Static(cohortRPC), + CreateTopoStore: func(*multigresv1alpha1.Shard) (topoclient.Store, error) { + return cohortStore, nil + }, } poolSpec := multigresv1alpha1.PoolSpec{ PVCDeletionPolicy: tc.policy, } - err := reconciler.cleanupDrainedPod( - context.Background(), shard, pod, poolName, poolSpec, replicas, + _, err := reconciler.cleanupDrainedPod( + context.Background(), + shard, pod, poolName, poolSpec, replicas, + &shardRolloutTracker{}, ) if err != nil { t.Fatalf("cleanupDrainedPod() returned unexpected error: %v", err) @@ -1492,6 +1549,305 @@ func TestCleanupDrainedPod_PVCDeletion(t *testing.T) { } } +// cohortMemberFixture is a ready-for-deletion pod whose PVC sits on the +// hard-delete path (pool+cell over the orphan threshold) while its pooler is +// still listed in the committed cohort, so confirmPoolerNotInCohort fails. +type cohortMemberFixture struct { + reconciler *ShardReconciler + client client.Client + recorder *record.FakeRecorder + shard *multigresv1alpha1.Shard + pod *corev1.Pod + pvcName string + poolName string + poolSpec multigresv1alpha1.PoolSpec + tracker *shardRolloutTracker +} + +func newCohortMemberFixture(t *testing.T, mutatePod func(*corev1.Pod)) *cohortMemberFixture { + t.Helper() + scheme := runtime.NewScheme() + _ = multigresv1alpha1.AddToScheme(scheme) + _ = corev1.AddToScheme(scheme) + + poolName := "primary" + cellName := "zone1" + + shard := &multigresv1alpha1.Shard{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-shard", + Namespace: "default", + 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(poolName): { + Cells: []multigresv1alpha1.CellName{multigresv1alpha1.CellName(cellName)}, + }, + }, + }, + } + + podName := BuildPoolPodName(shard, poolName, cellName, 5) // index >= replicas + pvcName := BuildPoolDataPVCName(shard, poolName, cellName, 5) + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: "default", + Labels: map[string]string{metadata.LabelMultigresCell: cellName}, + Annotations: map[string]string{ + metadata.AnnotationDrainState: metadata.DrainStateReadyForDeletion, + }, + }, + } + if mutatePod != nil { + mutatePod(pod) + } + pvc := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: pvcName, + Namespace: "default", + Labels: buildPoolLabelsWithCell(shard, poolName, cellName), + }, + } + // Filler siblings so the pool+cell is over the orphan-retention threshold, + // putting pvc on the in-line (hard) delete path. + objs := []client.Object{shard, pod, pvc} + for i := 0; i < 3; i++ { + objs = append(objs, &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("filler-pvc-%d", i), + Namespace: "default", + Labels: buildPoolLabelsWithCell(shard, poolName, cellName), + }, + }) + } + fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build() + + // The current committed rule still contains the target pooler. Shape the + // ID exactly as the multipooler registers it: Component set and Name equal + // to its --service-id (BuildPoolServiceID), not the pod name. + poolerID := &clustermetadata.ID{ + Component: clustermetadata.ID_MULTIPOOLER, + Cell: cellName, + Name: BuildPoolServiceID(podName), + } + store := &disruptionTopo{ + poolers: []*topoclient.MultipoolerInfo{ + {Multipooler: &clustermetadata.Multipooler{ + Id: poolerID, + Hostname: podName, + RoutingState: &clustermetadata.RoutingState{ + Role: clustermetadata.RoutingRole_ROUTING_ROLE_PRIMARY, + }, + }}, + }, + } + rpc := rpcclient.NewFakeClient() + rpc.SetStatusResponse( + topoclient.ComponentIDString(poolerID), + &multipoolermanagerdata.StatusResponse{ + Status: &multipoolermanagerdata.Status{ + PostgresStatus: multipoolermanagerdata.PostgresStatus_POSTGRES_STATUS_PRIMARY, + }, + ConsensusStatus: &clustermetadata.ConsensusStatus{ + Id: poolerID, + CurrentPosition: &clustermetadata.PoolerPosition{ + Position: &clustermetadata.RulePosition{ + Decision: &clustermetadata.ShardRule{ + RuleNumber: &clustermetadata.RuleNumber{CoordinatorTerm: 1}, + LeaderId: poolerID, + CohortMembers: []*clustermetadata.ID{poolerID}, + }, + }, + }, + }, + }, + ) + + recorder := record.NewFakeRecorder(100) + reconciler := &ShardReconciler{ + Client: fakeClient, + Scheme: scheme, + Recorder: recorder, + APIReader: fakeClient, + PoolerClients: poolerclient.Static(rpc), + CreateTopoStore: func(*multigresv1alpha1.Shard) (topoclient.Store, error) { + return store, nil + }, + } + + return &cohortMemberFixture{ + reconciler: reconciler, + client: fakeClient, + recorder: recorder, + shard: shard, + pod: pod, + pvcName: pvcName, + poolName: poolName, + poolSpec: multigresv1alpha1.PoolSpec{ + PVCDeletionPolicy: &multigresv1alpha1.PVCDeletionPolicy{ + WhenScaled: multigresv1alpha1.DeletePVCRetentionPolicy, + }, + }, + tracker: &shardRolloutTracker{}, + } +} + +// runScaleDown drives the fixture through handleScaleDown, the caller whose +// contract (keep the pod while cleanup is deferred) is under test. +func (f *cohortMemberFixture) runScaleDown(t *testing.T) { + t.Helper() + existingPods := map[string]*corev1.Pod{f.pod.Name: f.pod} + if _, _, err := f.reconciler.handleScaleDown( + context.Background(), f.shard, f.poolName, f.poolSpec, existingPods, + 3, 3, false, f.tracker, + ); err != nil { + t.Fatalf("handleScaleDown() returned unexpected error: %v", err) + } +} + +func (f *cohortMemberFixture) getPVC(t *testing.T) (*corev1.PersistentVolumeClaim, error) { + t.Helper() + pvc := &corev1.PersistentVolumeClaim{} + err := f.client.Get( + context.Background(), + client.ObjectKey{Namespace: "default", Name: f.pvcName}, + pvc, + ) + return pvc, err +} + +func (f *cohortMemberFixture) getPod(t *testing.T) (*corev1.Pod, error) { + t.Helper() + pod := &corev1.Pod{} + err := f.client.Get( + context.Background(), + client.ObjectKey{Namespace: "default", Name: f.pod.Name}, + pod, + ) + return pod, err +} + +// TestCleanupPodPVC_DefersOnCohortMembership covers the hard-delete safety check. +func TestCleanupPodPVC_DefersOnCohortMembership(t *testing.T) { + f := newCohortMemberFixture(t, nil) + f.runScaleDown(t) + + gotPVC, err := f.getPVC(t) + if err != nil { + t.Fatalf("PVC must still exist (deletion deferred), got err: %v", err) + } + if _, orphaned := gotPVC.Labels[metadata.LabelOrphan]; orphaned { + t.Error("PVC must not be orphaned either — it is still a cohort member, not excess") + } + + gotPod, err := f.getPod(t) + if err != nil { + t.Fatalf( + "Pod must survive when its PVC cleanup is deferred, "+ + "or no later reconcile could retry cleanup: got err: %v", err, + ) + } + if since := gotPod.Annotations[metadata.AnnotationPVCCleanupDeferredSince]; since != "" { + t.Errorf("confirmed membership must not start the fallback timer, got %q", since) + } + + if !f.tracker.waitingForRecovery { + t.Error("expected tracker.waitingForRecovery to be set so the reconcile requeues") + } +} + +// TestCleanupPodPVC_FallsBackToOrphanAfterDeadline verifies the deferral is +// bounded: once the pod has been waiting longer than pvcCleanupDeferralTimeout +// the PVC is orphaned (retention-window GC) rather than hard-deleted, and the +// pod is released, so a persistent failure cannot wedge scale-down. +func TestCleanupPodPVC_FallsBackToOrphanAfterDeadline(t *testing.T) { + expired := time.Now().Add(-pvcCleanupDeferralTimeout - time.Minute) + f := newCohortMemberFixture(t, func(pod *corev1.Pod) { + pod.Annotations[metadata.AnnotationPVCCleanupDeferredSince] = expired.Format(time.RFC3339) + }) + f.reconciler.PoolerClients = nil // Keep the failure ambiguous so fallback is allowed. + f.runScaleDown(t) + + gotPVC, err := f.getPVC(t) + if err != nil { + t.Fatalf("PVC must still exist (orphaned, not deleted), got err: %v", err) + } + if _, orphaned := gotPVC.Labels[metadata.LabelOrphan]; !orphaned { + t.Error("expected PVC to be marked orphan after the deferral deadline") + } + + if _, err := f.getPod(t); !errors.IsNotFound(err) { + t.Errorf("expected pod to be deleted once cleanup fell back to orphaning, got err: %v", err) + } + if f.tracker.waitingForRecovery { + t.Error("fallback must not keep the reconcile in a waiting state") + } + + var sawFallback bool + for _, e := range drainEvents(f.recorder) { + if strings.Contains(e, "Warning") && strings.Contains(e, "PVCCleanupFallback") { + sawFallback = true + } + } + if !sawFallback { + t.Error("expected a Warning PVCCleanupFallback event") + } +} + +func TestCleanupPodPVC_ConfirmedMemberNeverFallsBack(t *testing.T) { + expired := time.Now().Add(-pvcCleanupDeferralTimeout - time.Minute) + f := newCohortMemberFixture(t, func(pod *corev1.Pod) { + pod.Annotations[metadata.AnnotationPVCCleanupDeferredSince] = expired.Format(time.RFC3339) + }) + f.runScaleDown(t) + + gotPVC, err := f.getPVC(t) + if err != nil { + t.Fatalf("PVC must remain while its pooler is a confirmed cohort member: %v", err) + } + if _, orphaned := gotPVC.Labels[metadata.LabelOrphan]; orphaned { + t.Error("confirmed cohort member must not fall back to orphan cleanup") + } + gotPod, err := f.getPod(t) + if err != nil { + t.Fatalf("pod must remain while it is a confirmed cohort member: %v", err) + } + if since := gotPod.Annotations[metadata.AnnotationPVCCleanupDeferredSince]; since != "" { + t.Errorf("confirmed membership must clear the fallback timer, got %q", since) + } + if !f.tracker.waitingForRecovery { + t.Error("confirmed membership must keep reconciliation waiting") + } +} + +func TestCleanupPodPVC_DrainedVerifiesCohortMembership(t *testing.T) { + expired := time.Now().Add(-pvcCleanupDeferralTimeout - time.Minute) + f := newCohortMemberFixture(t, func(pod *corev1.Pod) { + pod.Labels[metadata.LabelPodRole] = "DRAINED" + pod.Annotations[metadata.AnnotationPVCCleanupDeferredSince] = expired.Format(time.RFC3339) + }) + f.runScaleDown(t) + + gotPVC, err := f.getPVC(t) + if err != nil { + t.Fatalf("DRAINED pod's PVC must remain while it is a cohort member: %v", err) + } + if _, orphaned := gotPVC.Labels[metadata.LabelOrphan]; orphaned { + t.Error("DRAINED cohort member's PVC must not be orphaned") + } + if _, err := f.getPod(t); err != nil { + t.Fatalf("DRAINED pod must remain while it is a cohort member: %v", err) + } + if !f.tracker.waitingForRecovery { + t.Error("DRAINED cohort member cleanup must defer") + } +} + func TestHandleExternalDeletion(t *testing.T) { scheme := runtime.NewScheme() _ = multigresv1alpha1.AddToScheme(scheme) @@ -3580,7 +3936,7 @@ func TestCleanupDrainedPod_ErrorPaths(t *testing.T) { }, } - err := r.cleanupDrainedPod(context.Background(), shard, pod, poolName, poolSpec, 3) + _, err := r.cleanupDrainedPod(context.Background(), shard, pod, poolName, poolSpec, 3, nil) if err == nil { t.Error("expected error on PVC Get failure") } @@ -3620,7 +3976,7 @@ func TestCleanupDrainedPod_ErrorPaths(t *testing.T) { }, } - err := r.cleanupDrainedPod(context.Background(), shard, pod, poolName, poolSpec, 3) + _, err := r.cleanupDrainedPod(context.Background(), shard, pod, poolName, poolSpec, 3, nil) if err == nil { t.Error("expected error on PVC orphan-patch failure") } @@ -3647,13 +4003,14 @@ func TestCleanupDrainedPod_ErrorPaths(t *testing.T) { r := &ShardReconciler{Client: c, Scheme: scheme, Recorder: record.NewFakeRecorder(10)} // nil PVCDeletionPolicy defaults to Delete -> orphans the PVC. - err := r.cleanupDrainedPod( + _, err := r.cleanupDrainedPod( context.Background(), shard, pod, poolName, multigresv1alpha1.PoolSpec{}, 3, + nil, ) if err != nil { t.Fatalf("unexpected error: %v", err) @@ -6106,7 +6463,7 @@ func TestReconcilePoolPods_AdditionalErrorPaths(t *testing.T) { }) r := &ShardReconciler{Client: fails, Scheme: scheme, Recorder: record.NewFakeRecorder(10)} - err := r.cleanupDrainedPod(context.Background(), shard, pod, "main", poolSpec, 1) + _, err := r.cleanupDrainedPod(context.Background(), shard, pod, "main", poolSpec, 1, nil) if err == nil || !strings.Contains(err.Error(), "failed to mark PVC") { t.Fatalf("expected PVC orphan-patch error, got %v", err) } diff --git a/pkg/util/metadata/labels.go b/pkg/util/metadata/labels.go index a741b17f..b004df6b 100644 --- a/pkg/util/metadata/labels.go +++ b/pkg/util/metadata/labels.go @@ -146,6 +146,12 @@ const ( // was first requested. Used to detect failover timeouts. AnnotationDrainRequestedAt = "drain.multigres.com/requested-at" + // AnnotationPVCCleanupDeferredSince stores the RFC3339 timestamp of when a + // ready-for-deletion pod's PVC hard-delete was first deferred because the + // pooler's absence from the committed cohort could not be confirmed. Once + // the deferral exceeds its deadline the PVC is orphaned instead of deleted. + AnnotationPVCCleanupDeferredSince = "drain.multigres.com/pvc-cleanup-deferred-since" + // AnnotationMaintenanceRequested asks the operator to provision enough // same-cell capacity for this pod to be voluntarily evicted. Maintenance // automation must wait for AnnotationMaintenanceReady before calling the