diff --git a/.github/workflows/replace-wedding-standby.yaml b/.github/workflows/replace-wedding-standby.yaml new file mode 100644 index 0000000000..46640b5acd --- /dev/null +++ b/.github/workflows/replace-wedding-standby.yaml @@ -0,0 +1,107 @@ +name: Replace Failed Wedding Standby + +on: + pull_request: + branches: [main] + paths: + - .github/workflows/replace-wedding-standby.yaml + - scripts/replace-wedding-standby/** + merge_group: + workflow_dispatch: + inputs: + confirm: + description: "Type exactly: retain-volumes-replace-failed-standby" + required: true + type: string + cluster_uid: + description: Current Wedding Cluster UID from an OIDC read + required: true + type: string + pod_uid: + description: Current failed standby Pod UID from an OIDC read + required: true + type: string + +permissions: {} + +jobs: + tests: + name: 🛡️ Validate standby replacement safety + runs-on: ubuntu-latest + timeout-minutes: 10 + permissions: + contents: read + steps: + - name: 📑 Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - name: ⚙️ Setup Go + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + cache: false + - name: Test complete observations and conditional writes + run: go test -race -count=1 -timeout=5m ./scripts/replace-wedding-standby + + repair: + name: 💒 Replace failed standby without deleting its storage + if: ${{ github.event_name == 'workflow_dispatch' }} + needs: tests + runs-on: ubuntu-latest + timeout-minutes: 40 + environment: prod + permissions: + contents: read + concurrency: + group: prod-deploy + cancel-in-progress: false + queue: max + env: + WEDDING_REPAIR_CONFIRM: ${{ inputs.confirm }} + GH_TOKEN: ${{ github.token }} + steps: + - name: Require main and explicit confirmation + shell: bash + env: + CONFIRM: ${{ inputs.confirm }} + DISPATCH_REF: ${{ github.ref }} + run: | + set -euo pipefail + [[ "${DISPATCH_REF}" == refs/heads/main && "${CONFIRM}" == retain-volumes-replace-failed-standby && "${GITHUB_RUN_ATTEMPT}" == 1 ]] + - name: 📑 Checkout reviewed main revision + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + ref: ${{ github.sha }} + - name: Require the current main recovery source + run: ./scripts/verify-prod-recovery-source.sh "${GITHUB_SHA}" + - name: ⚙️ Setup Go + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + cache: false + - name: 🔑 Restore production kubeconfig + shell: bash + env: + KUBE_CONFIG: ${{ secrets.KUBE_CONFIG }} + run: | + set -euo pipefail + test -n "${KUBE_CONFIG}" + umask 077 + mkdir -p ~/.kube + printf '%s' "${KUBE_CONFIG}" >~/.kube/config + chmod 600 ~/.kube/config + - name: 🎯 Select stable production API endpoint + shell: bash + env: + HCLOUD_TOKEN: ${{ secrets.HCLOUD_TOKEN }} + run: ./scripts/use-prod-stable-api-endpoint.sh + - name: Preserve old storage and prove the replacement healthy + shell: bash + env: + CLUSTER_UID: ${{ inputs.cluster_uid }} + POD_UID: ${{ inputs.pod_uid }} + run: | + set -euo pipefail + go run ./scripts/replace-wedding-standby --execute --cluster-uid "${CLUSTER_UID}" --pod-uid "${POD_UID}" diff --git a/scripts/replace-wedding-standby/README.md b/scripts/replace-wedding-standby/README.md new file mode 100644 index 0000000000..b61f40e9ac --- /dev/null +++ b/scripts/replace-wedding-standby/README.md @@ -0,0 +1,64 @@ +# Failed Wedding standby replacement + +This command repairs only the failed `wedding-db-1` replica. Its default mode is +an OIDC-only, read-only plan. Mutations require the first, explicitly confirmed +main dispatch of `Replace Failed Wedding Standby`, behind the `prod` environment +and the shared production deployment lock. Publishing or merging this procedure +does not authorize running it; the maintainer must approve the particular repair. + +The procedure requires three observed database instances, a stable healthy primary, +one other healthy replica, healthy archiving, and a completed cluster-bound backup +no older than 24 hours. Current Cluster and failed Pod UIDs must be provided. The +operator must be the audited, fully rolled-out CloudNativePG 1.30.1. An unknown +read, different volume layout, active database Job, existing fence, changed +identity or changed primary stops the repair. + +The failed replica's single existing volume is changed to `Retain`. The replica +is fenced, and its own metrics must acknowledge that PostgreSQL is fenced before +the claim is detached. Detachment follows the operator's +[`destroy --keep-pvc` procedure](https://github.com/cloudnative-pg/cloudnative-pg/blob/v1.30.1/internal/cmd/plugin/destroy/destroy.go): +remove only the verified Cluster owner reference and mark the claim `detached` +before deleting the exact failed Pod with UID and resource-version preconditions. +The operator's [fenced-candidate ordering](https://github.com/cloudnative-pg/cloudnative-pg/blob/v1.30.1/pkg/postgres/status.go) +keeps the fenced replica out of primary election. No PVC, PV, backup, primary, +healthy replica, disruption budget or namespace is deleted or relaxed. + +The fence is removed only after the old Pod is absent and the same protected +peers remain healthy. Success requires two separated, complete samples showing +three Ready instances, fresh replacement storage, healthy archiving, and the old +claim and volume still bound to their original identities. Only then may ordinary +maintenance proceed. This proves database repair, not overall platform health. + +## Read-only plan + +From current main, get the current Cluster and failed Pod UIDs with OIDC reads, +then run: + +```sh +go run ./scripts/replace-wedding-standby --cluster-uid "$CLUSTER_UID" --pod-uid "$POD_UID" +``` + +The command reports `PLAN=PASS no writes` only after complete observations pass. +The protected workflow requires confirmation +`retain-volumes-replace-failed-standby` and those same UIDs. It rechecks that its +validated checkout is current main before every conditional write. Never rerun +a failed dispatch: a new attempt requires fresh inspection and a new approval +for any continuation beyond the original procedure. + +## Failure and retained data + +A failure returns `REPAIR=HOLD`, never health clearance. Earlier successful +steps may remain: a Retain volume, a fenced failed replica, or a detached claim. +There is no automatic destructive cleanup, reverse ownership change, WAL reset, +or repeated mutation. Inspect the current state with OIDC reads and prepare a +reviewed continuation. Keep the old claim and volume for diagnosis; retirement +is a separately approved data-lifecycle operation. + +## Offline validation + +```sh +go test -race -count=1 ./scripts/replace-wedding-standby +``` + +The workflow runs this package on its pull request, merge group and explicit +dispatch before any production credentials are restored. diff --git a/scripts/replace-wedding-standby/main.go b/scripts/replace-wedding-standby/main.go new file mode 100644 index 0000000000..2e63b4a55f --- /dev/null +++ b/scripts/replace-wedding-standby/main.go @@ -0,0 +1,754 @@ +// Command replace-wedding-standby provides a narrowly scoped, volume-preserving +// repair for the failed Wedding replica. Without --execute it only reads. +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "os" + "os/exec" + "regexp" + "strings" + "time" +) + +type object = map[string]any + +type inventory struct { + cluster object + operator object + pods, claims, backups, jobs []object + volumes map[string]object +} + +type identity struct{ name, uid string } +type plan struct { + primary, healthy, target identity + claims []object +} + +type options struct { + clusterUID, podUID string + now time.Time + fenced, detached bool +} + +type client struct { + command func(context.Context, []string, []byte) ([]byte, error) + source func(context.Context) error + wait func(context.Context) error + now func() time.Time +} + +// read accepts only a complete JSON response from a successful Kubernetes read. +func (c client) read(ctx context.Context, args ...string) (object, error) { + b, err := c.command(ctx, append([]string{"get"}, args...), nil) + if err != nil { + return nil, fmt.Errorf("required %s read failed", args[0]) + } + var result object + if err = json.Unmarshal(b, &result); err != nil || result == nil { + return nil, fmt.Errorf("required %s response is incomplete", args[0]) + } + return result, nil +} + +// snapshot joins the cluster, operator, instances, old storage, jobs and backups. +func (c client) snapshot(ctx context.Context) (inventory, error) { + s := inventory{volumes: map[string]object{}} + var err error + s.cluster, err = c.read(ctx, "cluster.postgresql.cnpg.io", clusterName, "-n", namespace, "-o", "json") + if err != nil { + return s, err + } + s.operator, err = c.read(ctx, "deployment", "cloudnative-pg", "-n", "cnpg-system", "-o", "json") + if err != nil { + return s, err + } + for _, entry := range []struct { + kind, selector string + dst *[]object + }{ + {"pods", "cnpg.io/cluster=" + clusterName, &s.pods}, + {"pvc", "cnpg.io/instanceName=" + targetName, &s.claims}, + {"backups.postgresql.cnpg.io", "cnpg.io/cluster=" + clusterName, &s.backups}, + {"jobs", "cnpg.io/cluster=" + clusterName, &s.jobs}, + } { + obj, e := c.read(ctx, entry.kind, "-n", namespace, "-l", entry.selector, "-o", "json") + if e != nil { + return s, e + } + items, ok := value(obj, "items").([]any) + if !ok { + return s, errors.New("incomplete resource listing") + } + *entry.dst = list(obj, "items") + if len(*entry.dst) != len(items) { + return s, errors.New("malformed resource listing") + } + } + for _, claim := range s.claims { + name := str(claim, "spec", "volumeName") + if name == "" { + return s, errors.New("missing claim volume binding") + } + v, e := c.read(ctx, "pv", name, "-o", "json") + if e != nil { + return s, e + } + s.volumes[name] = v + } + return s, nil +} + +// proveSource binds the next fresh observation to the validated current main. +func (c client) proveSource(ctx context.Context) error { + if c.source == nil || c.source(ctx) != nil { + return errors.New("current-main source proof failed; no further write") + } + return nil +} + +// write sends a single conditional mutation after its proof and fresh observation. +// The repair caller never retries a rejected request or performs cleanup writes. +func (c client) write(ctx context.Context, args []string, body any) error { + b, err := json.Marshal(body) + if err != nil { + return err + } + if _, err = c.command(ctx, args, b); err != nil { + return fmt.Errorf("conditional %s failed; no retry or cleanup write", args[0]) + } + return nil +} + +// tests pins a JSON patch to one observed resource identity and version. +func tests(o object) []object { + return []object{{"op": "test", "path": "/metadata/uid", "value": id(o).uid}, {"op": "test", "path": "/metadata/resourceVersion", "value": str(o, "metadata", "resourceVersion")}} +} + +// testPath creates a semantic precondition without changing a resource. +func testPath(path string, v any) object { return object{"op": "test", "path": path, "value": v} } + +// editPath creates one reviewed JSON patch operation. +func editPath(op, path string, v any) object { return object{"op": op, "path": path, "value": v} } + +// patch scopes mutations to the observed resource and explicit API group. +func (c client) patch(ctx context.Context, kind string, o object, ops []object) error { + if kind == "cluster" { + kind = "cluster.postgresql.cnpg.io" + } + args := []string{"patch", kind, id(o).name, "--type=json", "--patch-file=/dev/stdin"} + if kind != "pv" { + args = append(args, "-n", namespace) + } + return c.write(ctx, args, ops) +} + +// fencePatch changes only this repair's fence while pinning the primary and size. +func fencePatch(cluster object, remove bool) []object { + ops := append(tests(cluster), testPath("/status/currentPrimary", str(cluster, "status", "currentPrimary")), testPath("/status/targetPrimary", str(cluster, "status", "targetPrimary")), testPath("/spec/instances", float64(3))) + path := "/metadata/annotations/cnpg.io~1fencedInstances" + if remove { + return append(ops, testPath(path, `["wedding-db-1"]`), object{"op": "remove", "path": path}) + } + if value(cluster, "metadata", "annotations") == nil { + return append(ops, editPath("add", "/metadata/annotations", object{fenceKey: `["wedding-db-1"]`})) + } + return append(ops, editPath("add", path, `["wedding-db-1"]`)) +} + +// detachPatch preserves the claim and removes only its verified cluster owner. +func detachPatch(claim object, uid string) ([]object, error) { + ops := append(tests(claim), testPath("/spec/volumeName", str(claim, "spec", "volumeName"))) + for i, ref := range list(claim, "metadata", "ownerReferences") { + if str(ref, "uid") == uid && value(ref, "controller") == true { + path := fmt.Sprintf("/metadata/ownerReferences/%d", i) + return append(ops, testPath(path, ref), object{"op": "remove", "path": path}, editPath("add", "/metadata/annotations/cnpg.io~1pvcStatus", "detached")), nil + } + } + return nil, errors.New("cluster ownership disappeared before detachment") +} + +// pause separates observations and respects the enclosing operation deadline. +func (c client) pause(ctx context.Context) error { + if c.wait != nil { + return c.wait(ctx) + } + timer := time.NewTimer(20 * time.Second) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +// fenced requires exactly one positive gauge from the target's own exporter. +func (c client) fenced(ctx context.Context) bool { + b, err := c.command(ctx, []string{"get", "--raw", "/api/v1/namespaces/" + namespace + "/pods/" + targetName + ":9187/proxy/metrics"}, nil) + if err != nil { + return false + } + count := 0 + for _, line := range strings.Split(string(b), "\n") { + fields := strings.Fields(line) + if len(fields) > 0 && fields[0] == "cnpg_collector_fencing_on" { + if len(fields) != 2 || fields[1] != "1" { + return false + } + count++ + } + } + return count == 1 +} + +// stable refuses drift in any protected instance or old storage identity. +func stable(s inventory, o options, p plan, claimID, volumeID identity) error { + current, err := validate(s, o) + if err != nil { + return err + } + if current.primary != p.primary || current.healthy != p.healthy || current.target != p.target || id(current.claims[0]) != claimID || id(s.volumes[str(current.claims[0], "spec", "volumeName")]) != volumeID { + return errors.New("protected instance or storage identity changed") + } + return nil +} + +// retained proves the old claim and volume remain bound with a Retain policy. +func retained(s inventory, claimID, volumeID identity, detached bool) bool { + if len(s.claims) != 1 { + return false + } + claim := s.claims[0] + volume := s.volumes[str(claim, "spec", "volumeName")] + if !validMeta(claim, namespace) || id(claim) != claimID || str(claim, "status", "phase") != "Bound" || !validMeta(volume, "") || id(volume) != volumeID || str(volume, "status", "phase") != "Bound" || str(volume, "spec", "persistentVolumeReclaimPolicy") != "Retain" || str(volume, "spec", "claimRef", "uid") != claimID.uid || str(volume, "spec", "claimRef", "name") != claimID.name || str(volume, "spec", "claimRef", "namespace") != namespace { + return false + } + if detached { + if str(claim, "metadata", "annotations", "cnpg.io/pvcStatus") != "detached" { + return false + } + for _, ref := range list(claim, "metadata", "ownerReferences") { + if value(ref, "controller") == true { + return false + } + } + } + return true +} + +// repair either validates without writes or performs the approved narrow sequence. +func repair(ctx context.Context, c client, o options, execute bool) error { + s, err := c.snapshot(ctx) + if err != nil { + return err + } + p, err := validate(s, o) + if err != nil { + return err + } + if !execute { + return nil + } + claimID := id(p.claims[0]) + volume := s.volumes[str(p.claims[0], "spec", "volumeName")] + volumeID := id(volume) + // Pin each write to a new complete observation, never retrying a rejected CAS. + refresh := func() error { + var e error + if e = c.proveSource(ctx); e != nil { + return e + } + s, e = c.snapshot(ctx) + if e != nil { + return e + } + if c.now != nil { + o.now = c.now() + } + return stable(s, o, p, claimID, volumeID) + } + if err = refresh(); err != nil { + return err + } + volume = s.volumes[volumeID.name] + if str(volume, "spec", "persistentVolumeReclaimPolicy") == "Delete" { + ops := append(tests(volume), testPath("/spec/claimRef", value(volume, "spec", "claimRef")), testPath("/spec/persistentVolumeReclaimPolicy", "Delete"), editPath("replace", "/spec/persistentVolumeReclaimPolicy", "Retain")) + if err = c.patch(ctx, "pv", volume, ops); err != nil { + return err + } + } + if err = refresh(); err != nil { + return err + } + if !retained(s, claimID, volumeID, false) { + return errors.New("volume Retain policy was not read back") + } + if err = c.patch(ctx, "cluster", s.cluster, fencePatch(s.cluster, false)); err != nil { + return err + } + o.fenced = true + fenceCtx, cancel := context.WithTimeout(ctx, 10*time.Minute) + defer cancel() + for { + if err = refresh(); err != nil { + return err + } + if c.fenced(fenceCtx) { + break + } + if err = c.pause(fenceCtx); err != nil { + return errors.New("instance did not acknowledge fencing; storage unchanged") + } + } + if err = refresh(); err != nil { + return err + } + if !retained(s, claimID, volumeID, false) || !c.fenced(ctx) { + return errors.New("volume retention or fencing proof lost") + } + ops, err := detachPatch(s.claims[0], o.clusterUID) + if err != nil { + return err + } + if err = c.patch(ctx, "pvc", s.claims[0], ops); err != nil { + return err + } + o.detached = true + if err = refresh(); err != nil { + return err + } + if !retained(s, claimID, volumeID, true) || !c.fenced(ctx) { + return errors.New("detachment or fencing proof lost; pod preserved") + } + var target object + for _, pod := range s.pods { + if id(pod) == p.target { + target = pod + } + } + deletion := object{"apiVersion": "v1", "kind": "DeleteOptions", "preconditions": object{"uid": p.target.uid, "resourceVersion": str(target, "metadata", "resourceVersion")}, "propagationPolicy": "Background"} + if err = c.write(ctx, []string{"delete", "--raw", "/api/v1/namespaces/" + namespace + "/pods/" + targetName, "-f", "-"}, deletion); err != nil { + return err + } + for { + if err = c.proveSource(ctx); err != nil { + return err + } + s, err = c.snapshot(ctx) + if err != nil { + return err + } + if err = protectedPeers(s, o, p); err != nil { + return err + } + if !retained(s, claimID, volumeID, true) { + return errors.New("old volume retention proof lost") + } + oldPresent := false + for _, pod := range s.pods { + if id(pod) == p.target { + oldPresent = true + } + } + if !oldPresent { + break + } + if err = c.pause(ctx); err != nil { + return err + } + } + if str(s.cluster, "metadata", "annotations", fenceKey) != `["wedding-db-1"]` { + return errors.New("repair fence changed; refusing to remove it") + } + if err = c.patch(ctx, "cluster", s.cluster, fencePatch(s.cluster, true)); err != nil { + return err + } + // Require two separated complete healthy samples before returning clearance. + good := 0 + for { + s, err = c.snapshot(ctx) + if err != nil { + return err + } + if err = protectedPeers(s, o, p); err != nil { + return err + } + if !retained(s, claimID, volumeID, true) { + return errors.New("old volume retention proof lost") + } + ok, err := c.recovered(ctx, s, o, p, claimID) + if err != nil { + return err + } + if ok { + good++ + } else { + good = 0 + } + if good == 2 { + return nil + } + if err = c.pause(ctx); err != nil { + return errors.New("replacement did not reach a stable healthy state") + } + } +} + +// protectedPeers requires the original healthy primary and standby throughout. +func protectedPeers(s inventory, o options, p plan) error { + if !auditedOperator(s.operator) || !condition(s.cluster, "ContinuousArchiving") || !condition(s.cluster, "LastBackupSucceeded") { + return errors.New("operator or backup health changed during repair") + } + if !validMeta(s.cluster, namespace) || id(s.cluster).uid != o.clusterUID || str(s.cluster, "status", "currentPrimary") != p.primary.name || str(s.cluster, "status", "targetPrimary") != p.primary.name || num(s.cluster, "spec", "instances") != 3 || value(s.cluster, "spec", "nodeMaintenanceWindow", "inProgress") == true { + return errors.New("cluster or primary changed during repair") + } + for _, peer := range []identity{p.primary, p.healthy} { + found := false + for _, pod := range s.pods { + if id(pod) == peer { + role := "replica" + if peer == p.primary { + role = "primary" + } + found = validMeta(pod, namespace) && owned(pod, o.clusterUID) && condition(pod, "Ready") && str(pod, "metadata", "labels", "cnpg.io/instanceRole") == role + } + } + if !found { + return errors.New("protected peer identity, readiness or role changed") + } + } + return nil +} + +// recovered joins three Ready instances to fresh replacement storage, not the old claim. +func (c client) recovered(ctx context.Context, s inventory, o options, p plan, oldClaim identity) (bool, error) { + if num(s.cluster, "status", "readyInstances") != 3 || !condition(s.cluster, "Ready") || !condition(s.cluster, "ContinuousArchiving") || str(s.cluster, "metadata", "annotations", fenceKey) != "" || len(s.pods) != 3 { + return false, nil + } + seenUID, seenName := map[string]bool{}, map[string]bool{} + for _, pod := range s.pods { + if seenUID[id(pod).uid] || seenName[id(pod).name] { + return false, errors.New("duplicate recovery observations") + } + seenUID[id(pod).uid] = true + seenName[id(pod).name] = true + if !validMeta(pod, namespace) || !owned(pod, o.clusterUID) || !condition(pod, "Ready") { + return false, nil + } + if id(pod) == p.primary || id(pod) == p.healthy { + continue + } + if id(pod).uid == p.target.uid || str(pod, "metadata", "labels", "cnpg.io/instanceRole") != "replica" { + return false, errors.New("replacement identity or role is unsafe") + } + mounted := 0 + for _, v := range list(pod, "spec", "volumes") { + name := str(v, "persistentVolumeClaim", "claimName") + if name == "" { + continue + } + mounted++ + if name == oldClaim.name { + return false, errors.New("replacement attempted to reuse retained storage") + } + claim, err := c.read(ctx, "pvc", name, "-n", namespace, "-o", "json") + if err != nil { + return false, err + } + if !validMeta(claim, namespace) || !owned(claim, o.clusterUID) || id(claim).uid == oldClaim.uid || str(claim, "status", "phase") != "Bound" { + return false, errors.New("replacement claim is not bound and owned") + } + pv, err := c.read(ctx, "pv", str(claim, "spec", "volumeName"), "-o", "json") + if err != nil { + return false, err + } + if !validMeta(pv, "") || str(pv, "status", "phase") != "Bound" || str(pv, "spec", "claimRef", "uid") != id(claim).uid || str(pv, "spec", "claimRef", "name") != name || str(pv, "spec", "claimRef", "namespace") != namespace { + return false, errors.New("replacement volume binding is incomplete") + } + } + if mounted != 1 { + return false, errors.New("replacement storage layout changed") + } + } + return true, nil +} + +const ( + namespace = "wedding-app" + clusterName = "wedding-db" + targetName = "wedding-db-1" + fenceKey = "cnpg.io/fencedInstances" +) + +// value traverses decoded Kubernetes objects without accepting incompatible shapes. +func value(o object, path ...string) any { + var v any = o + for _, key := range path { + m, ok := v.(map[string]any) + if !ok { + return nil + } + v = m[key] + } + return v +} + +// str reads a string field; absence or a wrong type cannot satisfy a string gate. +func str(o object, path ...string) string { s, _ := value(o, path...).(string); return s } + +// num reads the numeric form produced by standard JSON decoding. +func num(o object, path ...string) float64 { n, _ := value(o, path...).(float64); return n } + +// list refuses a mixed or malformed resource collection instead of dropping entries. +func list(o object, path ...string) []object { + var result []object + items, _ := value(o, path...).([]any) + for _, v := range items { + m, ok := v.(map[string]any) + if !ok { + return nil + } + result = append(result, m) + } + return result +} + +// condition accepts only an explicitly True condition of the requested type. +func condition(o object, name string) bool { + for _, c := range list(o, "status", "conditions") { + if str(c, "type") == name { + return str(c, "status") == "True" + } + } + return false +} + +// id binds names to immutable Kubernetes UIDs. +func id(o object) identity { return identity{str(o, "metadata", "name"), str(o, "metadata", "uid")} } + +// validMeta requires a versioned, nonterminating resource in the expected scope. +func validMeta(o object, ns string) bool { + return id(o).name != "" && id(o).uid != "" && str(o, "metadata", "namespace") == ns && str(o, "metadata", "resourceVersion") != "" && value(o, "metadata", "deletionTimestamp") == nil +} + +// owned requires exactly one controller owner from this CloudNativePG cluster. +func owned(o object, uid string) bool { + count := 0 + for _, ref := range list(o, "metadata", "ownerReferences") { + if value(ref, "controller") == true { + if str(ref, "apiVersion") != "postgresql.cnpg.io/v1" || str(ref, "kind") != "Cluster" || str(ref, "name") != clusterName || str(ref, "uid") != uid { + return false + } + count++ + } + } + return count == 1 +} + +// crashLoop recognizes the specific failed postgres container, not a sidecar. +func crashLoop(p object) bool { + for _, c := range list(p, "status", "containerStatuses") { + if str(c, "name") == "postgres" { + return value(c, "ready") == false && str(c, "state", "waiting", "reason") == "CrashLoopBackOff" + } + } + return false +} + +// auditedOperator binds fencing semantics to the inspected, fully rolled-out version. +func auditedOperator(o object) bool { + if !validMeta(o, "cnpg-system") || id(o).name != "cloudnative-pg" || num(o, "metadata", "generation") < 1 || num(o, "metadata", "generation") != num(o, "status", "observedGeneration") || num(o, "spec", "replicas") < 1 || num(o, "spec", "replicas") != num(o, "status", "availableReplicas") || num(o, "spec", "replicas") != num(o, "status", "updatedReplicas") { + return false + } + containers := list(o, "spec", "template", "spec", "containers") + return len(containers) == 1 && str(containers[0], "image") == "ghcr.io/cloudnative-pg/cloudnative-pg:1.30.1" +} + +// validate refuses incomplete, stale or unsafe observations before authorizing repair. +func validate(s inventory, o options) (plan, error) { + p := plan{} + if !auditedOperator(s.operator) { + return p, errors.New("operator version or rollout is outside the audited fencing procedure") + } + if o.clusterUID == "" || o.podUID == "" || o.now.IsZero() || !validMeta(s.cluster, namespace) || id(s.cluster).name != clusterName || id(s.cluster).uid != o.clusterUID { + return p, errors.New("cluster identity is stale or incomplete") + } + primary := str(s.cluster, "status", "currentPrimary") + if num(s.cluster, "spec", "instances") != 3 || num(s.cluster, "status", "readyInstances") != 2 || primary == "" || primary == targetName || primary != str(s.cluster, "status", "targetPrimary") || value(s.cluster, "spec", "replica") != nil || value(s.cluster, "spec", "nodeMaintenanceWindow", "inProgress") == true { + return p, errors.New("cluster is not a stable primary with one failed standby") + } + expectedFence := "" + if o.fenced { + expectedFence = `["wedding-db-1"]` + } + if !condition(s.cluster, "ContinuousArchiving") || !condition(s.cluster, "LastBackupSucceeded") || str(s.cluster, "metadata", "annotations", fenceKey) != expectedFence { + return p, errors.New("archiving, backup or fencing precondition is not clear") + } + if len(s.pods) != 3 { + return p, errors.New("expected three fully observed database pods") + } + var target object + seen := map[string]bool{} + for _, pod := range s.pods { + name := id(pod).name + if seen[name] || !validMeta(pod, namespace) || !owned(pod, o.clusterUID) || str(pod, "metadata", "labels", "cnpg.io/cluster") != clusterName || str(pod, "metadata", "labels", "cnpg.io/instanceName") != name || str(pod, "metadata", "labels", "cnpg.io/podRole") != "instance" { + return p, errors.New("pod identity, ownership or coverage is incomplete") + } + seen[name] = true + role := str(pod, "metadata", "labels", "cnpg.io/instanceRole") + if name == targetName { + if id(pod).uid != o.podUID || role != "replica" || condition(pod, "Ready") || (!o.fenced && !crashLoop(pod)) { + return p, errors.New("target is not the bound failed standby") + } + p.target = id(pod) + target = pod + } else { + if !condition(pod, "Ready") || str(pod, "status", "phase") != "Running" { + return p, errors.New("a protected database peer is not healthy") + } + if name == primary && role == "primary" { + p.primary = id(pod) + } else if name != primary && role == "replica" { + p.healthy = id(pod) + } else { + return p, errors.New("peer role does not match primary status") + } + } + } + if p.target.uid == "" || p.primary.uid == "" || p.healthy.uid == "" { + return p, errors.New("required instance identity missing") + } + var mounts []string + for _, v := range list(target, "spec", "volumes") { + if n := str(v, "persistentVolumeClaim", "claimName"); n != "" { + mounts = append(mounts, n) + } + } + // This repair has one known PG_DATA claim. A new WAL/tablespace layout needs + // its own reviewed procedure, not an inferred volume selection. + if len(mounts) != 1 || mounts[0] != targetName || len(s.claims) != 1 { + return p, errors.New("unexpected persistent volume layout") + } + claim := s.claims[0] + claimStatus := "ready" + ownership := owned(claim, o.clusterUID) + if o.detached { + claimStatus = "detached" + ownership = true + for _, ref := range list(claim, "metadata", "ownerReferences") { + if value(ref, "controller") == true { + ownership = false + } + } + } + if (!o.detached && len(list(claim, "metadata", "ownerReferences")) != 1) || !validMeta(claim, namespace) || id(claim).name != mounts[0] || !ownership || str(claim, "metadata", "labels", "cnpg.io/cluster") != clusterName || str(claim, "metadata", "labels", "cnpg.io/instanceName") != targetName || str(claim, "metadata", "annotations", "cnpg.io/pvcStatus") != claimStatus || str(claim, "status", "phase") != "Bound" { + return p, errors.New("target claim is not bound with the expected ownership") + } + volume, ok := s.volumes[str(claim, "spec", "volumeName")] + if !ok || !validMeta(volume, "") || id(volume).name != str(claim, "spec", "volumeName") || str(volume, "status", "phase") != "Bound" || str(volume, "spec", "claimRef", "uid") != id(claim).uid || str(volume, "spec", "claimRef", "name") != id(claim).name || str(volume, "spec", "claimRef", "namespace") != namespace { + return p, errors.New("volume binding is stale or incomplete") + } + policy := str(volume, "spec", "persistentVolumeReclaimPolicy") + if policy != "Retain" && policy != "Delete" { + return p, errors.New("unknown volume reclaim policy") + } + for _, j := range s.jobs { + if num(j, "status", "active") != 0 || num(j, "status", "succeeded") != 1 { + return p, errors.New("database job has not completed") + } + } + backupOK := false + for _, b := range s.backups { + if str(b, "status", "phase") == "running" || str(b, "status", "phase") == "pending" { + return p, errors.New("backup is active") + } + if !validMeta(b, namespace) || str(b, "status", "phase") != "completed" || str(b, "spec", "cluster", "name") != clusterName || str(b, "spec", "method") != "plugin" || str(b, "spec", "pluginConfiguration", "name") != "barman-cloud.cloudnative-pg.io" || str(b, "status", "pluginMetadata", "clusterUID") != o.clusterUID { + continue + } + done, err := time.Parse(time.RFC3339, str(b, "status", "stoppedAt")) + if err == nil && !done.After(o.now) && o.now.Sub(done) <= 24*time.Hour { + backupOK = true + } + } + if !backupOK { + return p, fmt.Errorf("no completed cluster-bound backup within %s", 24*time.Hour) + } + p.claims = []object{claim} + return p, nil +} + +// main reports a failed observation or repair as HOLD and exits unsuccessfully. +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "REPAIR=HOLD:", err) + os.Exit(1) + } +} + +// dispatchAllowed limits execution to the first confirmed dispatch of this main workflow. +func dispatchAllowed(env func(string) string) bool { + return env("GITHUB_WORKFLOW_REF") == "devantler-tech/platform/.github/workflows/replace-wedding-standby.yaml@refs/heads/main" && env("GITHUB_REPOSITORY") == "devantler-tech/platform" && env("GITHUB_REF") == "refs/heads/main" && env("GITHUB_EVENT_NAME") == "workflow_dispatch" && env("GITHUB_RUN_ATTEMPT") == "1" && env("WEDDING_REPAIR_CONFIRM") == "retain-volumes-replace-failed-standby" && regexp.MustCompile(`^[0-9a-f]{40}$`).MatchString(env("GITHUB_SHA")) +} + +// run selects OIDC reads by default and protected workflow execution only when bound. +func run() error { + execute := flag.Bool("execute", false, "replace the failed standby through the protected main workflow") + clusterUID := flag.String("cluster-uid", "", "expected current Cluster UID") + podUID := flag.String("pod-uid", "", "expected failed Pod UID") + flag.Parse() + if flag.NArg() != 0 { + return errors.New("unexpected argument") + } + uuid := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`) + if !uuid.MatchString(*clusterUID) || !uuid.MatchString(*podUID) { + return errors.New("explicit current cluster and pod UIDs are required") + } + contextName := "oidc@prod" + if *execute { + if !dispatchAllowed(os.Getenv) { + return errors.New("execution requires the explicitly confirmed first protected main dispatch") + } + contextName = "admin@prod" + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + defer cancel() + c := client{now: time.Now} + c.command = func(ctx context.Context, args []string, body []byte) ([]byte, error) { + cmd := exec.CommandContext(ctx, "kubectl", append([]string{"--context", contextName, "--request-timeout=30s"}, args...)...) + cmd.Stdin = bytes.NewReader(body) + return cmd.Output() + } + c.source = func(ctx context.Context) error { + return exec.CommandContext(ctx, "bash", "scripts/verify-prod-recovery-source.sh", os.Getenv("GITHUB_SHA")).Run() + } + if err := repair(ctx, c, options{clusterUID: *clusterUID, podUID: *podUID, now: time.Now()}, *execute); err != nil { + return err + } + result := "PLAN=PASS no writes" + if *execute { + result = "REPAIR=PASS readyInstances=3 retainedClaims=1 separatedSamples=2" + } + fmt.Println(result) + if *execute { + if path := os.Getenv("GITHUB_STEP_SUMMARY"); path != "" { + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0600) + if err != nil { + return err + } + _, err = fmt.Fprintln(f, result) + closeErr := f.Close() + if err != nil { + return err + } + return closeErr + } + } + return nil +} diff --git a/scripts/replace-wedding-standby/main_test.go b/scripts/replace-wedding-standby/main_test.go new file mode 100644 index 0000000000..9da62b75ee --- /dev/null +++ b/scripts/replace-wedding-standby/main_test.go @@ -0,0 +1,542 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "strings" + "testing" + "time" +) + +// TestFullyQualifiedDatabaseResources prevents selecting an unrelated Cluster API. +func TestFullyQualifiedDatabaseResources(t *testing.T) { + c := client{command: func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[1] == "cluster" || args[1] == "backups" { + t.Fatal("ambiguous CR kind could select a different API group") + } + return fakeRead(fixture(), args[1:]) + }} + if err := repair(context.Background(), c, testOptions(), false); err != nil { + t.Fatal(err) + } +} + +// TestDuplicateRecoveryObservationIsNotClean requires a distinct replacement observation. +func TestDuplicateRecoveryObservationIsNotClean(t *testing.T) { + s := fixture() + p := plan{primary: id(s.pods[1]), healthy: id(s.pods[2]), target: id(s.pods[0])} + s.pods = []object{s.pods[1], s.pods[2], s.pods[2]} + at(s.cluster, "status")["readyInstances"] = float64(3) + appendCondition(s.cluster, object{"type": "Ready", "status": "True"}) + if ok, err := (client{}).recovered(context.Background(), s, testOptions(), p, id(s.claims[0])); ok || err == nil { + t.Fatal("duplicate observations cleared recovery without a replacement") + } +} + +// TestDispatchBoundary refuses each independently invalid execution binding. +func TestDispatchBoundary(t *testing.T) { + good := map[string]string{"GITHUB_WORKFLOW_REF": "devantler-tech/platform/.github/workflows/replace-wedding-standby.yaml@refs/heads/main", "GITHUB_REPOSITORY": "devantler-tech/platform", "GITHUB_REF": "refs/heads/main", "GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_RUN_ATTEMPT": "1", "WEDDING_REPAIR_CONFIRM": "retain-volumes-replace-failed-standby", "GITHUB_SHA": strings.Repeat("a", 40)} + for key := range good { + t.Run(key, func(t *testing.T) { + for k, v := range good { + t.Setenv(k, v) + } + t.Setenv(key, "wrong") + previousFlags, previousArgs := flag.CommandLine, os.Args + t.Cleanup(func() { flag.CommandLine = previousFlags; os.Args = previousArgs }) + flag.CommandLine = flag.NewFlagSet("repair", flag.ContinueOnError) + flag.CommandLine.SetOutput(io.Discard) + os.Args = []string{"repair", "--execute", "--cluster-uid", "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", "--pod-uid", "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"} + err := run() + if err == nil || !strings.Contains(err.Error(), "execution requires") { + t.Fatalf("protected boundary returned %v", err) + } + }) + } +} + +// TestFenceMetricMustBeCompleteAndTrue rejects missing, duplicate and failed evidence. +func TestFenceMetricMustBeCompleteAndTrue(t *testing.T) { + for _, tc := range []struct { + body string + failed, ok bool + }{{"cnpg_collector_fencing_on 1\n", false, true}, {"cnpg_collector_fencing_on 1\n", true, false}, {"cnpg_collector_fencing_on 0\n", false, false}, {"cnpg_collector_fencing_on 1\ncnpg_collector_fencing_on 1\n", false, false}, {"", false, false}} { + c := client{command: func(context.Context, []string, []byte) ([]byte, error) { + if tc.failed { + return []byte(tc.body), errors.New("partial response") + } + return []byte(tc.body), nil + }} + if got := c.fenced(context.Background()); got != tc.ok { + t.Fatalf("fencing %q=%v", tc.body, got) + } + } +} + +// TestDispatchMustNameTheProtectedRecoveryWorkflow rejects a different dispatch caller. +func TestDispatchMustNameTheProtectedRecoveryWorkflow(t *testing.T) { + env := map[string]string{"GITHUB_REPOSITORY": "devantler-tech/platform", "GITHUB_REF": "refs/heads/main", "GITHUB_EVENT_NAME": "workflow_dispatch", "GITHUB_RUN_ATTEMPT": "1", "WEDDING_REPAIR_CONFIRM": "retain-volumes-replace-failed-standby", "GITHUB_SHA": strings.Repeat("a", 40), "GITHUB_WORKFLOW_REF": "devantler-tech/platform/.github/workflows/replace-wedding-standby.yaml@refs/heads/main"} + get := func(key string) string { return env[key] } + if !dispatchAllowed(get) { + t.Fatal("reviewed dispatch refused") + } + env["GITHUB_WORKFLOW_REF"] = "devantler-tech/platform/.github/workflows/other.yaml@refs/heads/main" + if dispatchAllowed(get) { + t.Fatal("another workflow received recovery authority") + } +} + +// fakeRead returns Kubernetes-shaped responses for the transaction's exact reads. +func fakeRead(s inventory, args []string) ([]byte, error) { + kind, name := strings.TrimSuffix(args[0], ".postgresql.cnpg.io"), "" + if len(args) > 1 { + name = args[1] + } + var result any + switch kind { + case "cluster": + result = s.cluster + case "deployment": + result = s.operator + case "pods": + result = object{"items": s.pods} + case "pvc": + if name == "wedding-db-4" { + claim := fixture().claims[0] + at(claim, "metadata")["name"] = "wedding-db-4" + at(claim, "metadata")["uid"] = "new-claim-uid" + at(claim, "spec")["volumeName"] = "new-pv" + result = claim + } else { + result = object{"items": s.claims} + } + case "pv": + if name == "new-pv" { + pv := fixture().volumes["old-pv"] + at(pv, "metadata")["name"] = "new-pv" + at(pv, "metadata")["uid"] = "new-pv-uid" + at(pv, "spec", "claimRef")["name"] = "wedding-db-4" + at(pv, "spec", "claimRef")["uid"] = "new-claim-uid" + result = pv + } else { + result = s.volumes[name] + } + case "backups": + result = object{"items": s.backups} + case "jobs": + jobs := s.jobs + if jobs == nil { + jobs = []object{} + } + result = object{"items": jobs} + default: + return nil, errors.New("unexpected read") + } + return json.Marshal(result) +} + +// TestReadOnlyPlanAndPartialReadFailure rejects plausible output followed by failure. +func TestReadOnlyPlanAndPartialReadFailure(t *testing.T) { + for _, fail := range []string{"", "cluster", "deployment", "pods", "pvc", "pv", "backups", "jobs"} { + t.Run("partial "+fail, func(t *testing.T) { + c := client{command: func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[0] != "get" { + t.Fatal("plan issued a write") + } + data, err := fakeRead(fixture(), args[1:]) + if len(args) > 1 && strings.TrimSuffix(args[1], ".postgresql.cnpg.io") == fail { + return data, errors.New("plausible partial response followed by failure") + } + return data, err + }} + err := repair(context.Background(), c, testOptions(), false) + if (err == nil) != (fail == "") { + t.Fatalf("error=%v on failed read %q", err, fail) + } + }) + } +} + +// TestMalformedJobListingCannotClearPlan rejects entries that cannot be examined. +func TestMalformedJobListingCannotClearPlan(t *testing.T) { + for _, entry := range []any{"not an object", nil} { + c := client{command: func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[0] != "get" { + t.Fatal("plan issued a write") + } + if args[1] == "jobs" { + return json.Marshal(object{"items": []any{entry}}) + } + return fakeRead(fixture(), args[1:]) + }} + if err := repair(context.Background(), c, testOptions(), false); err == nil { + t.Fatalf("unexamined job entry %v cleared the plan", entry) + } + } +} + +// TestExecutionRefusesFailedSourceBeforeWrites protects against stale-main execution. +func TestExecutionRefusesFailedSourceBeforeWrites(t *testing.T) { + writes := 0 + c := client{command: func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[0] != "get" { + writes++ + return nil, nil + } + return fakeRead(fixture(), args[1:]) + }, source: func(context.Context) error { return errors.New("source stale") }} + if err := repair(context.Background(), c, testOptions(), true); err == nil { + t.Fatal("stale source accepted") + } + if writes != 0 { + t.Fatal("write before source proof") + } +} + +// TestRepairStopsOnIntermediateDrift prevents later mutations after a safety gate changes. +func TestRepairStopsOnIntermediateDrift(t *testing.T) { + for _, tc := range []struct { + name string + after int + change func(inventory) + }{ + {"primary changed before fence", 1, func(s inventory) { at(s.cluster, "status")["currentPrimary"] = "wedding-db-3" }}, + {"healthy peer replaced before detach", 2, func(s inventory) { at(s.pods[2], "metadata")["uid"] = "changed" }}, + {"volume binding changed before detach", 2, func(s inventory) { at(s.volumes["old-pv"], "spec", "claimRef")["uid"] = "changed" }}, + {"target ready before detach", 2, func(s inventory) { + at(s.pods[0], "status")["conditions"] = []any{object{"type": "Ready", "status": "True"}} + }}, + {"pod replaced before delete", 3, func(s inventory) { at(s.pods[0], "metadata")["uid"] = "changed" }}, + {"primary changing before delete", 3, func(s inventory) { at(s.cluster, "status")["targetPrimary"] = "wedding-db-1" }}, + } { + t.Run(tc.name, func(t *testing.T) { + s := fixture() + writes := 0 + c := client{source: func(context.Context) error { return nil }, wait: func(context.Context) error { return errors.New("unexpected wait") }} + c.command = func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[0] == "get" { + if args[1] == "--raw" { + return []byte("cnpg_collector_fencing_on 1\n"), nil + } + return fakeRead(s, args[1:]) + } + writes++ + if args[0] != "patch" { + t.Fatal("unsafe pod deletion after drift") + } + switch args[1] { + case "pv": + at(s.volumes["old-pv"], "spec")["persistentVolumeReclaimPolicy"] = "Retain" + case "cluster.postgresql.cnpg.io": + at(s.cluster, "metadata", "annotations")[fenceKey] = `["wedding-db-1"]` + case "pvc": + at(s.claims[0], "metadata")["ownerReferences"] = []any{} + at(s.claims[0], "metadata", "annotations")["cnpg.io/pvcStatus"] = "detached" + default: + t.Fatal("unexpected mutation") + } + if writes == tc.after { + tc.change(s) + } + return []byte(`{}`), nil + } + if err := repair(context.Background(), c, testOptions(), true); err == nil { + t.Fatal("drift cleared repair") + } + if writes != tc.after { + t.Fatalf("made %d writes, expected stop at %d", writes, tc.after) + } + }) + } +} + +// TestFailedConditionalWriteDoesNotRetry keeps a rejected mutation in HOLD. +func TestFailedConditionalWriteDoesNotRetry(t *testing.T) { + writes := 0 + c := client{source: func(context.Context) error { return nil }} + c.command = func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[0] == "get" { + return fakeRead(fixture(), args[1:]) + } + writes++ + return []byte(`{"status":"Success"}`), errors.New("CAS rejected after output") + } + if err := repair(context.Background(), c, testOptions(), true); err == nil { + t.Fatal("failed write treated as clean") + } + if writes != 1 { + t.Fatal("write retried or cleanup attempted") + } +} + +// TestUnacknowledgedFenceCannotDetachStorage preserves ownership until fencing is proven. +func TestUnacknowledgedFenceCannotDetachStorage(t *testing.T) { + s := fixture() + writes := 0 + c := client{source: func(context.Context) error { return nil }, wait: func(context.Context) error { return context.DeadlineExceeded }} + c.command = func(_ context.Context, args []string, _ []byte) ([]byte, error) { + if args[0] == "get" { + if args[1] == "--raw" { + return []byte("cnpg_collector_fencing_on 0\n"), nil + } + return fakeRead(s, args[1:]) + } + writes++ + switch args[1] { + case "pv": + at(s.volumes["old-pv"], "spec")["persistentVolumeReclaimPolicy"] = "Retain" + case "cluster.postgresql.cnpg.io": + at(s.cluster, "metadata", "annotations")[fenceKey] = `["wedding-db-1"]` + default: + t.Fatal("mutated storage without acknowledged fencing") + } + return []byte(`{}`), nil + } + if err := repair(context.Background(), c, testOptions(), true); err == nil { + t.Fatal("unfenced instance cleared repair") + } + if writes != 2 || !owned(s.claims[0], "cluster-uid") { + t.Fatal("claim was changed without fencing") + } +} + +// TestWriteSequencePreservesDataAndPinsDeletes checks proof/read/write ordering and retention. +func TestWriteSequencePreservesDataAndPinsDeletes(t *testing.T) { + s := fixture() + var writes []string + proofs := 0 + c := client{source: func(context.Context) error { + // Controllers can publish harmless status changes during the forge read. + // Every write must observe the versions AFTER that read, not before it. + proofs++ + for _, o := range append([]object{s.cluster, s.operator, s.claims[0], s.volumes["old-pv"]}, s.pods...) { + at(o, "metadata")["resourceVersion"] = fmt.Sprint(proofs) + } + return nil + }, wait: func(context.Context) error { return nil }} + c.command = func(_ context.Context, args []string, data []byte) ([]byte, error) { + if args[0] == "get" { + if args[1] == "--raw" { + return []byte("# gauge\ncnpg_collector_fencing_on 1\n"), nil + } + return fakeRead(s, args[1:]) + } + writes = append(writes, strings.Join(args[:2], " ")) + if args[0] == "delete" { + if args[1] != "--raw" || !strings.Contains(strings.Join(args, " "), "/pods/wedding-db-1") { + t.Fatal("deletion escaped the single pod") + } + var body object + if err := json.Unmarshal(data, &body); err != nil { + t.Fatal(err) + } + if str(body, "preconditions", "uid") != "target-uid" || str(body, "preconditions", "resourceVersion") != str(s.pods[0], "metadata", "resourceVersion") { + t.Fatal("delete not bound to exact pod") + } + s.pods = s.pods[1:] + newPod := fixture().pods[2] + at(newPod, "metadata")["name"] = "wedding-db-4" + at(newPod, "metadata")["uid"] = "new-uid" + at(newPod, "metadata", "labels")["cnpg.io/instanceName"] = "wedding-db-4" + at(newPod, "spec")["volumes"] = []any{object{"persistentVolumeClaim": object{"claimName": "wedding-db-4"}}} + s.pods = append(s.pods, newPod) + at(s.cluster, "status")["readyInstances"] = float64(3) + appendCondition(s.cluster, object{"type": "Ready", "status": "True"}) + return []byte(`{}`), nil + } + if args[0] != "patch" { + t.Fatalf("unexpected write %v", args) + } + var ops []object + if err := json.Unmarshal(data, &ops); err != nil { + t.Fatal(err) + } + if len(ops) < 3 || str(ops[0], "op") != "test" || str(ops[0], "path") != "/metadata/uid" || str(ops[1], "op") != "test" || str(ops[1], "path") != "/metadata/resourceVersion" { + t.Fatal("patch omitted identity preconditions") + } + switch strings.TrimSuffix(args[1], ".postgresql.cnpg.io") { + case "pv": + if value(ops[1], "value") != str(s.volumes["old-pv"], "metadata", "resourceVersion") { + t.Fatal("PV snapshot preceded the source proof") + } + at(s.volumes["old-pv"], "spec")["persistentVolumeReclaimPolicy"] = "Retain" + case "cluster": + if value(ops[1], "value") != str(s.cluster, "metadata", "resourceVersion") { + t.Fatal("Cluster snapshot preceded the source proof") + } + last := ops[len(ops)-1] + if str(last, "op") == "remove" { + delete(at(s.cluster, "metadata", "annotations"), fenceKey) + } else { + at(s.cluster, "metadata", "annotations")[fenceKey] = `["wedding-db-1"]` + } + case "pvc": + if value(ops[1], "value") != str(s.claims[0], "metadata", "resourceVersion") { + t.Fatal("PVC snapshot preceded the source proof") + } + at(s.claims[0], "metadata")["ownerReferences"] = []any{} + at(s.claims[0], "metadata", "annotations")["cnpg.io/pvcStatus"] = "detached" + default: + t.Fatal("write outside authorized resources") + } + return []byte(`{}`), nil + } + if err := repair(context.Background(), c, testOptions(), true); err != nil { + t.Fatal(err) + } + want := "patch pv,patch cluster.postgresql.cnpg.io,patch pvc,delete --raw,patch cluster.postgresql.cnpg.io" + if strings.Join(writes, ",") != want { + t.Fatalf("write ordering %v; want %s", writes, want) + } + if proofs < len(writes) { + t.Fatal("a write reused an old source proof") + } + if id(s.claims[0]).uid != "claim-uid" || str(s.volumes["old-pv"], "spec", "claimRef", "uid") != "claim-uid" { + t.Fatal("old data binding lost") + } +} + +var testNow = time.Date(2026, 10, 9, 1, 0, 0, 0, time.UTC) + +// fixture models the one failed standby and its protected peers and bound storage. +func fixture() inventory { + owner := []any{object{"apiVersion": "postgresql.cnpg.io/v1", "kind": "Cluster", "name": "wedding-db", "uid": "cluster-uid", "controller": true}} + meta := func(name, uid string) object { + return object{"name": name, "namespace": "wedding-app", "uid": uid, "resourceVersion": "10", "ownerReferences": owner} + } + pod := func(name, uid, role, ready string) object { + m := meta(name, uid) + m["labels"] = object{"cnpg.io/cluster": "wedding-db", "cnpg.io/instanceName": name, "cnpg.io/instanceRole": role, "cnpg.io/podRole": "instance"} + status := object{"phase": "Running", "conditions": []any{object{"type": "Ready", "status": ready}}, "containerStatuses": []any{object{"name": "postgres", "ready": ready == "True", "state": object{"waiting": object{"reason": "CrashLoopBackOff"}}}}} + return object{"metadata": m, "status": status, "spec": object{"volumes": []any{object{"persistentVolumeClaim": object{"claimName": name}}}}} + } + c := object{"apiVersion": "postgresql.cnpg.io/v1", "kind": "Cluster", "metadata": object{"name": "wedding-db", "namespace": "wedding-app", "uid": "cluster-uid", "resourceVersion": "10", "annotations": object{}}, "spec": object{"instances": float64(3)}, "status": object{"readyInstances": float64(2), "currentPrimary": "wedding-db-2", "targetPrimary": "wedding-db-2", "conditions": []any{object{"type": "ContinuousArchiving", "status": "True"}, object{"type": "LastBackupSucceeded", "status": "True"}}}} + claim := object{"metadata": meta("wedding-db-1", "claim-uid"), "spec": object{"volumeName": "old-pv"}, "status": object{"phase": "Bound"}} + at(claim, "metadata")["labels"] = object{"cnpg.io/cluster": "wedding-db", "cnpg.io/instanceName": "wedding-db-1"} + at(claim, "metadata")["annotations"] = object{"cnpg.io/pvcStatus": "ready"} + volume := object{"metadata": object{"name": "old-pv", "uid": "pv-uid", "resourceVersion": "10"}, "spec": object{"claimRef": object{"name": "wedding-db-1", "namespace": "wedding-app", "uid": "claim-uid"}, "persistentVolumeReclaimPolicy": "Delete"}, "status": object{"phase": "Bound"}} + backup := object{"metadata": meta("daily-backup", "backup-uid"), "spec": object{"cluster": object{"name": "wedding-db"}, "method": "plugin", "pluginConfiguration": object{"name": "barman-cloud.cloudnative-pg.io"}}, "status": object{"phase": "completed", "stoppedAt": testNow.Add(-time.Hour).Format(time.RFC3339), "pluginMetadata": object{"clusterUID": "cluster-uid"}}} + operator := object{"metadata": object{"name": "cloudnative-pg", "namespace": "cnpg-system", "uid": "operator-uid", "resourceVersion": "10", "generation": float64(1)}, "spec": object{"replicas": float64(2), "template": object{"spec": object{"containers": []any{object{"image": "ghcr.io/cloudnative-pg/cloudnative-pg:1.30.1"}}}}}, "status": object{"observedGeneration": float64(1), "availableReplicas": float64(2), "updatedReplicas": float64(2)}} + return inventory{cluster: c, operator: operator, pods: []object{pod("wedding-db-1", "target-uid", "replica", "False"), pod("wedding-db-2", "primary-uid", "primary", "True"), pod("wedding-db-3", "healthy-uid", "replica", "True")}, claims: []object{claim}, volumes: map[string]object{"old-pv": volume}, backups: []object{backup}} +} + +// at requires the fixture's expected map shape and fails loudly on fixture mistakes. +func at(o object, path ...string) object { + for _, k := range path { + next, ok := o[k].(object) + if !ok { + panic("fixture map is missing at " + k) + } + o = next + } + return o +} + +// appendCondition extends decoded conditions without unchecked type assertions. +func appendCondition(o object, c object) { + items := []any{} + for _, existing := range list(o, "status", "conditions") { + items = append(items, existing) + } + at(o, "status")["conditions"] = append(items, c) +} + +// testOptions binds deterministic identities and freshness for the fixture. +func testOptions() options { + return options{clusterUID: "cluster-uid", podUID: "target-uid", now: testNow} +} + +// TestValidateSafeRepair proves complete safe evidence produces a bound plan. +func TestValidateSafeRepair(t *testing.T) { + p, err := validate(fixture(), testOptions()) + if err != nil { + t.Fatal(err) + } + if p.primary.uid != "primary-uid" || p.healthy.uid != "healthy-uid" || p.target.uid != "target-uid" || len(p.claims) != 1 { + t.Fatalf("unbound plan: %+v", p) + } +} + +// TestRefuseUnsafeRepair checks invalid identities, storage, backups and roles. +func TestRefuseUnsafeRepair(t *testing.T) { + cases := []struct { + name string + change func(*inventory, *options) + }{ + {"unaudited operator", func(s *inventory, _ *options) { + at(s.operator, "spec", "template", "spec")["containers"] = []any{object{"image": "other:latest"}} + }}, + {"operator rolling", func(s *inventory, _ *options) { at(s.operator, "status")["updatedReplicas"] = float64(1) }}, + {"stale cluster", func(_ *inventory, o *options) { o.clusterUID = "previous" }}, + {"stale pod", func(_ *inventory, o *options) { o.podUID = "previous" }}, + {"target primary", func(s *inventory, _ *options) { at(s.cluster, "status")["currentPrimary"] = "wedding-db-1" }}, + {"pending switchover", func(s *inventory, _ *options) { at(s.cluster, "status")["targetPrimary"] = "wedding-db-3" }}, + {"target primary label", func(s *inventory, _ *options) { + at(s.pods[0], "metadata", "labels")["cnpg.io/instanceRole"] = "primary" + }}, + {"target ready", func(s *inventory, _ *options) { + at(s.pods[0], "status")["conditions"] = []any{object{"type": "Ready", "status": "True"}} + }}, + {"unknown failure", func(s *inventory, _ *options) { at(s.pods[0], "status")["containerStatuses"] = []any{} }}, + {"unhealthy peer", func(s *inventory, _ *options) { + at(s.pods[2], "status")["conditions"] = []any{object{"type": "Ready", "status": "False"}} + }}, + {"missing primary", func(s *inventory, _ *options) { s.pods = s.pods[:1] }}, + {"foreign pod owner", func(s *inventory, _ *options) { at(s.pods[0], "metadata")["ownerReferences"] = []any{} }}, + {"deleting pod", func(s *inventory, _ *options) { at(s.pods[0], "metadata")["deletionTimestamp"] = "now" }}, + {"missing pod version", func(s *inventory, _ *options) { delete(at(s.pods[0], "metadata"), "resourceVersion") }}, + {"missing claim", func(s *inventory, _ *options) { s.claims = nil }}, + {"extra claim", func(s *inventory, _ *options) { s.claims = append(s.claims, s.claims[0]) }}, + {"foreign claim owner", func(s *inventory, _ *options) { at(s.claims[0], "metadata")["ownerReferences"] = []any{} }}, + {"deleting claim", func(s *inventory, _ *options) { at(s.claims[0], "metadata")["deletionTimestamp"] = "now" }}, + {"unbound claim", func(s *inventory, _ *options) { at(s.claims[0], "status")["phase"] = "Pending" }}, + {"missing volume", func(s *inventory, _ *options) { s.volumes = nil }}, + {"foreign volume binding", func(s *inventory, _ *options) { at(s.volumes["old-pv"], "spec", "claimRef")["uid"] = "foreign" }}, + {"deleting volume", func(s *inventory, _ *options) { at(s.volumes["old-pv"], "metadata")["deletionTimestamp"] = "now" }}, + {"existing fence", func(s *inventory, _ *options) { + at(s.cluster, "metadata", "annotations")["cnpg.io/fencedInstances"] = `["other"]` + }}, + {"active job", func(s *inventory, _ *options) { s.jobs = []object{{"status": object{"active": float64(1)}}} }}, + {"no backup", func(s *inventory, _ *options) { s.backups = nil }}, + {"stale backup", func(s *inventory, _ *options) { + at(s.backups[0], "status")["stoppedAt"] = testNow.Add(-25 * time.Hour).Format(time.RFC3339) + }}, + {"future backup", func(s *inventory, _ *options) { + at(s.backups[0], "status")["stoppedAt"] = testNow.Add(time.Hour).Format(time.RFC3339) + }}, + {"foreign backup", func(s *inventory, _ *options) { at(s.backups[0], "status", "pluginMetadata")["clusterUID"] = "foreign" }}, + {"failed backup", func(s *inventory, _ *options) { at(s.backups[0], "status")["phase"] = "failed" }}, + {"archiving unhealthy", func(s *inventory, _ *options) { at(s.cluster, "status")["conditions"] = []any{} }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s, o := fixture(), testOptions() + tc.change(&s, &o) + if _, err := validate(s, o); err == nil { + t.Fatal("unsafe repair accepted") + } + }) + } +} + +// TestFixtureDecodesLikeKubernetes exercises kubectl's JSON numeric and list forms. +func TestFixtureDecodesLikeKubernetes(t *testing.T) { + s := fixture() + b, err := json.Marshal(s.cluster) + if err != nil { + t.Fatal(err) + } + var decoded object + if err := json.Unmarshal(b, &decoded); err != nil { + t.Fatal(err) + } + s.cluster = decoded + if _, err := validate(s, testOptions()); err != nil { + t.Fatal(err) + } +}