diff --git a/CLAUDE.md b/CLAUDE.md index 135b3614..99446971 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1912,6 +1912,24 @@ the Trino backend selection). below compares the two). - **The pool keeps the cell's identity**: routing group and namespace are unchanged. Replacing compute never rewrites which warehouse lives where. +- **Voluntary node replacement is opt-in** through + `DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED` (default false, requires the pool + operator). Charts protect new coordinator and worker templates with + `karpenter.sh/do-not-disrupt=true`. The enabled operator protects existing pod + metadata after checking pod/ReplicaSet/recorded Deployment UID ownership. + A complete NodeClaim inventory and exact node/provider/claim identity establish + Drifted or deleting-node evidence. Migration 000043 preserves the first request + per instance under the pool fence. The operator then cordons that exact node + with a UID/resource-version update; it never evicts pods or deletes nodes. + This is not SIGTERM-triggered draining. Normal desired+surge capacity admits a + replacement before draining the original, even when desired exceeds the floor. + No repair allowance or query timeout is added. Current Gateway obligations and + UID-scoped retirement still control deletion. Fresh candidates reject unsafe + nodes; ambiguous admission first replays its original Gateway intent. + Node lookup errors block voluntary rollout, not health checks, explicit + recovery, or already-authorized retirement. Disabling observation does not + cancel a durable replacement request. See the + [node replacement runbook](docs/runbooks/trino-pool-node-replacement.md). - **The catalog store's partition is its OWN identity** (`DUCKGRES_TRINO_POOL_CATALOG_CELL_ID`, resolved by `trinoPoolCatalogCellID`): it is the value every coordinator of the cell diff --git a/README.md b/README.md index 7e6c5452..520e7ab4 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,11 @@ Trino operators: see [cell registration and placement](docs/trino-cells.md) for the optional registry, unchanged legacy defaults, and migration limitations. +`DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED` defaults to `false`. Enabled shared-pool +operators protect owned pods and replace instances on drifted or deleting Karpenter +nodes through normal admission and drain. See the [node replacement runbook](docs/runbooks/trino-pool-node-replacement.md) +for required permissions, rollout order, and failure boundaries. +

Duckgres Mascot

diff --git a/controlplane/admin/trino_recovery.go b/controlplane/admin/trino_recovery.go index 8321bacc..c9c4f88e 100644 --- a/controlplane/admin/trino_recovery.go +++ b/controlplane/admin/trino_recovery.go @@ -38,10 +38,11 @@ type trinoRecoveryRequest struct { } type trinoRecoveryInstanceSummary struct { - InstanceID string `json:"instance_id"` - Phase string `json:"phase"` - GatewayState string `json:"gateway_state"` - PhaseChangedAt time.Time `json:"phase_changed_at"` + InstanceID string `json:"instance_id"` + Phase string `json:"phase"` + GatewayState string `json:"gateway_state"` + PhaseChangedAt time.Time `json:"phase_changed_at"` + NodeReplacement *trinopool.NodeReplacementEvidence `json:"node_replacement,omitempty"` } func (a *TrinoAPI) handleRecoveryInstances(c *gin.Context) { @@ -75,10 +76,17 @@ func (a *TrinoAPI) handleRecoveryInstances(c *gin.Context) { if instance.PoolID != pool.PoolID || trinopool.Phase(instance.Phase).Terminal() { continue } - rows = append(rows, trinoRecoveryInstanceSummary{ + row := trinoRecoveryInstanceSummary{ InstanceID: instance.InstanceID, Phase: instance.Phase, GatewayState: instance.GatewayState, PhaseChangedAt: instance.PhaseChangedAt, - }) + } + if instance.NodeReplacementEvidence != nil { + var evidence trinopool.NodeReplacementEvidence + if json.Unmarshal([]byte(*instance.NodeReplacementEvidence), &evidence) == nil && evidence.Validate() == nil { + row.NodeReplacement = &evidence + } + } + rows = append(rows, row) } sort.Slice(rows, func(i, j int) bool { return rows[i].InstanceID < rows[j].InstanceID }) c.JSON(http.StatusOK, gin.H{"cell": a.cell.ID, "instances": rows}) diff --git a/controlplane/admin/trino_recovery_test.go b/controlplane/admin/trino_recovery_test.go index fd3a341d..27d1f6c2 100644 --- a/controlplane/admin/trino_recovery_test.go +++ b/controlplane/admin/trino_recovery_test.go @@ -94,6 +94,50 @@ func TestTrinoRecoveryInstancesAreScopedReadOnlyAndSanitized(t *testing.T) { } } +func TestTrinoRecoveryInstancesExposeTypedNodeReplacementEvidence(t *testing.T) { + api, store := recoveryTrinoAPI() + for _, test := range []struct { + name, evidence string + present bool + }{ + {"valid", `{"node_name":"node-a","node_uid":"node-uid-a","nodeclaim_name":"claim-a","nodeclaim_uid":"claim-uid-a","reason":"Drifted","private_extra":"do-not-expose"}`, true}, + {"malformed", `not-json`, false}, + {"incomplete", `{"reason":"Drifted"}`, false}, + } { + t.Run(test.name, func(t *testing.T) { + instance := *store.instance + instance.NodeReplacementEvidence = &test.evidence + store.instances = []configstore.TrinoPoolInstance{instance} + code, data := doTrinoJSON(t, trinoTestRouter(api, RoleAdmin), http.MethodGet, "/api/v1/trino/instances?cell=pool-a", "") + if code != http.StatusOK || store.writes != 0 { + t.Fatalf("inventory: %d %#v", code, data) + } + row := data["instances"].([]any)[0].(map[string]any) + projection, ok := row["node_replacement"].(map[string]any) + if ok != test.present { + t.Fatalf("unexpected evidence projection: %#v", row) + } + if ok { + want := map[string]string{"node_name": "node-a", "node_uid": "node-uid-a", "nodeclaim_name": "claim-a", "nodeclaim_uid": "claim-uid-a", "reason": "Drifted"} + if len(projection) != len(want) || len(row) != 5 { + t.Fatalf("unexpected fields: %#v", row) + } + for key, value := range want { + if projection[key] != value { + t.Fatalf("%s = %#v", key, projection[key]) + } + } + } + encoded, _ := json.Marshal(data) + for _, forbidden := range []string{"do-not-expose", "private.example.test", "blueprint", "endpoint_url", "private_extra"} { + if strings.Contains(string(encoded), forbidden) { + t.Fatalf("exposes %s", forbidden) + } + } + }) + } +} + func TestTrinoRecoveryInstancesRejectViewerAndUnavailablePool(t *testing.T) { api, store := recoveryTrinoAPI() path := "/api/v1/trino/instances?cell=pool-a" diff --git a/controlplane/configstore/migrations/000043_add_trino_node_replacement.sql b/controlplane/configstore/migrations/000043_add_trino_node_replacement.sql new file mode 100644 index 00000000..fd666157 --- /dev/null +++ b/controlplane/configstore/migrations/000043_add_trino_node_replacement.sql @@ -0,0 +1,7 @@ +-- +goose Up +ALTER TABLE duckgres_trino_pool_instances + ADD COLUMN node_replacement_evidence JSONB; + +-- +goose Down +ALTER TABLE duckgres_trino_pool_instances + DROP COLUMN node_replacement_evidence; diff --git a/controlplane/configstore/trino_pool_models.go b/controlplane/configstore/trino_pool_models.go index a2e63bfe..7ac1d506 100644 --- a/controlplane/configstore/trino_pool_models.go +++ b/controlplane/configstore/trino_pool_models.go @@ -129,20 +129,21 @@ type TrinoPoolInstance struct { // RepairFor names the failed instance this one replaces. The Gateway // charges an activation to the repair budget only when it is set; without // it a repair spends the single planned surge instead. - RepairFor string `gorm:"column:repair_for"` - FailureReason string `gorm:"column:failure_reason"` - CoordinatorDeploymentName string `gorm:"column:coordinator_deployment_name"` - CoordinatorDeploymentUID string `gorm:"column:coordinator_deployment_uid"` - WorkerDeploymentName string `gorm:"column:worker_deployment_name"` - WorkerDeploymentUID string `gorm:"column:worker_deployment_uid"` - ServiceName string `gorm:"column:service_name"` - ServiceUID string `gorm:"column:service_uid"` - ConfigMapName string `gorm:"column:config_map_name"` - ConfigMapUID string `gorm:"column:config_map_uid"` - WorkerConfigMapName string `gorm:"column:worker_config_map_name"` - WorkerConfigMapUID string `gorm:"column:worker_config_map_uid"` - CoordinatorPodUID string `gorm:"column:coordinator_pod_uid"` - CoordinatorNodeID string `gorm:"column:coordinator_node_id"` + RepairFor string `gorm:"column:repair_for"` + NodeReplacementEvidence *string `gorm:"column:node_replacement_evidence;type:jsonb"` + FailureReason string `gorm:"column:failure_reason"` + CoordinatorDeploymentName string `gorm:"column:coordinator_deployment_name"` + CoordinatorDeploymentUID string `gorm:"column:coordinator_deployment_uid"` + WorkerDeploymentName string `gorm:"column:worker_deployment_name"` + WorkerDeploymentUID string `gorm:"column:worker_deployment_uid"` + ServiceName string `gorm:"column:service_name"` + ServiceUID string `gorm:"column:service_uid"` + ConfigMapName string `gorm:"column:config_map_name"` + ConfigMapUID string `gorm:"column:config_map_uid"` + WorkerConfigMapName string `gorm:"column:worker_config_map_name"` + WorkerConfigMapUID string `gorm:"column:worker_config_map_uid"` + CoordinatorPodUID string `gorm:"column:coordinator_pod_uid"` + CoordinatorNodeID string `gorm:"column:coordinator_node_id"` // CoordinatorID is the coordinator identity the GATEWAY observed when the // member registered. It is a distinct value from the node id, and a loss // claim has to carry both exactly as the Gateway recorded them, or the @@ -177,12 +178,13 @@ func (TrinoPoolInstance) TableName() string { return "duckgres_trino_pool_instan // View projects the row onto the planner's read-only input. func (i TrinoPoolInstance) View() trinopool.InstanceView { return trinopool.InstanceView{ - ID: i.InstanceID, - Phase: trinopool.Phase(i.Phase), - ReleaseID: i.ReleaseID, - Repair: i.Repair, - RepairFor: i.RepairFor, - CreatedAt: i.CreatedAt.UnixNano(), + ID: i.InstanceID, + Phase: trinopool.Phase(i.Phase), + ReleaseID: i.ReleaseID, + Repair: i.Repair, + RepairFor: i.RepairFor, + NodeReplacement: i.NodeReplacementEvidence != nil, + CreatedAt: i.CreatedAt.UnixNano(), } } diff --git a/controlplane/configstore/trino_pool_node_replacement.go b/controlplane/configstore/trino_pool_node_replacement.go new file mode 100644 index 00000000..b0a655b7 --- /dev/null +++ b/controlplane/configstore/trino_pool_node_replacement.go @@ -0,0 +1,33 @@ +package configstore + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/posthog/duckgres/controlplane/trinopool" + "gorm.io/gorm" +) + +// RecordTrinoPoolNodeReplacement preserves the first infrastructure replacement request. +// Retries and subsequent drift observations cannot rewrite the original node identity. +func (cs *ConfigStore) RecordTrinoPoolNodeReplacement(ctx context.Context, lease TrinoPoolLease, instanceID string, evidence trinopool.NodeReplacementEvidence) error { + if err := evidence.Validate(); err != nil { + return err + } + encoded, err := json.Marshal(evidence) + if err != nil { + return err + } + return cs.withPoolAuthority(ctx, lease, func(tx *gorm.DB, _ *TrinoPool) error { + var instance TrinoPoolInstance + if err := tx.Where("instance_id = ? AND pool_id = ?", instanceID, lease.PoolID).First(&instance).Error; err != nil { + return fmt.Errorf("read instance for node replacement: %w", err) + } + if instance.NodeReplacementEvidence != nil || trinopool.Phase(instance.Phase).Terminal() { + return nil + } + return tx.Model(&TrinoPoolInstance{}).Where("instance_id = ? AND pool_id = ? AND node_replacement_evidence IS NULL", instanceID, lease.PoolID). + Update("node_replacement_evidence", string(encoded)).Error + }) +} diff --git a/controlplane/trino_pool_config.go b/controlplane/trino_pool_config.go index 2fd7e663..a107926a 100644 --- a/controlplane/trino_pool_config.go +++ b/controlplane/trino_pool_config.go @@ -26,9 +26,10 @@ import ( // (`DUCKGRES_TRINO_POOL_CATALOG_CELL_ID`, see trino_pool_catalog.go) to match // what the cell's coordinators read. const ( - envTrinoPoolEnabled = "DUCKGRES_TRINO_POOL_ENABLED" - envTrinoPoolOperatorEnabled = "DUCKGRES_TRINO_POOL_OPERATOR_ENABLED" - envTrinoPoolGatewayURL = "DUCKGRES_TRINO_POOL_GATEWAY_URL" + envTrinoPoolEnabled = "DUCKGRES_TRINO_POOL_ENABLED" + envTrinoPoolOperatorEnabled = "DUCKGRES_TRINO_POOL_OPERATOR_ENABLED" + envTrinoPoolGatewayURL = "DUCKGRES_TRINO_POOL_GATEWAY_URL" + envTrinoPoolNodeDisruptionEnabled = "DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED" trinoPoolModeFixed = "fixed" trinoPoolModeShared = "shared-pool" diff --git a/controlplane/trino_pool_node_disruption.go b/controlplane/trino_pool_node_disruption.go new file mode 100644 index 00000000..e6dea70e --- /dev/null +++ b/controlplane/trino_pool_node_disruption.go @@ -0,0 +1,143 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/controlplane/trinogateway" + "github.com/posthog/duckgres/controlplane/trinopool" +) + +type trinoPoolNodeReplacementStore interface { + RecordTrinoPoolNodeReplacement(context.Context, configstore.TrinoPoolLease, string, trinopool.NodeReplacementEvidence) error +} + +var errTrinoCandidateNodeReplacement = errors.New("candidate node requires replacement before admission") +var errTrinoCandidateNodeRetired = errors.New("candidate retired before admission because its node requires replacement") + +func definitiveTrinoNodePlacement(err error) bool { + return errors.Is(err, errTrinoCandidateNodeReplacement) || errors.Is(err, errTrinoUnschedulableNode) || errors.Is(err, errTrinoNodePlacementInvalid) +} + +// protectPoolNodes visits all live instances before any instance advances. +// A busy candidate cannot starve the protection of existing serving pods. +func (o *trinoPoolOperator) protectPoolNodes(ctx context.Context, instances []configstore.TrinoPoolInstance) error { + o.nodeProtectionErrors = make(map[string]error) + if o.nodeGuard == nil { + return nil + } + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + o.nodeGuard.reset() + o.nodeGuard.epoch = o.lease.Epoch + store, ok := o.store.(trinoPoolNodeReplacementStore) + if !ok { + return errors.New("node disruption handling requires durable replacement storage") + } + var failures []error + start := int(o.nodeProtectionCursor % uint64(max(1, len(instances)))) + o.nodeProtectionCursor++ + for offset := range instances { + index := (start + offset) % len(instances) + instance := &instances[index] + switch trinopool.Phase(instance.Phase) { + case trinopool.PhasePending, trinopool.PhaseRetiring, trinopool.PhaseRetired, trinopool.PhaseFailureRetired, trinopool.PhaseFailedPreparing, trinopool.PhaseLost: + continue + } + evidence, err := o.nodeGuard.inspect(ctx, inventoryOf(*instance)) + if err == errTrinoUnschedulableNode && instance.NodeReplacementEvidence != nil { + err = nil + } + if len(evidence) > 0 { + // Exact positive evidence remains valid when another pod or node is unavailable. + recordErr := o.dropAuthority(store.RecordTrinoPoolNodeReplacement(ctx, o.lease, instance.InstanceID, evidence[0])) + if recordErr == nil && instance.NodeReplacementEvidence == nil { + encoded, _ := json.Marshal(evidence[0]) + value := string(encoded) + instance.NodeReplacementEvidence = &value + slog.Info("Trino pool requested voluntary node replacement.", "pool", o.config.PublicID, "instance", instance.InstanceID, "node_uid", evidence[0].NodeUID, "nodeclaim_uid", evidence[0].NodeClaimUID, "reason", evidence[0].Reason) + } + if recordErr != nil { + err = errors.Join(err, recordErr) + } + if err == nil { + for _, node := range evidence { + if err = o.nodeGuard.cordon(ctx, node, o.config.PoolID, instance.InstanceID); err != nil { + break + } + } + } + } + if err == errTrinoPodProtectionIncomplete && (instance.Phase == string(trinopool.PhaseCreating) || instance.Phase == string(trinopool.PhasePreparing)) { + // Pending pods are normal convergence, not failed protection of serving work. + continue + } + if err != nil { + o.nodeProtectionErrors[instance.InstanceID] = err + failures = append(failures, fmt.Errorf("protect instance %s from node disruption: %w", instance.InstanceID, err)) + if o.fenced { + return errors.Join(failures...) + } + } + } + return errors.Join(failures...) +} + +func (o *trinoPoolOperator) candidateNodeCheck(ctx context.Context, instance configstore.TrinoPoolInstance) error { + if o.nodeGuard == nil { + return nil + } + if instance.NodeReplacementEvidence != nil { + return errTrinoCandidateNodeReplacement + } + o.nodeGuard.reset() + defer o.nodeGuard.reset() + o.nodeGuard.namedClaims = true + o.nodeGuard.epoch = o.lease.Epoch + evidence, err := o.nodeGuard.inspect(ctx, inventoryOf(instance)) + if err != nil { + return err + } + if len(evidence) != 0 || instance.NodeReplacementEvidence != nil { + return errTrinoCandidateNodeReplacement + } + return nil +} + +// nodeSafeAdmission resolves committed admission before inspecting new infrastructure facts. +// A failed fresh check cannot erase an admission whose response was lost. +func (o *trinoPoolOperator) nodeSafeAdmission(ctx context.Context, instance configstore.TrinoPoolInstance, request trinogateway.AdmitMemberRequest) (trinogateway.Member, error) { + if o.nodeGuard != nil { + current, err := o.gateway.GetMember(ctx, o.config.RoutingGroup, instance.InstanceID) + if err != nil { + return trinogateway.Member{}, err + } + if current.Incarnation != instance.GatewayIncarnation || current.PodUID != instance.CoordinatorPodUID || current.BootID != instance.CoordinatorBootID { + return trinogateway.Member{}, errors.New("admission read-back identity changed") + } + if current.Phase == "ACTIVE" { + return o.gateway.AdmitMember(ctx, o.config.RoutingGroup, instance.InstanceID, request) + } + if current.Phase == "RETIRING" || current.Phase == "RETIRED" { + return trinogateway.Member{}, errTrinoCandidateNodeRetired + } + if err := o.candidateNodeCheck(ctx, instance); err != nil { + if definitiveTrinoNodePlacement(err) { + _, retireErr := o.gateway.RetireMember(ctx, o.config.RoutingGroup, instance.InstanceID, trinogateway.MemberStepRequest{Step: o.step(instance.InstanceID, "retire"), ExpectedGeneration: current.Generation}) + if retireErr != nil { + return trinogateway.Member{}, retireErr + } + return trinogateway.Member{}, errTrinoCandidateNodeRetired + } + return trinogateway.Member{}, err + } + } + return o.gateway.AdmitMember(ctx, o.config.RoutingGroup, instance.InstanceID, request) +} diff --git a/controlplane/trino_pool_node_disruption_test.go b/controlplane/trino_pool_node_disruption_test.go new file mode 100644 index 00000000..bad25b7d --- /dev/null +++ b/controlplane/trino_pool_node_disruption_test.go @@ -0,0 +1,574 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/controlplane/trinogateway" + "github.com/posthog/duckgres/controlplane/trinopool" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + ktesting "k8s.io/client-go/testing" +) + +func attachNodeFixture(t *testing.T, h *operatorHarness, instance *configstore.TrinoPoolInstance) (*trinoPoolNodeGuard, trinoPoolInventory, *trinoNodeClaim) { + t.Helper() + g, _, inventory, claim := nodeGuardFixture(t) + instance.ServiceName = inventory.ServiceName + instance.CoordinatorDeploymentName = inventory.CoordinatorDeploymentName + instance.CoordinatorDeploymentUID = inventory.CoordinatorDeploymentUID + instance.WorkerDeploymentName = inventory.WorkerDeploymentName + instance.WorkerDeploymentUID = inventory.WorkerDeploymentUID + blueprint, _, _ := testPoolObjects(t, 7) + instance.BlueprintSnapshot, _ = blueprint.MarshalSnapshot() + h.operator.nodeGuard = g + h.operator.lease.Epoch = 7 + h.store.epoch = 7 + return g, inventory, claim +} + +func TestTrinoNodePreparingCandidateRejectsDefinitivePlacement(t *testing.T) { + for _, scenario := range []string{"drift", "unschedulable", "missing-claim", "mismatched-claim", "api-error", "scheduling"} { + t.Run(scenario, func(t *testing.T) { + h, instance := validatingNodeFixture(t) + instance.Phase = string(trinopool.PhasePreparing) + g, inventory, claim := attachNodeFixture(t, h, instance) + switch scenario { + case "drift": + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + case "unschedulable": + node, _ := g.client.CoreV1().Nodes().Get(context.Background(), "compute-node", metav1.GetOptions{}) + node.Spec.Unschedulable = true + if _, err := g.client.CoreV1().Nodes().Update(context.Background(), node, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + case "missing-claim": + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return nil, nil } + case "mismatched-claim": + claim.Status.ProviderID = "different" + case "api-error": + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return nil, errors.New("API unavailable") } + case "scheduling": + pod, _ := g.client.CoreV1().Pods(inventory.Namespace).Get(context.Background(), inventory.WorkerDeploymentName+"-0", metav1.GetOptions{}) + pod.Spec.NodeName = "" + if _, err := g.client.CoreV1().Pods(inventory.Namespace).Update(context.Background(), pod, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + } + progressed, err := h.operator.validateCandidate(context.Background(), *instance) + switch scenario { + case "api-error": + if err == nil || progressed || instance.Phase != string(trinopool.PhasePreparing) { + t.Fatal("API error retired or admitted candidate") + } + case "scheduling": + if err != nil || progressed || instance.Phase != string(trinopool.PhasePreparing) { + t.Fatalf("scheduling was treated as failure: %v", err) + } + default: + if err != nil || !progressed || instance.Phase != string(trinopool.PhaseFailedPreparing) { + t.Fatalf("invalid candidate retained preparation slot: phase=%s err=%v", instance.Phase, err) + } + } + }) + } +} + +func TestTrinoNodeProtectionTreatsCandidateSchedulingAsProgress(t *testing.T) { + for _, phase := range []trinopool.Phase{trinopool.PhaseCreating, trinopool.PhasePreparing, trinopool.PhaseServing} { + t.Run(string(phase), func(t *testing.T) { + h, instance := validatingNodeFixture(t) + instance.Phase = string(phase) + g, inventory, _ := attachNodeFixture(t, h, instance) + if err := g.client.CoreV1().Pods(inventory.Namespace).Delete(context.Background(), inventory.WorkerDeploymentName+"-0", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + err := h.operator.protectPoolNodes(context.Background(), []configstore.TrinoPoolInstance{*instance}) + if phase == trinopool.PhaseServing { + if err == nil { + t.Fatal("incomplete serving inventory was ignored") + } + } else if err != nil || len(h.operator.nodeProtectionErrors) != 0 { + t.Fatalf("normal scheduling reported reconcile failure: %v", err) + } + }) + } +} + +func TestTrinoNodeValidatingCandidateRetiresOnlyDefinitivePlacement(t *testing.T) { + for _, scenario := range []string{"mismatched-claim", "finite-grace", "api-error"} { + t.Run(scenario, func(t *testing.T) { + h, instance := validatingNodeFixture(t) + g, _, claim := attachNodeFixture(t, h, instance) + switch scenario { + case "mismatched-claim": + claim.Status.ProviderID = "different" + case "finite-grace": + duration := "1h" + claim.Spec.TerminationGracePeriod = &duration + case "api-error": + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return nil, errors.New("API unavailable") } + } + err := h.operator.admitCandidate(context.Background(), *instance) + if scenario == "api-error" { + if err == nil || instance.Phase != string(trinopool.PhaseValidating) || countCalls(h.gateway.calls, "retire:") != 0 { + t.Fatal("transient API failure authorized candidate retirement") + } + } else if err != nil || instance.Phase != string(trinopool.PhaseFailedPreparing) || h.gateway.members[instance.InstanceID].Phase != "RETIRING" { + t.Fatalf("invalid placement did not obtain retirement claim: phase=%s err=%v", instance.Phase, err) + } + if len(h.kube.deleted) != 0 { + t.Fatal("candidate deleted before retirement cleanup") + } + }) + } +} + +func TestTrinoNodeObservedReplacementPersistsAndUsesNormalRollout(t *testing.T) { + for _, deleting := range []bool{false, true} { + t.Run(map[bool]string{false: "drift", true: "deletion"}[deleting], func(t *testing.T) { + h := newOperatorHarness(t) + h.tick(t, 20) + instance := h.store.instances[h.store.order[0]] + g, _, claim := attachNodeFixture(t, h, instance) + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + if deleting { + stamp := metav1.Now() + claim.DeletionTimestamp = &stamp + claim.Status.Conditions = nil + } + if err := h.operator.protectPoolNodes(context.Background(), []configstore.TrinoPoolInstance{*instance}); err != nil { + t.Fatal(err) + } + if instance.NodeReplacementEvidence == nil { + t.Fatal("observation did not persist evidence") + } + var evidence trinopool.NodeReplacementEvidence + if err := json.Unmarshal([]byte(*instance.NodeReplacementEvidence), &evidence); err != nil { + t.Fatal(err) + } + if evidence.NodeUID != "node-uid" || evidence.NodeClaimUID != string(claim.UID) || evidence.Reason != map[bool]string{false: "Drifted", true: "Deleting"}[deleting] { + t.Fatalf("wrong evidence: %+v", evidence) + } + node, _ := g.client.CoreV1().Nodes().Get(context.Background(), "compute-node", metav1.GetOptions{}) + if !node.Spec.Unschedulable { + t.Fatal("durable request did not exclude old node") + } + instances, _ := h.store.ListTrinoPoolInstances(context.Background(), h.operator.config.PoolID) + before := len(instances) + if err := h.operator.applyPlan(context.Background(), h.store.pool, instances); err != nil { + t.Fatal(err) + } + if len(h.store.order) != before+1 || instance.Phase != string(trinopool.PhaseServing) { + t.Fatal("drained before normal replacement existed") + } + replacement := h.store.instances[h.store.order[len(h.store.order)-1]] + if replacement.Repair || replacement.RepairFor != "" { + t.Fatal("voluntary replacement consumed repair allowance") + } + // Drive replacement admission on both sides before the original may drain. + h.operator.nodeGuard = nil + for range 8 { + if replacement.Phase == string(trinopool.PhaseServing) { + break + } + if _, err := h.operator.progressInstance(context.Background(), *replacement); err != nil { + t.Fatal(err) + } + } + h.operator.nodeGuard = g + if replacement.Phase != string(trinopool.PhaseServing) || h.gateway.members[replacement.InstanceID].Phase != "ACTIVE" { + t.Fatal("replacement was not admitted before drain") + } + instances, _ = h.store.ListTrinoPoolInstances(context.Background(), h.operator.config.PoolID) + if err := h.operator.applyPlan(context.Background(), h.store.pool, instances); err != nil { + t.Fatal(err) + } + if instance.Phase != string(trinopool.PhaseDraining) { + t.Fatalf("serving replacement did not release drain: %s", instance.Phase) + } + }) + } +} + +type trinoNodeAdmissionRaceGateway struct { + *fakePoolGateway + beforeRetire func() +} + +func (g *trinoNodeAdmissionRaceGateway) RetireMember(ctx context.Context, pool, id string, request trinogateway.MemberStepRequest) (trinogateway.Member, error) { + if g.beforeRetire != nil { + call := g.beforeRetire + g.beforeRetire = nil + call() + } + return g.fakePoolGateway.RetireMember(ctx, pool, id, request) +} + +func TestTrinoNodeAdmissionWinningRetirementRaceKeepsAcceptedWork(t *testing.T) { + h, instance := validatingNodeFixture(t) + value := `{"node_name":"node","node_uid":"uid","nodeclaim_name":"claim","nodeclaim_uid":"claim-uid","reason":"Drifted"}` + instance.NodeReplacementEvidence = &value + h.operator.nodeGuard = newTrinoPoolNodeGuard(fake.NewClientset()) + validation, err := unmarshalValidationReceipt(instance.ValidationReceipt) + if err != nil { + t.Fatal(err) + } + request := trinogateway.AdmitMemberRequest{Step: h.operator.step(instance.InstanceID, "admit"), ExpectedGeneration: instance.GatewayGeneration, Receipt: trinogateway.ValidationReceipt{ + CertificateHash: validation.CertificateHash, ConfigRevision: instance.ReleaseID, AuthRevision: validation.AuthRevision, + PodUID: instance.CoordinatorPodUID, BootID: validation.ProcessID, NodeID: validation.NodeID, CoordinatorID: validation.CoordinatorID, ReadyWorkers: validation.ReadyWorkers, Checks: validation.Checks, + }} + race := &trinoNodeAdmissionRaceGateway{fakePoolGateway: h.gateway} + race.beforeRetire = func() { + if _, err := h.gateway.AdmitMember(context.Background(), h.operator.config.RoutingGroup, instance.InstanceID, request); err != nil { + t.Fatal(err) + } + } + h.operator.gateway = race + if err := h.operator.admitCandidate(context.Background(), *instance); err == nil { + t.Fatal("retirement did not lose its generation CAS") + } + if instance.Phase != string(trinopool.PhaseValidating) || h.gateway.members[instance.InstanceID].Phase != "ACTIVE" || len(h.kube.deleted) != 0 { + t.Fatal("retirement race discarded admitted work") + } + if err := h.operator.admitCandidate(context.Background(), *instance); err != nil { + t.Fatal(err) + } + if instance.Phase != string(trinopool.PhaseAdmitted) || len(h.kube.deleted) != 0 { + t.Fatal("lost admission did not resume safely") + } +} + +func (f *fakePoolStore) RecordTrinoPoolNodeReplacement(_ context.Context, lease configstore.TrinoPoolLease, id string, evidence trinopool.NodeReplacementEvidence) error { + if lease.Epoch != f.epoch { + return configstore.ErrTrinoPoolConflict + } + if err := evidence.Validate(); err != nil { + return err + } + if f.instances[id].NodeReplacementEvidence == nil { + encoded, _ := json.Marshal(evidence) + value := string(encoded) + f.instances[id].NodeReplacementEvidence = &value + } + return nil +} + +func TestTrinoNodeProtectionFailureDoesNotBlockHealthOrRecovery(t *testing.T) { + t.Run("health", func(t *testing.T) { + h := newOperatorHarness(t) + h.tick(t, 20) + instance := h.store.instances[h.store.order[0]] + instance.PhaseChangedAt = time.Now().Add(-time.Hour) + h.kube.observed.CoordinatorReady = false + h.operator.nodeGuard = newTrinoPoolNodeGuard(fake.NewClientset()) + h.tickTolerant(1) + if instance.Phase != string(trinopool.PhaseSuspect) { + t.Fatalf("protection lookup error blocked health: %s", instance.Phase) + } + }) + t.Run("authorized recovery", func(t *testing.T) { + h, _, id := recoveryHarness(t) + h.operator.nodeGuard = newTrinoPoolNodeGuard(fake.NewClientset()) + h.tickTolerant(14) + if h.store.instances[id].Phase != string(trinopool.PhaseFailureRetired) { + t.Fatalf("protection lookup error blocked authorized recovery: %s", h.store.instances[id].Phase) + } + }) +} + +func validatingNodeFixture(t *testing.T) (*operatorHarness, *configstore.TrinoPoolInstance) { + t.Helper() + h := newOperatorHarness(t) + for range 12 { + h.tick(t, 1) + for _, instance := range h.store.instances { + if instance.Phase == string(trinopool.PhaseValidating) { + return h, instance + } + } + } + t.Fatal("candidate did not reach validating") + return nil, nil +} + +func TestTrinoNodeAdmissionLostReplyStillReplaysExactIntent(t *testing.T) { + for _, changed := range []bool{false, true} { + t.Run(map[bool]string{false: "matching intent", true: "changed certificate"}[changed], func(t *testing.T) { + h, instance := validatingNodeFixture(t) + h.gateway.loseResponse = map[string]bool{"admit": true} + if err := h.operator.admitCandidate(context.Background(), *instance); err == nil { + t.Fatal("admission response was not lost") + } + if h.gateway.members[instance.InstanceID].Phase != "ACTIVE" || instance.Phase != string(trinopool.PhaseValidating) { + t.Fatal("fixture did not retain ambiguous admission") + } + h.operator.nodeGuard = newTrinoPoolNodeGuard(fake.NewClientset()) + if changed { + validation, err := unmarshalValidationReceipt(instance.ValidationReceipt) + if err != nil { + t.Fatal(err) + } + validation.CertificateHash = "changed-certificate" + instance.ValidationReceipt, err = marshalValidationReceipt(validation) + if err != nil { + t.Fatal(err) + } + } + err := h.operator.admitCandidate(context.Background(), *instance) + if changed { + if err == nil || instance.Phase != string(trinopool.PhaseValidating) { + t.Fatal("adopted ACTIVE member from a different admission intent") + } + } else if err != nil || instance.Phase != string(trinopool.PhaseAdmitted) { + t.Fatalf("matching admission replay blocked on new node observation: %v", err) + } + }) + } +} + +func TestTrinoNodeAdmissionRetiresOnlyNeverAdmittedCandidate(t *testing.T) { + h, instance := validatingNodeFixture(t) + value := `{"node_name":"node","node_uid":"uid","nodeclaim_name":"claim","nodeclaim_uid":"claim-uid","reason":"Drifted"}` + instance.NodeReplacementEvidence = &value + h.operator.nodeGuard = newTrinoPoolNodeGuard(fake.NewClientset()) + if err := h.operator.admitCandidate(context.Background(), *instance); err != nil { + t.Fatal(err) + } + if instance.Phase != string(trinopool.PhaseFailedPreparing) || h.gateway.members[instance.InstanceID].Phase != "RETIRING" { + t.Fatal("drifted candidate was not guardedly retired before admission") + } + if len(h.kube.deleted) != 0 { + t.Fatal("retirement claim deleted candidate before cleanup") + } +} + +func TestTrinoNodeReplacementRemainsAfterDriftClears(t *testing.T) { + g, _, inventory, claim := nodeGuardFixture(t) + evidence := trinopool.NodeReplacementEvidence{NodeName: "compute-node", NodeUID: "node-uid", NodeClaimName: claim.Name, NodeClaimUID: string(claim.UID), Reason: "Drifted"} + if err := g.cordon(context.Background(), evidence, "pool", "instance"); err != nil { + t.Fatal(err) + } + h := newOperatorHarness(t) + h.tick(t, 20) + id := h.store.order[0] + i := h.store.instances[id] + i.ServiceName = inventory.ServiceName + i.CoordinatorDeploymentName = inventory.CoordinatorDeploymentName + i.CoordinatorDeploymentUID = inventory.CoordinatorDeploymentUID + i.WorkerDeploymentName = inventory.WorkerDeploymentName + i.WorkerDeploymentUID = inventory.WorkerDeploymentUID + blueprint, _, _ := testPoolObjects(t, 7) + encoded, _ := blueprint.MarshalSnapshot() + i.BlueprintSnapshot = encoded + proof, _ := json.Marshal(evidence) + value := string(proof) + i.NodeReplacementEvidence = &value + h.operator.nodeGuard = g + h.operator.lease.Epoch = 7 + h.store.epoch = 7 + if err := h.operator.protectPoolNodes(context.Background(), []configstore.TrinoPoolInstance{*i}); err != nil { + t.Fatalf("cleared drift blocked recorded replacement: %v", err) + } + if h.operator.nodeProtectionErrors[id] != nil { + t.Fatal("recorded replacement was blocked") + } + if err := h.operator.candidateNodeCheck(context.Background(), *i); !errors.Is(err, errTrinoCandidateNodeReplacement) { + t.Fatalf("candidate bypassed durable replacement: %v", err) + } +} + +func TestTrinoNodeDurableReplacementSurvivesProtectionFailure(t *testing.T) { + for _, evidenceState := range []string{"valid", "absent", "invalid"} { + t.Run(evidenceState, func(t *testing.T) { + h := newOperatorHarness(t) + h.tick(t, 20) + instance := h.store.instances[h.store.order[0]] + value := `{"node_name":"node","node_uid":"uid","nodeclaim_name":"claim","nodeclaim_uid":"claim-uid","reason":"Drifted"}` + if evidenceState == "invalid" { + value = `{}` + } + if evidenceState != "absent" { + instance.NodeReplacementEvidence = &value + } + h.operator.nodeProtectionErrors = map[string]error{instance.InstanceID: errors.New("node API unavailable")} + instances, _ := h.store.ListTrinoPoolInstances(context.Background(), h.operator.config.PoolID) + before := len(instances) + err := h.operator.applyPlan(context.Background(), h.store.pool, instances) + if evidenceState != "valid" { + if len(h.store.order) != before { + t.Fatal("unproven replacement bypassed protection failure") + } + return + } + if err != nil || len(h.store.order) != before+1 { + t.Fatalf("durable replacement blocked: %v", err) + } + replacement := h.store.instances[h.store.order[len(h.store.order)-1]] + for range 8 { + if replacement.Phase == string(trinopool.PhaseServing) { + break + } + if _, err := h.operator.progressInstance(context.Background(), *replacement); err != nil { + t.Fatal(err) + } + } + if replacement.Phase != string(trinopool.PhaseServing) || h.gateway.members[replacement.InstanceID].Phase != "ACTIVE" { + t.Fatal("replacement not admitted") + } + instances, _ = h.store.ListTrinoPoolInstances(context.Background(), h.operator.config.PoolID) + if err := h.operator.applyPlan(context.Background(), h.store.pool, instances); err != nil { + t.Fatal(err) + } + if instance.Phase != string(trinopool.PhaseDraining) { + t.Fatal("accepted replacement could not drain") + } + }) + } +} + +func TestTrinoNodePartialInventoryPreservesEvidenceWithoutCordon(t *testing.T) { + for _, phase := range []trinopool.Phase{trinopool.PhaseServing, trinopool.PhasePreparing} { + t.Run(string(phase), func(t *testing.T) { + h := newOperatorHarness(t) + h.tick(t, 20) + instance := h.store.instances[h.store.order[0]] + instance.Phase = string(phase) + g, inventory, claim := attachNodeFixture(t, h, instance) + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + ctx := context.Background() + if err := g.client.CoreV1().Pods(inventory.Namespace).Delete(ctx, inventory.WorkerDeploymentName+"-0", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + err := h.operator.protectPoolNodes(ctx, []configstore.TrinoPoolInstance{*instance}) + if phase == trinopool.PhaseServing && !errors.Is(err, errTrinoPodProtectionIncomplete) { + t.Fatalf("lost inventory error: %v", err) + } + if phase == trinopool.PhasePreparing && err != nil { + t.Fatalf("scheduling became failure: %v", err) + } + if instance.NodeReplacementEvidence == nil { + t.Fatal("proven drift was discarded") + } + node, _ := g.client.CoreV1().Nodes().Get(ctx, "compute-node", metav1.GetOptions{}) + if node.Spec.Unschedulable { + t.Fatal("incomplete protection authorized cordon") + } + }) + } +} + +func TestTrinoNodeCompoundProtectionErrorsRetainEvidenceAndFaults(t *testing.T) { + h := newOperatorHarness(t) + h.tick(t, 20) + instance := h.store.instances[h.store.order[0]] + g, inventory, claim := attachNodeFixture(t, h, instance) + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + ctx := context.Background() + other := *claim + other.Name, other.UID = "other-claim", "other-uid" + other.Status.NodeName, other.Status.ProviderID, other.Status.Conditions = "a-cordoned", "test://other", nil + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return []trinoNodeClaim{*claim, other}, nil } + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "a-cordoned", UID: "other-node-uid", ResourceVersion: "1", OwnerReferences: []metav1.OwnerReference{{APIVersion: "karpenter.sh/v1", Kind: "NodeClaim", Name: other.Name, UID: other.UID}}}, Spec: corev1.NodeSpec{ProviderID: other.Status.ProviderID, Unschedulable: true}} + if _, err := g.client.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + for suffix, nodeName := range map[string]string{"-0": "a-cordoned", "-1": "b-unavailable", "-2": ""} { + pod, err := g.client.CoreV1().Pods(inventory.Namespace).Get(ctx, inventory.WorkerDeploymentName+suffix, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + pod.Spec.NodeName = nodeName + if _, err := g.client.CoreV1().Pods(inventory.Namespace).Update(ctx, pod, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + } + apiFailure := errors.New("node API unavailable") + g.client.(*fake.Clientset).PrependReactor("get", "nodes", func(action ktesting.Action) (bool, runtime.Object, error) { + if action.(ktesting.GetAction).GetName() == "b-unavailable" { + return true, nil, apiFailure + } + return false, nil, nil + }) + // Repeat after persistence to exercise the existing-evidence suppression path. + for range 2 { + err := h.operator.protectPoolNodes(ctx, []configstore.TrinoPoolInstance{*instance}) + if !errors.Is(err, apiFailure) || !errors.Is(err, errTrinoUnschedulableNode) || !errors.Is(err, errTrinoPodProtectionIncomplete) { + t.Fatalf("compound error was lost: %v", err) + } + if instance.NodeReplacementEvidence == nil { + t.Fatal("exact drift discarded beside unrelated fault") + } + drifted, _ := g.client.CoreV1().Nodes().Get(ctx, "compute-node", metav1.GetOptions{}) + if drifted.Spec.Unschedulable { + t.Fatal("compound fault authorized cordon") + } + } +} + +func TestTrinoNodeCandidateUsesFreshNamedClaimsWithoutRelisting(t *testing.T) { + h, instance := validatingNodeFixture(t) + g, _, claim := attachNodeFixture(t, h, instance) + listCalls, getCalls := 0, 0 + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { listCalls++; return []trinoNodeClaim{*claim}, nil } + g.getClaim = func(ctx context.Context, name string) (trinoNodeClaim, error) { + getCalls++ + if name != claim.Name { + t.Fatalf("wrong named claim: %s", name) + } + if _, ok := ctx.Deadline(); !ok { + t.Fatal("claim read has no deadline") + } + return *claim, nil + } + ctx := context.Background() + if err := h.operator.protectPoolNodes(ctx, []configstore.TrinoPoolInstance{*instance}); err != nil { + t.Fatal(err) + } + if err := h.operator.candidateNodeCheck(ctx, *instance); err != nil { + t.Fatal(err) + } + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + if err := h.operator.candidateNodeCheck(ctx, *instance); !errors.Is(err, errTrinoCandidateNodeReplacement) { + t.Fatalf("cached claim admitted new drift: %v", err) + } + claim.Status.Conditions = nil + node, _ := g.client.CoreV1().Nodes().Get(ctx, "compute-node", metav1.GetOptions{}) + node.Spec.Unschedulable = true + if _, err := g.client.CoreV1().Nodes().Update(ctx, node, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + if err := h.operator.candidateNodeCheck(ctx, *instance); !errors.Is(err, errTrinoUnschedulableNode) { + t.Fatalf("cached node admitted new cordon: %v", err) + } + if listCalls != 1 || getCalls != 3 { + t.Fatalf("inventory=%d named reads=%d", listCalls, getCalls) + } + if g.namedClaims || g.loaded || len(g.claims) != 0 || len(g.nodes) != 0 { + t.Fatal("candidate snapshot escaped its check") + } +} + +func TestTrinoNodeCandidateDoesNotHideSchedulingAndAPIFailure(t *testing.T) { + h, instance := validatingNodeFixture(t) + instance.Phase = string(trinopool.PhasePreparing) + g, inventory, _ := attachNodeFixture(t, h, instance) + ctx := context.Background() + if err := g.client.CoreV1().Pods(inventory.Namespace).Delete(ctx, inventory.WorkerDeploymentName+"-0", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + apiFailure := errors.New("claim API unavailable") + g.getClaim = func(context.Context, string) (trinoNodeClaim, error) { return trinoNodeClaim{}, apiFailure } + progressed, err := h.operator.validateCandidate(ctx, *instance) + if progressed || !errors.Is(err, apiFailure) || !errors.Is(err, errTrinoPodProtectionIncomplete) || instance.Phase != string(trinopool.PhasePreparing) { + t.Fatalf("candidate hid API fault or retired: progressed=%v phase=%s err=%v", progressed, instance.Phase, err) + } +} diff --git a/controlplane/trino_pool_node_guard.go b/controlplane/trino_pool_node_guard.go new file mode 100644 index 00000000..f5fe9b1e --- /dev/null +++ b/controlplane/trino_pool_node_guard.go @@ -0,0 +1,312 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "strings" + + "github.com/posthog/duckgres/controlplane/trinopool" + appsv1 "k8s.io/api/apps/v1" + 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/types" + "k8s.io/client-go/kubernetes" +) + +const trinoNodeRetirementAnnotation = "posthog.com/trino-node-retirement" + +var errTrinoUnschedulableNode = errors.New("pool pod is on an unschedulable node") +var errTrinoPodProtectionIncomplete = errors.New("pool pod protection inventory is not complete") +var errTrinoNodePlacementInvalid = errors.New("pool pod has an invalid NodeClaim placement") + +type trinoNodeClaim struct { + metav1.ObjectMeta `json:"metadata"` + Spec struct { + TerminationGracePeriod *string `json:"terminationGracePeriod"` + } `json:"spec"` + Status struct { + NodeName string `json:"nodeName"` + ProviderID string `json:"providerID"` + Conditions []metav1.Condition `json:"conditions"` + } `json:"status"` +} + +type trinoPoolNodeGuard struct { + client kubernetes.Interface + listClaims func(context.Context) ([]trinoNodeClaim, error) + getClaim func(context.Context, string) (trinoNodeClaim, error) + namedClaims bool + claims map[string]trinoNodeClaim + claimsErr error + loaded bool + nodes map[string]*corev1.Node + epoch int64 +} + +func newTrinoPoolNodeGuard(client kubernetes.Interface) *trinoPoolNodeGuard { + g := &trinoPoolNodeGuard{client: client} + g.getClaim = func(ctx context.Context, name string) (trinoNodeClaim, error) { + ctx, cancel := context.WithTimeout(ctx, trinoPoolRequestBudget) + defer cancel() + raw, err := client.CoreV1().RESTClient().Get().AbsPath("/apis/karpenter.sh/v1").Resource("nodeclaims").Name(name).SetHeader("Accept", "application/json").Do(ctx).Raw() + if err != nil { + return trinoNodeClaim{}, err + } + var claim trinoNodeClaim + err = json.Unmarshal(raw, &claim) + return claim, err + } + g.listClaims = func(ctx context.Context) ([]trinoNodeClaim, error) { + ctx, cancel := context.WithTimeout(ctx, trinoPoolRequestBudget) + defer cancel() + var claims []trinoNodeClaim + continuation := "" + for { + request := client.CoreV1().RESTClient().Get().AbsPath("/apis/karpenter.sh/v1/nodeclaims").SetHeader("Accept", "application/json").Param("limit", "500") + if continuation != "" { + request = request.Param("continue", continuation) + } + raw, err := request.Do(ctx).Raw() + if err != nil { + return nil, err + } + var page struct { + Metadata metav1.ListMeta `json:"metadata"` + Items []trinoNodeClaim `json:"items"` + } + if err := json.Unmarshal(raw, &page); err != nil { + return nil, err + } + claims = append(claims, page.Items...) + if len(claims) > 10000 { + return nil, errors.New("NodeClaim inventory exceeds safety limit") + } + if page.Metadata.Continue == "" { + return claims, nil + } + if page.Metadata.Continue == continuation { + return nil, errors.New("NodeClaim pagination did not advance") + } + continuation = page.Metadata.Continue + } + } + return g +} + +func (g *trinoPoolNodeGuard) reset() { + g.namedClaims = false + g.loaded = false + g.claims = nil + g.claimsErr = nil + g.nodes = make(map[string]*corev1.Node) +} + +func (g *trinoPoolNodeGuard) nodeClaim(ctx context.Context, node *corev1.Node) (trinoNodeClaim, error) { + if !g.loaded && !g.namedClaims { + g.loaded = true + claims, err := g.listClaims(ctx) + g.claimsErr = err + g.claims = make(map[string]trinoNodeClaim) + for _, claim := range claims { + g.claims[string(claim.UID)] = claim + } + } + if g.claimsErr != nil { + return trinoNodeClaim{}, g.claimsErr + } + for _, owner := range node.OwnerReferences { + if owner.APIVersion != "karpenter.sh/v1" || owner.Kind != "NodeClaim" { + continue + } + claim, ok := g.claims[string(owner.UID)] + if g.namedClaims && !ok { + var err error + claim, err = g.getClaim(ctx, owner.Name) + if apierrors.IsNotFound(err) { + return trinoNodeClaim{}, fmt.Errorf("%w: NodeClaim no longer exists", errTrinoNodePlacementInvalid) + } + if err != nil { + return trinoNodeClaim{}, err + } + ok = claim.UID == owner.UID + if g.claims == nil { + g.claims = make(map[string]trinoNodeClaim) + } + g.claims[string(owner.UID)] = claim + } + if !ok || owner.UID == "" || claim.UID != owner.UID || claim.Name != owner.Name || claim.Status.NodeName != node.Name || node.Spec.ProviderID == "" || claim.Status.ProviderID != node.Spec.ProviderID { + return trinoNodeClaim{}, fmt.Errorf("%w: node lacks an exact matching NodeClaim", errTrinoNodePlacementInvalid) + } + if claim.Spec.TerminationGracePeriod != nil { + return trinoNodeClaim{}, fmt.Errorf("%w: finite NodeClaim termination grace period can override query protection", errTrinoNodePlacementInvalid) + } + return claim, nil + } + return trinoNodeClaim{}, fmt.Errorf("%w: pool pod node is not owned by a NodeClaim", errTrinoNodePlacementInvalid) +} + +// inspect protects existing pods without modifying their Deployment templates. +// Labels select candidates; exact Deployment and ReplicaSet UIDs establish ownership. +func (g *trinoPoolNodeGuard) inspect(ctx context.Context, inventory trinoPoolInventory) ([]trinopool.NodeReplacementEvidence, error) { + ctx, cancel := context.WithTimeout(ctx, trinoPoolRequestBudget) + defer cancel() + deployments := make(map[types.UID]*appsv1.Deployment) + for _, wanted := range []struct{ name, uid string }{ + {inventory.CoordinatorDeploymentName, inventory.CoordinatorDeploymentUID}, + {inventory.WorkerDeploymentName, inventory.WorkerDeploymentUID}, + } { + if wanted.name == "" || wanted.uid == "" { + return nil, errors.New("pod protection requires recorded Deployment identities") + } + deployment, err := g.client.AppsV1().Deployments(inventory.Namespace).Get(ctx, wanted.name, metav1.GetOptions{}) + if err != nil { + return nil, err + } + if string(deployment.UID) != wanted.uid || deployment.Labels[trinopool.LabelInstance] != inventory.ServiceName || deployment.Labels[trinopool.LabelManagedBy] != trinopool.ManagedByValue { + return nil, errTrinoPoolForeignObject + } + epoch, err := strconv.ParseInt(deployment.Annotations[trinopool.AnnotationAuthorityEpoch], 10, 64) + if err != nil || epoch > g.epoch { + return nil, errTrinoPoolStaleEpoch + } + if deployment.Spec.Replicas == nil { + return nil, errTrinoPodProtectionIncomplete + } + deployments[deployment.UID] = deployment + } + pods, err := g.client.CoreV1().Pods(inventory.Namespace).List(ctx, metav1.ListOptions{LabelSelector: trinopool.LabelInstance + "=" + inventory.ServiceName, Limit: trinoPoolListLimit}) + if err != nil { + return nil, err + } + if pods.Continue != "" { + return nil, errors.New("pool pod protection inventory exceeds listing limit") + } + replicaSets := make(map[types.UID]types.UID) + scheduled := make(map[types.UID]int) + nodeNames := make(map[string]bool) + for _, pod := range pods.Items { + if pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodFailed { + continue + } + owner := metav1.GetControllerOf(&pod) + if owner == nil || owner.APIVersion != "apps/v1" || owner.Kind != "ReplicaSet" || owner.UID == "" { + return nil, errTrinoPoolForeignObject + } + if replicaSets[owner.UID] == "" { + rs, err := g.client.AppsV1().ReplicaSets(inventory.Namespace).Get(ctx, owner.Name, metav1.GetOptions{}) + if err != nil { + return nil, err + } + parent := metav1.GetControllerOf(rs) + if rs.UID != owner.UID || parent == nil || parent.APIVersion != "apps/v1" || parent.Kind != "Deployment" || deployments[parent.UID] == nil || deployments[parent.UID].Name != parent.Name { + return nil, errTrinoPoolForeignObject + } + replicaSets[owner.UID] = parent.UID + } + if pod.UID == "" || pod.ResourceVersion == "" { + return nil, errors.New("pod protection requires UID and resource version") + } + if pod.Annotations[trinopool.AnnotationDoNotDisrupt] != "true" { + patch, _ := json.Marshal(map[string]any{"metadata": map[string]any{"uid": pod.UID, "resourceVersion": pod.ResourceVersion, "annotations": map[string]string{trinopool.AnnotationDoNotDisrupt: "true"}}}) + if _, err := g.client.CoreV1().Pods(inventory.Namespace).Patch(ctx, pod.Name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { + return nil, err + } + } + if pod.Spec.NodeName != "" { + nodeNames[pod.Spec.NodeName] = true + if pod.DeletionTimestamp == nil { + scheduled[replicaSets[owner.UID]]++ + } + } + } + var evidence []trinopool.NodeReplacementEvidence + var failures []error + ordered := make([]string, 0, len(nodeNames)) + for name := range nodeNames { + ordered = append(ordered, name) + } + sort.Strings(ordered) + for _, name := range ordered { + node := g.nodes[name] + if node == nil { + node, err = g.client.CoreV1().Nodes().Get(ctx, name, metav1.GetOptions{}) + if err != nil { + failures = append(failures, err) + continue + } + if g.nodes == nil { + g.nodes = make(map[string]*corev1.Node) + } + g.nodes[name] = node + } + claim, err := g.nodeClaim(ctx, node) + if err != nil { + failures = append(failures, err) + continue + } + reason := "" + for _, condition := range claim.Status.Conditions { + if condition.Type == "Drifted" && condition.Status == metav1.ConditionTrue { + reason = "Drifted" + } + } + if node.DeletionTimestamp != nil || claim.DeletionTimestamp != nil { + reason = "Deleting" + } + if reason != "" { + e := trinopool.NodeReplacementEvidence{NodeName: node.Name, NodeUID: string(node.UID), NodeClaimName: claim.Name, NodeClaimUID: string(claim.UID), Reason: reason} + if err := e.Validate(); err != nil { + failures = append(failures, err) + continue + } + evidence = append(evidence, e) + } else if node.Spec.Unschedulable { + failures = append(failures, errTrinoUnschedulableNode) + } + } + for uid, deployment := range deployments { + if scheduled[uid] != int(*deployment.Spec.Replicas) { + failures = append(failures, errTrinoPodProtectionIncomplete) + break + } + } + if len(failures) == 1 { + return evidence, failures[0] + } + return evidence, errors.Join(failures...) +} + +// cordon only excludes future scheduling. It never evicts pods or deletes nodes. +// The durable replacement request precedes this UID-bound, optimistic update. +func (g *trinoPoolNodeGuard) cordon(ctx context.Context, evidence trinopool.NodeReplacementEvidence, poolID, instanceID string) error { + ctx, cancel := context.WithTimeout(ctx, trinoPoolRequestBudget) + defer cancel() + node, err := g.client.CoreV1().Nodes().Get(ctx, evidence.NodeName, metav1.GetOptions{}) + if err != nil { + return err + } + if string(node.UID) != evidence.NodeUID || node.ResourceVersion == "" { + return errors.New("node identity changed before cordon") + } + claim, err := g.nodeClaim(ctx, node) + if err != nil { + return err + } + if string(claim.UID) != evidence.NodeClaimUID || claim.Name != evidence.NodeClaimName { + return errors.New("NodeClaim identity changed before cordon") + } + if node.Spec.Unschedulable { + return nil + } + owner := strings.Join([]string{poolID, instanceID, evidence.NodeClaimUID, evidence.Reason}, "/") + patch, _ := json.Marshal(map[string]any{"metadata": map[string]any{"uid": node.UID, "resourceVersion": node.ResourceVersion, "annotations": map[string]string{trinoNodeRetirementAnnotation: owner}}, "spec": map[string]any{"unschedulable": true}}) + _, err = g.client.CoreV1().Nodes().Patch(ctx, node.Name, types.MergePatchType, patch, metav1.PatchOptions{}) + return err +} diff --git a/controlplane/trino_pool_node_guard_test.go b/controlplane/trino_pool_node_guard_test.go new file mode 100644 index 00000000..3d74d1b7 --- /dev/null +++ b/controlplane/trino_pool_node_guard_test.go @@ -0,0 +1,303 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "testing" + + "github.com/posthog/duckgres/controlplane/trinopool" + appsv1 "k8s.io/api/apps/v1" + 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/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/rest" +) + +func nodeGuardFixture(t *testing.T) (*trinoPoolNodeGuard, *fake.Clientset, trinoPoolInventory, *trinoNodeClaim) { + t.Helper() + ctx := context.Background() + effects, client := newEffects(t, 7) + _, _, objects := testPoolObjects(t, 7) + inventory, err := effects.Apply(ctx, objects) + if err != nil { + t.Fatal(err) + } + claim := &trinoNodeClaim{ObjectMeta: metav1.ObjectMeta{Name: "compute-claim", UID: "claim-uid"}} + claim.Status.NodeName = "compute-node" + claim.Status.ProviderID = "test://compute" + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "compute-node", UID: "node-uid", ResourceVersion: "1", OwnerReferences: []metav1.OwnerReference{{APIVersion: "karpenter.sh/v1", Kind: "NodeClaim", Name: claim.Name, UID: claim.UID}}}, Spec: corev1.NodeSpec{ProviderID: claim.Status.ProviderID}} + if _, err := client.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + for _, name := range []string{inventory.CoordinatorDeploymentName, inventory.WorkerDeploymentName} { + deployment, err := client.AppsV1().Deployments(inventory.Namespace).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + t.Fatal(err) + } + controller := true + rs := &appsv1.ReplicaSet{ObjectMeta: metav1.ObjectMeta{Name: name + "-rs", Namespace: inventory.Namespace, UID: types.UID(name + "-rs-uid"), OwnerReferences: []metav1.OwnerReference{{APIVersion: "apps/v1", Kind: "Deployment", Name: name, UID: deployment.UID, Controller: &controller}}}} + if _, err := client.AppsV1().ReplicaSets(inventory.Namespace).Create(ctx, rs, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + for index := int32(0); index < *deployment.Spec.Replicas; index++ { + pod := &corev1.Pod{ObjectMeta: *deployment.Spec.Template.ObjectMeta.DeepCopy(), Spec: *deployment.Spec.Template.Spec.DeepCopy()} + pod.Name = fmt.Sprintf("%s-%d", name, index) + pod.Namespace = inventory.Namespace + pod.UID = types.UID(pod.Name + "-uid") + pod.ResourceVersion = "1" + pod.Spec.NodeName = node.Name + pod.OwnerReferences = []metav1.OwnerReference{{APIVersion: "apps/v1", Kind: "ReplicaSet", Name: rs.Name, UID: rs.UID, Controller: &controller}} + delete(pod.Annotations, trinopool.AnnotationDoNotDisrupt) + if _, err := client.CoreV1().Pods(inventory.Namespace).Create(ctx, pod, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + } + } + g := newTrinoPoolNodeGuard(client) + g.epoch = 7 + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return []trinoNodeClaim{*claim}, nil } + g.getClaim = func(ctx context.Context, name string) (trinoNodeClaim, error) { + claims, err := g.listClaims(ctx) + if err != nil { + return trinoNodeClaim{}, err + } + for _, candidate := range claims { + if candidate.Name == name { + return candidate, nil + } + } + return trinoNodeClaim{}, apierrors.NewNotFound(schema.GroupResource{Group: "karpenter.sh", Resource: "nodeclaims"}, name) + } + g.reset() + return g, client, inventory, claim +} + +func TestTrinoNodeGuardProtectsOwnedCoordinatorAndWorkersWithoutRolling(t *testing.T) { + g, client, inventory, _ := nodeGuardFixture(t) + ctx := context.Background() + before, _ := client.AppsV1().Deployments(inventory.Namespace).List(ctx, metav1.ListOptions{}) + evidence, err := g.inspect(ctx, inventory) + if err != nil || len(evidence) != 0 { + t.Fatalf("inspect = %v, %v", evidence, err) + } + pods, _ := client.CoreV1().Pods(inventory.Namespace).List(ctx, metav1.ListOptions{}) + if len(pods.Items) < 2 { + t.Fatal("fixture does not cover both roles") + } + for _, pod := range pods.Items { + if pod.Annotations[trinopool.AnnotationDoNotDisrupt] != "true" { + t.Fatalf("pod not protected: %s", pod.Name) + } + } + after, _ := client.AppsV1().Deployments(inventory.Namespace).List(ctx, metav1.ListOptions{}) + if !reflect.DeepEqual(before, after) { + t.Fatal("pod protection changed a Deployment") + } +} + +func TestTrinoNodeGuardFailsClosedOnAmbiguousOwnershipAndNodeState(t *testing.T) { + for _, scenario := range []string{"deployment-uid", "replicaset-uid", "claim-uid", "provider-id", "claim-read-error", "finite-grace", "missing-pods", "newer-epoch"} { + t.Run(scenario, func(t *testing.T) { + g, client, inventory, claim := nodeGuardFixture(t) + ctx := context.Background() + switch scenario { + case "deployment-uid": + inventory.WorkerDeploymentUID = "different" + case "replicaset-uid": + rs, _ := client.AppsV1().ReplicaSets(inventory.Namespace).Get(ctx, inventory.CoordinatorDeploymentName+"-rs", metav1.GetOptions{}) + rs.UID = "different" + if _, err := client.AppsV1().ReplicaSets(inventory.Namespace).Update(ctx, rs, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + case "claim-uid": + claim.UID = "different" + case "provider-id": + claim.Status.ProviderID = "test://other" + case "claim-read-error": + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return nil, errors.New("unavailable") } + case "finite-grace": + value := "1h" + claim.Spec.TerminationGracePeriod = &value + case "missing-pods": + if err := client.CoreV1().Pods(inventory.Namespace).Delete(ctx, inventory.WorkerDeploymentName+"-0", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + case "newer-epoch": + g.epoch = 6 + } + if evidence, err := g.inspect(ctx, inventory); err == nil || len(evidence) != 0 { + t.Fatalf("unsafe inspection: %v, %v", evidence, err) + } + node, _ := client.CoreV1().Nodes().Get(ctx, "compute-node", metav1.GetOptions{}) + if node.Spec.Unschedulable { + t.Fatal("observation error cordoned node") + } + }) + } +} + +func TestTrinoNodeGuardDriftAndDeletingClaimsKeepWorkRunning(t *testing.T) { + for _, deleting := range []bool{false, true} { + t.Run(fmt.Sprint(deleting), func(t *testing.T) { + g, client, inventory, claim := nodeGuardFixture(t) + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + if deleting { + stamp := metav1.Now() + claim.DeletionTimestamp = &stamp + claim.Status.Conditions = nil + } + evidence, err := g.inspect(context.Background(), inventory) + if err != nil || len(evidence) != 1 { + t.Fatalf("evidence: %v, %v", evidence, err) + } + if err := g.cordon(context.Background(), evidence[0], "pool", "instance"); err != nil { + t.Fatal(err) + } + node, _ := client.CoreV1().Nodes().Get(context.Background(), "compute-node", metav1.GetOptions{}) + if !node.Spec.Unschedulable || node.Annotations[trinoNodeRetirementAnnotation] == "" { + t.Fatal("node lacks owned scheduling exclusion") + } + for _, action := range client.Actions() { + if action.GetVerb() == "delete" || action.GetVerb() == "delete-collection" { + t.Fatalf("node replacement deleted a resource: %v", action) + } + } + }) + } +} + +func TestTrinoNodeGuardRejectsChangedNodeBeforeCordon(t *testing.T) { + g, client, inventory, claim := nodeGuardFixture(t) + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + evidence, err := g.inspect(context.Background(), inventory) + if err != nil { + t.Fatal(err) + } + node, _ := client.CoreV1().Nodes().Get(context.Background(), "compute-node", metav1.GetOptions{}) + node.UID = "replacement-node" + if _, err := client.CoreV1().Nodes().Update(context.Background(), node, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + if err := g.cordon(context.Background(), evidence[0], "pool", "instance"); err == nil { + t.Fatal("cordoned a replacement node") + } +} + +func TestTrinoNodeGuardSharesClaimInventoryPerPass(t *testing.T) { + g, _, inventory, claim := nodeGuardFixture(t) + calls := 0 + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { calls++; return []trinoNodeClaim{*claim}, nil } + for range 3 { + if _, err := g.inspect(context.Background(), inventory); err != nil { + t.Fatal(err) + } + } + if calls != 1 { + t.Fatalf("claim inventory fetched %d times", calls) + } + g.reset() + if _, err := g.inspect(context.Background(), inventory); err != nil { + t.Fatal(err) + } + if calls != 2 { + t.Fatal("new pass reused old NodeClaim inventory") + } +} + +func TestTrinoNodeClaimInventoryRequestsJSONAndAllPages(t *testing.T) { + pages := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/apis/karpenter.sh/v1/nodeclaims" || r.Header.Get("Accept") != "application/json" || r.URL.Query().Get("limit") != "500" { + t.Errorf("unexpected request: %s accept=%s", r.URL, r.Header.Get("Accept")) + } + w.Header().Set("Content-Type", "application/json") + pages++ + if r.URL.Query().Get("continue") == "" { + _, _ = w.Write([]byte(`{"metadata":{"continue":"page-2"},"items":[{"metadata":{"name":"claim-a","uid":"uid-a"}}]}`)) + return + } + _, _ = w.Write([]byte(`{"metadata":{},"items":[{"metadata":{"name":"claim-b","uid":"uid-b"}}]}`)) + })) + defer server.Close() + client, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + claims, err := newTrinoPoolNodeGuard(client).listClaims(context.Background()) + if err != nil || len(claims) != 2 || pages != 2 { + t.Fatalf("claims=%v pages=%d err=%v", claims, pages, err) + } + encoded, _ := json.Marshal(claims) + if len(encoded) == 0 { + t.Fatal("empty encoded claims") + } +} + +func TestTrinoNodeGuardCollectsDriftPastUnrelatedCordon(t *testing.T) { + for _, name := range []string{"a-cordoned", "z-cordoned"} { + t.Run(name, func(t *testing.T) { + g, client, inventory, claim := nodeGuardFixture(t) + ctx := context.Background() + claim.Status.Conditions = []metav1.Condition{{Type: "Drifted", Status: metav1.ConditionTrue}} + other := *claim + other.Name, other.UID = "other-claim", "other-uid" + other.Status.NodeName, other.Status.ProviderID, other.Status.Conditions = name, "test://other", nil + g.listClaims = func(context.Context) ([]trinoNodeClaim, error) { return []trinoNodeClaim{*claim, other}, nil } + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: name, UID: "other-node-uid", ResourceVersion: "1", OwnerReferences: []metav1.OwnerReference{{APIVersion: "karpenter.sh/v1", Kind: "NodeClaim", Name: other.Name, UID: other.UID}}}, Spec: corev1.NodeSpec{ProviderID: other.Status.ProviderID, Unschedulable: true}} + if _, err := client.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + pod, _ := client.CoreV1().Pods(inventory.Namespace).Get(ctx, inventory.WorkerDeploymentName+"-0", metav1.GetOptions{}) + pod.Spec.NodeName = name + if _, err := client.CoreV1().Pods(inventory.Namespace).Update(ctx, pod, metav1.UpdateOptions{}); err != nil { + t.Fatal(err) + } + evidence, err := g.inspect(ctx, inventory) + if !errors.Is(err, errTrinoUnschedulableNode) || len(evidence) != 1 || evidence[0].NodeUID != "node-uid" { + t.Fatalf("order-dependent observation: %v %v", evidence, err) + } + }) + } +} + +func TestTrinoNodeNamedClaimRejectsCachedWrongUID(t *testing.T) { + g, client, _, claim := nodeGuardFixture(t) + g.namedClaims = true + claim.UID = "replaced-claim-uid" + node, _ := client.CoreV1().Nodes().Get(context.Background(), "compute-node", metav1.GetOptions{}) + for range 2 { + if _, err := g.nodeClaim(context.Background(), node); !errors.Is(err, errTrinoNodePlacementInvalid) { + t.Fatalf("accepted wrong cached UID: %v", err) + } + } +} + +func TestTrinoNodeNamedClaimRequestsFreshJSON(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/apis/karpenter.sh/v1/nodeclaims/claim-a" || r.Header.Get("Accept") != "application/json" { + t.Errorf("unexpected request: %s accept=%s", r.URL, r.Header.Get("Accept")) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"metadata":{"name":"claim-a","uid":"uid-a"}}`)) + })) + defer server.Close() + client, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + claim, err := newTrinoPoolNodeGuard(client).getClaim(context.Background(), "claim-a") + if err != nil || claim.Name != "claim-a" || claim.UID != "uid-a" { + t.Fatalf("claim=%v err=%v", claim, err) + } +} diff --git a/controlplane/trino_pool_operator.go b/controlplane/trino_pool_operator.go index f6a183d5..242526e0 100644 --- a/controlplane/trino_pool_operator.go +++ b/controlplane/trino_pool_operator.go @@ -6,6 +6,7 @@ import ( "context" "crypto/rand" "encoding/hex" + "encoding/json" "errors" "fmt" "log/slog" @@ -95,15 +96,18 @@ type trinoPoolValidator func(ctx context.Context, endpoint string, observed trin type trinoPoolIdentityProbe func(ctx context.Context, endpoint string) (string, error) type trinoPoolOperator struct { - config trinoPoolConfig - store trinoPoolStore - gateway trinoPoolGateway - kube func(epoch int64) trinoPoolKube - validate trinoPoolValidator - identity trinoPoolIdentityProbe - queryDrainStatus func(context.Context, string, string) (trinoQueryDrainStatus, error) - drainCursors map[string]string - recoveryEvidence trinoPoolRecoveryEvidence + config trinoPoolConfig + store trinoPoolStore + gateway trinoPoolGateway + kube func(epoch int64) trinoPoolKube + validate trinoPoolValidator + identity trinoPoolIdentityProbe + queryDrainStatus func(context.Context, string, string) (trinoQueryDrainStatus, error) + drainCursors map[string]string + recoveryEvidence trinoPoolRecoveryEvidence + nodeGuard *trinoPoolNodeGuard + nodeProtectionErrors map[string]error + nodeProtectionCursor uint64 // projection reports what this control plane currently serves: the // authorization bundle's revision and the fingerprints of the projected // password and group files. It is what a member is compared against when @@ -346,6 +350,10 @@ func (o *trinoPoolOperator) reconcileOnce(ctx context.Context) error { return fmt.Errorf("list pool instances: %w", err) } poolObservation(ctx).snapshot(instances, time.Now()) + nodeErr := o.protectPoolNodes(ctx, instances) + if o.fenced { + return errors.Join(tenantErr, nodeErr) + } // Advance the instances already in flight before starting anything new, so // a slow rollout cannot be overtaken by its own successor. // @@ -355,12 +363,12 @@ func (o *trinoPoolOperator) reconcileOnce(ctx context.Context) error { // lifecycle with it. progressed, instanceErr := o.progressInstances(ctx, instances) if instanceErr != nil && o.fenced { - return errors.Join(tenantErr, instanceErr) + return errors.Join(tenantErr, nodeErr, instanceErr) } if progressed { - return errors.Join(tenantErr, instanceErr) + return errors.Join(tenantErr, nodeErr, instanceErr) } - return errors.Join(tenantErr, instanceErr, o.applyPlan(ctx, pool, instances)) + return errors.Join(tenantErr, nodeErr, instanceErr, o.applyPlan(ctx, pool, instances)) } // refreshConfig replaces the desired configuration with what the authoritative @@ -541,6 +549,19 @@ func (o *trinoPoolOperator) applyPlan(ctx context.Context, pool *configstore.Tri desiredBlueprintDigest := o.config.Blueprint.Digest() for _, instance := range instances { view := instance.View() + validNodeEvidence := false + if instance.NodeReplacementEvidence != nil { + var evidence trinopool.NodeReplacementEvidence + if err := json.Unmarshal([]byte(*instance.NodeReplacementEvidence), &evidence); err != nil || evidence.Validate() != nil { + view.RolloutBlocked = true + failures = append(failures, fmt.Errorf("instance %s has unreadable node replacement evidence", instance.InstanceID)) + } else { + validNodeEvidence = true + } + } + if o.nodeProtectionErrors[instance.InstanceID] != nil && !validNodeEvidence { + view.RolloutBlocked = true + } if view.Phase.Serving() { blueprint, err := trinopool.ParseBlueprint([]byte(instance.BlueprintSnapshot)) if err != nil { diff --git a/controlplane/trino_pool_progress.go b/controlplane/trino_pool_progress.go index 942b780b..ddfc6cec 100644 --- a/controlplane/trino_pool_progress.go +++ b/controlplane/trino_pool_progress.go @@ -289,6 +289,18 @@ func runningCoordinatorContainer(observed trinoPoolObservation, podUID string) s } func (o *trinoPoolOperator) validateCandidate(ctx context.Context, instance configstore.TrinoPoolInstance) (bool, error) { + if instance.NodeReplacementEvidence != nil { + return true, o.failCandidate(ctx, instance, trinopool.PhasePreparing, "candidate node requires replacement before admission") + } + if err := o.candidateNodeCheck(ctx, instance); err != nil { + if definitiveTrinoNodePlacement(err) { + return true, o.failCandidate(ctx, instance, trinopool.PhasePreparing, "candidate has an invalid node placement before admission") + } + if err == errTrinoPodProtectionIncomplete { + return false, nil + } + return false, err + } observed, err := o.kube(o.lease.Epoch).Observe(ctx, inventoryOf(instance)) if err != nil { return false, fmt.Errorf("observe %s: %w", instance.InstanceID, err) @@ -377,7 +389,7 @@ func (o *trinoPoolOperator) admitCandidate(ctx context.Context, instance configs BootID: validation.ProcessID, }, func(ctx context.Context) (string, error) { - admitted, err := o.gateway.AdmitMember(ctx, o.config.RoutingGroup, instance.InstanceID, request) + admitted, err := o.nodeSafeAdmission(ctx, instance, request) if err != nil { return "", err } @@ -385,6 +397,9 @@ func (o *trinoPoolOperator) admitCandidate(ctx context.Context, instance configs return fmt.Sprintf(`{"phase":%q,"generation":%d}`, admitted.Phase, admitted.Generation), nil }, ); err != nil { + if errors.Is(err, errTrinoCandidateNodeRetired) { + return o.failCandidate(ctx, instance, trinopool.PhaseValidating, "candidate node required replacement before admission") + } if errors.Is(err, trinogateway.ErrPublicationBarrier) { // A member joining while a tenant publication is open must // acknowledge that publication's target revision, which a member diff --git a/controlplane/trino_pool_wiring.go b/controlplane/trino_pool_wiring.go index a7de81f0..342d5403 100644 --- a/controlplane/trino_pool_wiring.go +++ b/controlplane/trino_pool_wiring.go @@ -9,6 +9,7 @@ import ( "fmt" "log/slog" "os" + "strconv" "strings" "sync/atomic" @@ -89,6 +90,9 @@ func buildTrinoPoolOperators( publications: store, operations: store, } + if enabled, _ := strconv.ParseBool(strings.TrimSpace(os.Getenv(envTrinoPoolNodeDisruptionEnabled))); enabled && operatorEnabled { + operator.nodeGuard = newTrinoPoolNodeGuard(clientset) + } // Desired state is published from the ConfigMap the chart projects the // registry and blueprint from, read through the Kubernetes API. // diff --git a/controlplane/trinopool/instantiate.go b/controlplane/trinopool/instantiate.go index 4de3f802..805b93d8 100644 --- a/controlplane/trinopool/instantiate.go +++ b/controlplane/trinopool/instantiate.go @@ -28,6 +28,7 @@ const ( AnnotationSpecDigest = "posthog.com/trino-spec-digest" AnnotationAuthorityEpoch = "posthog.com/trino-authority-epoch" AnnotationReleaseID = "posthog.com/trino-release-id" + AnnotationDoNotDisrupt = "karpenter.sh/do-not-disrupt" componentCoordinator = "coordinator" componentWorker = "worker" diff --git a/controlplane/trinopool/node_replacement.go b/controlplane/trinopool/node_replacement.go new file mode 100644 index 00000000..48b2aa3f --- /dev/null +++ b/controlplane/trinopool/node_replacement.go @@ -0,0 +1,23 @@ +package trinopool + +import "errors" + +// NodeReplacementEvidence identifies the first observed infrastructure replacement request. +// It never authorizes resource deletion or classifies a process as dead. +type NodeReplacementEvidence struct { + NodeName string `json:"node_name"` + NodeUID string `json:"node_uid"` + NodeClaimName string `json:"nodeclaim_name"` + NodeClaimUID string `json:"nodeclaim_uid"` + Reason string `json:"reason"` +} + +func (e NodeReplacementEvidence) Validate() error { + if e.NodeName == "" || e.NodeUID == "" || e.NodeClaimName == "" || e.NodeClaimUID == "" { + return errors.New("node replacement requires exact node and NodeClaim identities") + } + if e.Reason != "Drifted" && e.Reason != "Deleting" { + return errors.New("node replacement requires authoritative drift or deletion evidence") + } + return nil +} diff --git a/controlplane/trinopool/plan.go b/controlplane/trinopool/plan.go index a99724c7..258efbce 100644 --- a/controlplane/trinopool/plan.go +++ b/controlplane/trinopool/plan.go @@ -20,6 +20,8 @@ type InstanceView struct { ReleaseID string // SpecOutdated compares desired configuration with this instance's immutable specification. SpecOutdated bool + // NodeReplacement requires a serving replacement before voluntary infrastructure retirement. + NodeReplacement bool // RolloutBlocked retains capacity accounting but excludes a member with unreadable rollout evidence. RolloutBlocked bool // Repair marks an instance created against the capacity-repair budget. @@ -145,7 +147,10 @@ func PlanNext(state PoolState) Plan { } // 4. Drain only when the floor survives it. - if serving-1 >= state.MinServing { + if serving-1 >= state.MinServing && (!outdated[0].NodeReplacement || serving > state.DesiredInstances) { + if outdated[0].NodeReplacement { + return Plan{Action: PlanActionDrain, InstanceID: outdated[0].ID, Reason: "replacing an instance whose node is retiring"} + } return Plan{Action: PlanActionDrain, InstanceID: outdated[0].ID, Reason: "replacing an instance with an outdated specification"} } @@ -161,7 +166,7 @@ func PlanNext(state PoolState) Plan { func outdatedServing(state PoolState) []InstanceView { var outdated []InstanceView for _, instance := range state.Instances { - if instance.Phase.Serving() && !instance.RolloutBlocked && (instance.ReleaseID != state.DesiredReleaseID || instance.SpecOutdated) { + if instance.Phase.Serving() && !instance.RolloutBlocked && (instance.ReleaseID != state.DesiredReleaseID || instance.SpecOutdated || instance.NodeReplacement) { outdated = append(outdated, instance) } } diff --git a/controlplane/trinopool/plan_test.go b/controlplane/trinopool/plan_test.go index bd3bbf33..e14d9937 100644 --- a/controlplane/trinopool/plan_test.go +++ b/controlplane/trinopool/plan_test.go @@ -5,6 +5,33 @@ import ( "testing" ) +func TestPlanNodeReplacementWaitsForServingReplacement(t *testing.T) { + for _, test := range []struct { + name string + serving, surge, floor int + action PlanAction + }{ + {"at desired count above floor", 3, 1, 2, PlanActionCreate}, + {"replacement serving", 4, 1, 2, PlanActionDrain}, + {"no surge budget", 3, 0, 2, PlanActionNone}, + {"serving floor", 3, 1, 3, PlanActionCreate}, + } { + t.Run(test.name, func(t *testing.T) { + state := servingPool("r1", test.serving) + state.MinServing = test.floor + state.MaxSurge = test.surge + state.Instances[0].NodeReplacement = true + plan := PlanNext(state) + if plan.Action != test.action || plan.Repair || plan.RepairFor != "" { + t.Fatalf("node replacement plan = %+v", plan) + } + if plan.Action == PlanActionDrain && plan.InstanceID != state.Instances[0].ID { + t.Fatalf("drained unrelated member: %+v", plan) + } + }) + } +} + func TestPlanConfigurationOnlyRollout(t *testing.T) { for _, test := range []struct { name string diff --git a/docs/runbooks/trino-pool-node-replacement.md b/docs/runbooks/trino-pool-node-replacement.md new file mode 100644 index 00000000..88b3293f --- /dev/null +++ b/docs/runbooks/trino-pool-node-replacement.md @@ -0,0 +1,117 @@ +# Shared-pool node replacement + +This optional controller replaces a shared-pool Trino instance before its original +compute is removed for a voluntary infrastructure change. It does not change legacy +Trino, intercept SIGTERM, expire transactions, or convert failure into successful drain. + +## Enable safely + +1. Deploy the Duckgres image and migration 000043 with + `DUCKGRES_TRINO_POOL_NODE_DISRUPTION_ENABLED=false` (the default). + Wait for every control-plane replica to run that image before enabling the + chart flag. An older leader cannot honor the new replacement evidence. +2. Review the required permissions: namespace-scoped pod `get/list/patch`, + ReplicaSet `get`, existing Deployment reads, cluster-scoped Node `get/patch`, + and NodeClaim `get/list`. Kubernetes RBAC cannot restrict dynamic Node patch + permission by label. Code checks exact workload ownership and Node/NodeClaim + UIDs before every cordon. Grant this only to the pool operator. +3. Add `karpenter.sh/do-not-disrupt: "true"` to both blueprint pod templates and + enable the environment variable alongside the existing pool/operator flags. + The source binary alone does not inject this annotation with the feature off. + Every node eligible for these pods must have a `karpenter.sh/v1` NodeClaim + owner, with matching Node/NodeClaim UIDs, names, and provider IDs. Verify the + pool's scheduling constraints before enabling this feature. Non-Karpenter + nodes and older owner-reference API versions are not supported. +4. Confirm every live coordinator and worker has the annotation before testing + disruption. Existing immutable Deployments are not rolled to add it: the + controller patches only the metadata of their exact owned pods. +5. In development, run the shared-pool E2E harness with + `E2E_TRINO_POOL_NODE_PROTECTION=1`. This checks live pod protection, not an + actual EC2/Karpenter replacement. A controlled isolated replacement test is + still required before claiming end-to-end zero-loss infrastructure rotation. + +Do not enable a finite NodeClaim `terminationGracePeriod`. Karpenter can override +pod protection after that deadline. The controller rejects these claims rather +than pretending their running work is protected indefinitely. + +## What happens + +The leader protects all live instances before advancing the lifecycle. Pods must +belong to the recorded Deployment through an exact ReplicaSet UID chain; labels +alone do not establish ownership. A protection pass has a 30-second budget and +rotates its starting instance. Individual Kubernetes calls use a 10-second budget. +One paginated NodeClaim inventory is shared within the pass, capped at 10,000 +claims. Candidate validation and admission read fresh Nodes and fetch only their +named NodeClaims, instead of listing every claim again. These reads share a +10-second inspection deadline. Each candidate check caches its own claim reads +and discards them afterward; it cannot reuse the protection pass's older snapshot. +API errors never prove drift. Missing pods do not invalidate positive drift +evidence from other exactly identified nodes. + +The first exact Node/NodeClaim identity with `Drifted=True`, or a deletion timestamp, +is recorded once in `node_replacement_evidence`. Later observations cannot rewrite +that request. The node is then cordoned with UID and resource-version preconditions +and the `posthog.com/trino-node-retirement` annotation. This excludes replacement +pods from the old node without removing its running pods. Cross-system fencing is +not atomic: the database authorizes the intent and Kubernetes rejects a changed +node identity or resource version. A delayed cordon is an idempotent scheduling +exclusion, never permission to delete an instance. + +The existing planner creates and admits a replacement using the normal surge slot, +then drains the original. It never uses repair capacity merely because a node +drifted. With `max_surge=0`, replacement waits safely. The Gateway keeps queries, +transactions, and result obligations on the original instance until they finish. +Only normal retirement authorization permits UID-scoped deletion of its resources. + +New candidates on drifted, deleting, unschedulable, or unverifiable nodes cannot +be admitted. Admission whose response was lost still replays the same Gateway +request; a concurrent successful admission cannot be deleted as a failed candidate. + +API errors and missing or pending worker inventory block new cordons. Pending +pods alone are normal convergence for a new candidate. Independently verified +drift is still recorded when another pod or node is unavailable. Valid, durable +replacement evidence permits the normal create/admit/drain sequence to continue +despite later observation faults. Missing or malformed evidence does not. +A candidate with definitively invalid placement fails through the guarded +candidate-retirement path. `FAILED_PREPARING` is not terminal: its capacity slot +stays charged until cleanup finishes. A consistently unsupported node placement +can still cause repeated creation failures after cleanup; fix the scheduling +constraints or disable new node handling instead of waiting for those retries. +A candidate waiting for scheduling or an unavailable API retains its slot and retries. +Health detection, explicit administrative recovery, and irreversible retirement +continue through their existing guards. A node lookup failure is neither drift +evidence nor proof that the admitted process died. + +## Observe and recover + +The first accepted request emits `Trino pool requested voluntary node replacement` +with the instance, Node UID, NodeClaim UID, and reason. The authenticated instance +inventory exposes a typed, read-only `node_replacement` object. Existing blocked +plan messages explain an exhausted surge slot or serving-floor constraint. + +Recorded replacement is deliberate: it completes even if drift later clears. +Disabling the feature stops new observation, protection, and cordons, but does not +cancel an already-recorded replacement. Do not clear its database evidence to +reuse the old instance. + +The controller never uncordons nodes. Normally Karpenter removes the emptied node. +If drift clears and unrelated workloads keep it occupied, an owned cordon may +remain. Before manually uncordoning, inspect the exact Node UID, retirement +annotation, matching NodeClaim UID, and all managed instances on that node. Confirm +their recorded replacements have retired. Do not undo an unrelated administrator's +cordon or assume that one retired instance proves all co-located work has left. + +## Guarantee boundary + +Karpenter 1.9.0's [eviction eligibility](https://github.com/kubernetes-sigs/karpenter/blob/v1.9.0/pkg/utils/pod/scheduling.go) +excludes protected pods but still waits for them in its drain accounting. +Its [termination controller](https://github.com/kubernetes-sigs/karpenter/blob/v1.9.0/pkg/controllers/node/termination/terminator/terminator.go) +only applies forced pod expiry when a node termination deadline exists. Therefore +ordinary expiration with no NodeClaim termination grace period can use the same +protected deletion-observation path; no NodePool expiry change is required. + +Hardware failure, interruption deadlines, force deletion, kubelet failure, direct +pod deletion, and finite node termination deadlines remain outside the zero-loss +claim. Protection cannot recover an already-dead coordinator or worker. Do not +trigger drift on shared nodes to validate this feature; use an isolated approved +test pool and retain client results and transaction evidence throughout the test. diff --git a/tests/configstore/migrations_postgres_test.go b/tests/configstore/migrations_postgres_test.go index 525986de..d25addfd 100644 --- a/tests/configstore/migrations_postgres_test.go +++ b/tests/configstore/migrations_postgres_test.go @@ -60,7 +60,10 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) { requireGooseMigrationRecorded(t, db, 40) requireGooseMigrationRecorded(t, db, 41) requireGooseMigrationRecorded(t, db, 42) - requireGooseLatestVersion(t, db, 42) + requireGooseMigrationRecorded(t, db, 43) + requireGooseLatestVersion(t, db, 43) + requireColumnType(t, db, "duckgres_trino_pool_instances", "node_replacement_evidence", "jsonb") + requireColumnNullable(t, db, "duckgres_trino_pool_instances", "node_replacement_evidence") requireTablePresent(t, db, "duckgres_trino_pool_recoveries") requireTablePresent(t, db, "duckgres_trino_cell_lifecycle") for _, column := range []string{"reconcile_owner", "reconcile_epoch", "intent_sequence", "intent", "admission_epoch", "freeze_operation_id", "freeze_stable", "certificate"} { @@ -391,7 +394,7 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { DROP TABLE IF EXISTS duckgres_trino_pool_instances; DROP TABLE IF EXISTS duckgres_trino_pools; DROP FUNCTION IF EXISTS duckgres_select_trino_backend_on_enable(); - DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42); + DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43); `).Error; err != nil { t.Fatalf("downgrade baseline schema to pre-v9 shape: %v", err) } @@ -444,7 +447,10 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { requireGooseMigrationRecorded(t, upgradedDB, 38) requireGooseMigrationRecorded(t, upgradedDB, 41) requireGooseMigrationRecorded(t, upgradedDB, 42) - requireGooseLatestVersion(t, upgradedDB, 42) + requireGooseMigrationRecorded(t, upgradedDB, 43) + requireGooseLatestVersion(t, upgradedDB, 43) + requireColumnType(t, upgradedDB, "duckgres_trino_pool_instances", "node_replacement_evidence", "jsonb") + requireColumnNullable(t, upgradedDB, "duckgres_trino_pool_instances", "node_replacement_evidence") requireTablePresent(t, upgradedDB, "duckgres_trino_pool_recoveries") requireColumnPresent(t, upgradedDB, "duckgres_reshard_operations", "password_url") requireTablePresent(t, upgradedDB, "duckgres_worker_spawn_log") @@ -503,7 +509,7 @@ func TestConfigStoreSQLMigration34VersionsExistingAndNewOrgs(t *testing.T) { DROP TABLE IF EXISTS duckgres_trino_pool_instances; DROP TABLE IF EXISTS duckgres_trino_pools; DROP FUNCTION IF EXISTS duckgres_select_trino_backend_on_enable(); - DELETE FROM goose_db_version WHERE version_id IN (34, 35, 36, 37, 38, 39, 40, 41, 42); + DELETE FROM goose_db_version WHERE version_id IN (34, 35, 36, 37, 38, 39, 40, 41, 42, 43); `).Error; err != nil { t.Fatalf("restore pre-migration-34 schema: %v", err) } diff --git a/tests/configstore/trino_pool_node_replacement_test.go b/tests/configstore/trino_pool_node_replacement_test.go new file mode 100644 index 00000000..c720f7c2 --- /dev/null +++ b/tests/configstore/trino_pool_node_replacement_test.go @@ -0,0 +1,49 @@ +//go:build linux || darwin + +package configstore_test + +import ( + "context" + "encoding/json" + "errors" + "testing" + + cpconfigstore "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/controlplane/trinopool" +) + +func TestTrinoPoolNodeReplacementFirstEvidenceIsFencedAndImmutable(t *testing.T) { + ctx := context.Background() + store := newPoolStore(t) + lease := claimPool(t, store, "first") + if err := store.CreateTrinoPoolInstance(ctx, lease, newInstance("drifted-instance", trinopool.PhaseServing)); err != nil { + t.Fatal(err) + } + evidence := trinopool.NodeReplacementEvidence{NodeName: "node-a", NodeUID: "node-uid-a", NodeClaimName: "claim-a", NodeClaimUID: "claim-uid-a", Reason: "Drifted"} + if err := store.RecordTrinoPoolNodeReplacement(ctx, lease, "drifted-instance", evidence); err != nil { + t.Fatal(err) + } + changed := evidence + changed.NodeUID = "node-uid-b" + if err := store.RecordTrinoPoolNodeReplacement(ctx, lease, "drifted-instance", changed); err != nil { + t.Fatal(err) + } + instance, err := store.GetTrinoPoolInstance(ctx, "drifted-instance") + if err != nil { + t.Fatal(err) + } + var recorded trinopool.NodeReplacementEvidence + if instance.NodeReplacementEvidence == nil { + t.Fatal("missing durable evidence") + } + if err := json.Unmarshal([]byte(*instance.NodeReplacementEvidence), &recorded); err != nil { + t.Fatal(err) + } + if recorded != evidence { + t.Fatalf("first evidence rewritten: %+v", recorded) + } + _ = claimPool(t, store, "successor") + if err := store.RecordTrinoPoolNodeReplacement(ctx, lease, "drifted-instance", evidence); !errors.Is(err, cpconfigstore.ErrTrinoPoolConflict) { + t.Fatalf("stale leader reused evidence authority: %v", err) + } +} diff --git a/tests/mw-dev/e2e/harness.sh b/tests/mw-dev/e2e/harness.sh index 4a7a08e5..4bde5c23 100755 --- a/tests/mw-dev/e2e/harness.sh +++ b/tests/mw-dev/e2e/harness.sh @@ -1192,6 +1192,15 @@ trino_shared_pool_active() { log "shared pool OK [structure]: $ready ready instance(s), each with its own service and workers" + if [ "${E2E_TRINO_POOL_NODE_PROTECTION:-0}" = "1" ]; then + for pods in "$body" "$workers"; do + printf %s "$pods" | jq -e 'all(.items[]; + .metadata.annotations["karpenter.sh/do-not-disrupt"] == "true")' >/dev/null \ + || fail "shared pool: a coordinator or worker lacks node-disruption protection" + done + log "shared pool OK [node protection]: coordinators and workers retain do-not-disrupt" + fi + # Validate the recovery UI's read-only contract without authorizing retirement. # Absent-pod recovery needs an isolated pool and destructive authorization. # This shared-pool Job must not delete coordinators to manufacture that evidence. @@ -1200,7 +1209,7 @@ trino_shared_pool_active() { || fail "shared pool: recovery inventory request failed" printf %s "$inventory" | jq -e --arg cell "$pool" ' .cell == $cell and (.instances | length > 0) and all(.instances[]; - (keys == ["gateway_state", "instance_id", "phase", "phase_changed_at"]) + ((keys - ["node_replacement"]) == ["gateway_state", "instance_id", "phase", "phase_changed_at"]) and (.instance_id | type == "string" and length > 0) and .phase != "RETIRED" and .phase != "FAILURE_RETIRED")' >/dev/null \ || fail "shared pool: recovery inventory is not scoped and sanitized"