feat(canopy): materialise PostgresPhysicalReplica CRs from worklist (#82)

Author: passcod

README.md

@@ -83,7 +83,8 @@ Defines a continuously-refreshed replica of a PostgreSQL database restored from
 
 | Field | Type | Required | Default | Description |
 |-------|------|----------|---------|-------------|
-| `kopiaSecretRef` | `SecretReference` | Yes | — | Reference to a Secret containing Kopia repository credentials (`bucket`, `region`, `repositoryPassword`, `accessKeyId`, `secretAccessKey`). |
+| `kopiaSecretRef` | `SecretReference` | One of | — | Reference to a Secret containing Kopia repository credentials (`bucket`, `region`, `repositoryPassword`, `accessKeyId`, `secretAccessKey`). Mutually exclusive with `canopySource`. |
+| `canopySource` | `CanopySource` | One of | — | Route kopia through the canopy-mediated proxy sidecar instead of a static Secret. `{ group, type }`. Managed by the canopy worklist syncer; humans usually don't hand-author these. Mutually exclusive with `kopiaSecretRef`. |
 | `snapshotFilter` | `SnapshotFilter` | No | — | Filter criteria to select which Kopia snapshot to restore. |
 | `schedule` | `string` | Yes | — | Cron expression controlling how often new restores are triggered. |
 | `scheduleJitter` | `string` | No | `"10m"` | Random jitter added to scheduled restores (friendly duration, e.g. `"5m"`, `"1h"`). |
@@ -175,6 +176,7 @@ Additional fields for `target: graphQL`:
 | `lastRestoreCompletedAt` | `Time` | When the last restore completed. |
 | `nextScheduledRestore` | `Time` | When the next scheduled restore will occur. |
 | `latestAvailableSnapshot` | `string` | Snapshot ID of the latest available snapshot matching the filter. |
+| `canopyDesiredSnapshotId` | `string` | For canopy-sourced replicas: the snapshot the canopy worklist syncer wants restored. The reconciler triggers a new restore when this differs from the current one. |
 | `connectionInfo` | `ConnectionInfo` | Connection details (host, port, database, username, password secret). |
 | `queuePosition` | `uint32` | Position in the global restore queue. |
 | `notifications` | `[]NotificationStatus` | Status of each configured notification target. |

_typos.toml

@@ -1,3 +1,8 @@
 [default.extend-words]
 # BRIN: postgres "Block Range Index" type.
 BRIN = "BRIN"
+
+[files]
+# Generated from k8s-openapi schemas; upstream docstrings use
+# "ANDed" / "ORed" style jargon typos flags as misspellings.
+extend-exclude = ["crds.yaml"]

crds.yaml

@@ -222,9 +222,6 @@ spec:
                       # operators can pin to a specific tag.
                       # - name: CANOPY_PROXY_IMAGE
                       #   value: ghcr.io/beyondessential/pgro-canopy-proxy:latest
-                      # Default PVC size for canopy-provisioned pgdata.
-                      # - name: CANOPY_PGDATA_PVC_SIZE
-                      #   value: "20Gi"
                   ports:
                       - name: http
                         containerPort: 8080

operator.yaml

@@ -22,10 +22,10 @@ use tracing::{debug, info, warn};
 use postgres_restore_operator::{
 	canopy::{self, DEFAULT_SOCKS5_PROXY},
 	context::{
-		Context, DEFAULT_CANOPY_PGDATA_PVC_SIZE, DEFAULT_CANOPY_PROXY_IMAGE,
-		DEFAULT_DEPLOYMENT_READY_TIMEOUT_SECS, DEFAULT_KOPIA_IMAGE,
+		Context, DEFAULT_CANOPY_PROXY_IMAGE, DEFAULT_DEPLOYMENT_READY_TIMEOUT_SECS,
+		DEFAULT_KOPIA_IMAGE,
 	},
-	controllers,
+	controllers::{self, canopy::intent::SUPPORTED as PGRO_SUPPORTED_INTENTS},
 	types::{PostgresPhysicalReplica, PostgresPhysicalRestore},
 };
 
@@ -44,12 +44,6 @@ const DEFAULT_BROKER_ADDR: &str = "[::]:9091";
 const DEFAULT_CANOPY_RECONCILE_INTERVAL_SECS: u64 = 30;
 const CONFIGMAP_NAME: &str = "postgres-restore-operator-config";
 
-/// Intent set pgro registers with canopy on startup; only worklist entries
-/// with a matching intent will be dispatched. `disaster-recovery` is not
-/// yet supported — the code has no rehearsal lifecycle beyond "make it
-/// writable", which is not what DR actually needs.
-const PGRO_SUPPORTED_INTENTS: &[&str] = &["verify", "analytics"];
-
 /// Annotate the operator's own pod with the running version.
 async fn annotate_own_pod(client: &Client, namespace: &str) {
 	let pod_name = match std::env::var("HOSTNAME") {
@@ -220,8 +214,6 @@ async fn main() -> anyhow::Result<()> {
 		});
 	ctx.canopy_proxy_image = std::env::var("CANOPY_PROXY_IMAGE")
 		.unwrap_or_else(|_| DEFAULT_CANOPY_PROXY_IMAGE.to_string());
-	ctx.canopy_pgdata_pvc_size = std::env::var("CANOPY_PGDATA_PVC_SIZE")
-		.unwrap_or_else(|_| DEFAULT_CANOPY_PGDATA_PVC_SIZE.to_string());
 	ctx.canopy_broker_base_url = if let Ok(url) = std::env::var("CANOPY_BROKER_BASE_URL") {
 		url
 	} else if let Ok(svc) = std::env::var("OPERATOR_SERVICE_NAME") {
@@ -564,8 +556,8 @@ async fn post_cache_pressure(
 }
 
 /// Accept the canopy-proxy sidecar's final TrafficStats POST on shutdown.
-/// The body is opaque JSON — the reporter deserializes it when building
-/// the RestoreVerification.
+/// The body is opaque JSON — the canopy notification target deserializes
+/// it when building the RestoreVerification.
 async fn post_canopy_stats(
 	State(state): State<ServerState>,
 	Path((namespace, job)): Path<(String, String)>,

src/bin/operator.rs

@@ -17,7 +17,6 @@ pub const DEFAULT_DEPLOYMENT_READY_TIMEOUT_SECS: u64 = 30 * 60;
 /// sidecar to a different tag from the operator itself.
 pub const DEFAULT_CANOPY_PROXY_IMAGE: &str =
 	"ghcr.io/beyondessential/postgres-restore-operator:latest";
-pub const DEFAULT_CANOPY_PGDATA_PVC_SIZE: &str = "20Gi";
 
 pub struct Context {
 	pub client: Client,
@@ -39,17 +38,14 @@ pub struct Context {
 	/// Image reference for the pgro-canopy-proxy sidecar container. Set
 	/// by the operator startup from `CANOPY_PROXY_IMAGE`.
 	pub canopy_proxy_image: String,
-	/// Default pgdata PVC size used when the operator provisions a
-	/// canopy-backed replica (per intent).
-	pub canopy_pgdata_pvc_size: String,
 	/// In-memory store for snapshot-list results POSTed by jobs.
 	pub snapshot_results: Arc<CallbackStore>,
 	/// In-memory store for schema migration results POSTed by jobs.
 	pub schema_migration_results: Arc<CallbackStore>,
 	/// In-memory store for canopy-proxy sidecar TrafficStats keyed by
 	/// `{namespace}/{job}`. Written on sidecar exit via the operator's
-	/// `/api/v1/canopy-stats/...` callback; read by the reporter when
-	/// building `RestoreVerification.s3_*_bytes`.
+	/// `/api/v1/canopy-stats/...` callback; read by the canopy
+	/// notification target when building `RestoreVerification.s3_*_bytes`.
 	pub canopy_stats: Arc<CallbackStore>,
 	/// Base URL the operator is reachable at from within the cluster,
 	/// e.g. `http://postgres-restore-operator.pgro-system.svc:8080`.
@@ -89,7 +85,6 @@ impl Context {
 			canopy: None,
 			canopy_broker_base_url: String::new(),
 			canopy_proxy_image: DEFAULT_CANOPY_PROXY_IMAGE.to_string(),
-			canopy_pgdata_pvc_size: DEFAULT_CANOPY_PGDATA_PVC_SIZE.to_string(),
 			snapshot_results: Arc::new(CallbackStore::default()),
 			schema_migration_results: Arc::new(CallbackStore::default()),
 			canopy_stats: Arc::new(CallbackStore::default()),