From ae60403756be5de90ce62aff8103f2ee4d931f28 Mon Sep 17 00:00:00 2001 From: Leo Belanger Date: Mon, 19 Oct 2026 09:31:35 -0400 Subject: [PATCH 1/3] feat(postgresql): request extensions on CNPG clusters PostgreSQLConfig gains `extensions`. When non-empty, the CNPG interpret applies a postgresql.cnpg.io/v1 Database for the app database and waits until CNPG reports it applied, so superuser-only extensions like `vector` work, on existing clusters too. --- harmony/src/modules/postgresql/capability.rs | 4 + harmony/src/modules/postgresql/cnpg/crd.rs | 38 ++++++ harmony/src/modules/postgresql/failover.rs | 2 + harmony/src/modules/postgresql/score_k8s.rs | 125 ++++++++++++++++++- 4 files changed, 166 insertions(+), 3 deletions(-) diff --git a/harmony/src/modules/postgresql/capability.rs b/harmony/src/modules/postgresql/capability.rs index b8dbb610..48190323 100644 --- a/harmony/src/modules/postgresql/capability.rs +++ b/harmony/src/modules/postgresql/capability.rs @@ -72,6 +72,9 @@ pub struct PostgreSQLConfig { /// exist before returning. This ensures that callers like `get_endpoint()` /// won't fail with "service not found" immediately after deployment. pub wait_for_ready: bool, + /// Created in the `app` database by the CNPG superuser, so untrusted + /// extensions such as `vector` work. + pub extensions: Vec, } impl PostgreSQLConfig { @@ -92,6 +95,7 @@ impl Default for PostgreSQLConfig { namespace: "harmony".to_string(), wait_for_ready: true, version: None, + extensions: Vec::new(), } } } diff --git a/harmony/src/modules/postgresql/cnpg/crd.rs b/harmony/src/modules/postgresql/cnpg/crd.rs index 9144a0d4..8f74eff2 100644 --- a/harmony/src/modules/postgresql/cnpg/crd.rs +++ b/harmony/src/modules/postgresql/cnpg/crd.rs @@ -55,6 +55,44 @@ impl Default for ClusterSpec { } } +#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, Default)] +#[kube( + group = "postgresql.cnpg.io", + version = "v1", + kind = "Database", + plural = "databases", + namespaced = true, + status = "DatabaseStatus", + derive = "Default", + schema = "disabled" +)] +#[serde(rename_all = "camelCase")] +pub struct DatabaseSpec { + pub name: String, + pub owner: String, + pub cluster: ClusterReference, + pub extensions: Vec, +} + +#[derive(Deserialize, Serialize, Clone, Debug, Default)] +pub struct ClusterReference { + pub name: String, +} + +#[derive(Deserialize, Serialize, Clone, Debug)] +pub struct DatabaseExtension { + pub name: String, + pub ensure: String, +} + +#[derive(Deserialize, Serialize, Clone, Debug, Default)] +#[serde(rename_all = "camelCase")] +pub struct DatabaseStatus { + pub applied: Option, + pub message: Option, + pub observed_generation: Option, +} + #[derive(Deserialize, Serialize, Clone, Debug, Default)] #[serde(rename_all = "camelCase")] pub struct Storage { diff --git a/harmony/src/modules/postgresql/failover.rs b/harmony/src/modules/postgresql/failover.rs index 40ab4881..5cd1dc91 100644 --- a/harmony/src/modules/postgresql/failover.rs +++ b/harmony/src/modules/postgresql/failover.rs @@ -31,6 +31,7 @@ impl PostgreSQL for FailoverTopology { namespace: config.namespace.clone(), wait_for_ready: config.wait_for_ready, version: config.version.clone(), + extensions: config.extensions.clone(), }; info!( @@ -148,6 +149,7 @@ impl PostgreSQL for FailoverTopology { namespace: config.namespace.clone(), wait_for_ready: config.wait_for_ready, version: config.version.clone(), + extensions: config.extensions.clone(), }; info!( diff --git a/harmony/src/modules/postgresql/score_k8s.rs b/harmony/src/modules/postgresql/score_k8s.rs index 9c0d43c1..964e1ffd 100644 --- a/harmony/src/modules/postgresql/score_k8s.rs +++ b/harmony/src/modules/postgresql/score_k8s.rs @@ -4,8 +4,8 @@ use crate::inventory::Inventory; use crate::modules::k8s::resource::K8sResourceScore; use crate::modules::postgresql::capability::{PostgreSQLConfig, PostgreSQLRootAccountRef}; use crate::modules::postgresql::cnpg::{ - Bootstrap, Cluster, ClusterSpec, ExternalCluster, Initdb, PgBaseBackup, ReplicaSpec, - SecretKeySelector, Storage, + Bootstrap, Cluster, ClusterReference, ClusterSpec, Database, DatabaseExtension, DatabaseSpec, + ExternalCluster, Initdb, PgBaseBackup, ReplicaSpec, SecretKeySelector, Storage, }; use crate::modules::postgresql::operator::{ CloudNativePgOperatorHelmScore, CloudNativePgOperatorScore, @@ -45,7 +45,7 @@ fn is_forbidden(e: &E) -> bool { /// ``` /// /// # Limitations (Happy Path) -/// - No backups, monitoring, extensions configured. +/// - No backups or monitoring configured. #[derive(Debug, Clone, Serialize, Default)] pub struct K8sPostgreSQLScore { pub config: PostgreSQLConfig, @@ -281,6 +281,86 @@ impl K8sPostgreSQLInterpret { tokio::time::sleep(std::time::Duration::from_secs(2)).await; } } + + // Initdb's postInitApplicationSQL only runs at cluster creation; a Database + // resource also reconciles extensions on existing clusters. + fn app_database(&self) -> Database { + Database { + metadata: ObjectMeta { + name: Some(format!("{}-app", self.config.cluster_name)), + namespace: Some(self.config.namespace.clone()), + ..ObjectMeta::default() + }, + spec: DatabaseSpec { + name: "app".to_string(), + owner: "app".to_string(), + cluster: ClusterReference { + name: self.config.cluster_name.clone(), + }, + extensions: self + .config + .extensions + .iter() + .map(|name| DatabaseExtension { + name: name.clone(), + ensure: "present".to_string(), + }) + .collect(), + }, + status: None, + } + } + + async fn wait_for_database( + &self, + topology: &T, + name: &str, + ) -> Result<(), InterpretError> { + let k8s_client = topology + .k8s_client() + .await + .map_err(|e| InterpretError::new(format!("Failed to get k8s client: {}", e)))?; + + let timeout = std::time::Duration::from_secs(120); + let start = std::time::Instant::now(); + let mut last_message = None; + + loop { + match k8s_client + .get_resource::(name, Some(&self.config.namespace)) + .await + { + // A status from an earlier generation says nothing about the extensions just applied. + Ok(Some(db)) => match db.status { + Some(status) if status.observed_generation == db.metadata.generation => { + if status.applied == Some(true) { + info!("Extensions {:?} applied", self.config.extensions); + return Ok(()); + } + last_message = status.message; + } + _ => debug!("CNPG Database '{name}' not reconciled yet"), + }, + Ok(None) => debug!("CNPG Database '{name}' not found yet"), + Err(e) => debug!("Error checking CNPG Database '{name}': {e}"), + } + + if start.elapsed() > timeout { + return Err(InterpretError::new(format!( + "CNPG Database '{}' in '{}' did not apply extensions {:?} within {}s: {}", + name, + self.config.namespace, + self.config.extensions, + timeout.as_secs(), + last_message + .as_deref() + .unwrap_or("no status reported by CNPG") + ))); + } + + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + } + } } #[async_trait] @@ -330,6 +410,18 @@ impl Interpret for K8sPostgr self.wait_for_primary(topology).await?; } + if !self.config.extensions.is_empty() { + let database = self.app_database(); + let name = database.metadata.name.clone().unwrap_or_default(); + K8sResourceScore::single(database, Some(self.config.namespace.clone())) + .create_interpret() + .execute(inventory, topology) + .await?; + if self.config.wait_for_ready { + self.wait_for_database(topology, &name).await?; + } + } + Ok(outcome) } super::capability::PostgreSQLClusterRole::Replica(replica_config) => { @@ -498,6 +590,33 @@ mod tests { ); } + #[test] + fn app_database_requests_extensions_on_the_app_database() { + let interpret = K8sPostgreSQLInterpret { + config: PostgreSQLConfig { + cluster_name: "iaai-pg".to_string(), + namespace: "iaai".to_string(), + extensions: vec!["vector".to_string()], + ..Default::default() + }, + }; + + assert_eq!( + serde_json::to_value(interpret.app_database()).unwrap(), + serde_json::json!({ + "apiVersion": "postgresql.cnpg.io/v1", + "kind": "Database", + "metadata": { "name": "iaai-pg-app", "namespace": "iaai" }, + "spec": { + "name": "app", + "owner": "app", + "cluster": { "name": "iaai-pg" }, + "extensions": [{ "name": "vector", "ensure": "present" }] + } + }) + ); + } + #[test] fn cnpg_image_name_maps_tags_and_full_refs() { assert_eq!(cnpg_image_name(None), None); -- 2.39.5 From 8d3b63b7b07fbcfa184d97ebc419dc02483293cc Mon Sep 17 00:00:00 2001 From: Leo Belanger Date: Mon, 19 Oct 2026 09:39:33 -0400 Subject: [PATCH 2/3] chore(postgresql): drop redundant comments --- harmony/src/modules/postgresql/capability.rs | 2 -- harmony/src/modules/postgresql/score_k8s.rs | 2 -- 2 files changed, 4 deletions(-) diff --git a/harmony/src/modules/postgresql/capability.rs b/harmony/src/modules/postgresql/capability.rs index 48190323..d7bc2270 100644 --- a/harmony/src/modules/postgresql/capability.rs +++ b/harmony/src/modules/postgresql/capability.rs @@ -72,8 +72,6 @@ pub struct PostgreSQLConfig { /// exist before returning. This ensures that callers like `get_endpoint()` /// won't fail with "service not found" immediately after deployment. pub wait_for_ready: bool, - /// Created in the `app` database by the CNPG superuser, so untrusted - /// extensions such as `vector` work. pub extensions: Vec, } diff --git a/harmony/src/modules/postgresql/score_k8s.rs b/harmony/src/modules/postgresql/score_k8s.rs index 964e1ffd..5e231d4d 100644 --- a/harmony/src/modules/postgresql/score_k8s.rs +++ b/harmony/src/modules/postgresql/score_k8s.rs @@ -282,8 +282,6 @@ impl K8sPostgreSQLInterpret { } } - // Initdb's postInitApplicationSQL only runs at cluster creation; a Database - // resource also reconciles extensions on existing clusters. fn app_database(&self) -> Database { Database { metadata: ObjectMeta { -- 2.39.5 From 8cbdaff3bbe66cbc7fef4e9dce9389d5c2fb3700 Mon Sep 17 00:00:00 2001 From: Leo Belanger Date: Mon, 19 Oct 2026 10:00:18 -0400 Subject: [PATCH 3/3] fix(postgresql): apply the extensions Database without blocking the deploy --- harmony/src/modules/postgresql/cnpg/crd.rs | 9 --- harmony/src/modules/postgresql/score_k8s.rs | 68 +++------------------ 2 files changed, 7 insertions(+), 70 deletions(-) diff --git a/harmony/src/modules/postgresql/cnpg/crd.rs b/harmony/src/modules/postgresql/cnpg/crd.rs index 8f74eff2..7433c543 100644 --- a/harmony/src/modules/postgresql/cnpg/crd.rs +++ b/harmony/src/modules/postgresql/cnpg/crd.rs @@ -62,7 +62,6 @@ impl Default for ClusterSpec { kind = "Database", plural = "databases", namespaced = true, - status = "DatabaseStatus", derive = "Default", schema = "disabled" )] @@ -85,14 +84,6 @@ pub struct DatabaseExtension { pub ensure: String, } -#[derive(Deserialize, Serialize, Clone, Debug, Default)] -#[serde(rename_all = "camelCase")] -pub struct DatabaseStatus { - pub applied: Option, - pub message: Option, - pub observed_generation: Option, -} - #[derive(Deserialize, Serialize, Clone, Debug, Default)] #[serde(rename_all = "camelCase")] pub struct Storage { diff --git a/harmony/src/modules/postgresql/score_k8s.rs b/harmony/src/modules/postgresql/score_k8s.rs index 5e231d4d..18914e75 100644 --- a/harmony/src/modules/postgresql/score_k8s.rs +++ b/harmony/src/modules/postgresql/score_k8s.rs @@ -305,58 +305,6 @@ impl K8sPostgreSQLInterpret { }) .collect(), }, - status: None, - } - } - - async fn wait_for_database( - &self, - topology: &T, - name: &str, - ) -> Result<(), InterpretError> { - let k8s_client = topology - .k8s_client() - .await - .map_err(|e| InterpretError::new(format!("Failed to get k8s client: {}", e)))?; - - let timeout = std::time::Duration::from_secs(120); - let start = std::time::Instant::now(); - let mut last_message = None; - - loop { - match k8s_client - .get_resource::(name, Some(&self.config.namespace)) - .await - { - // A status from an earlier generation says nothing about the extensions just applied. - Ok(Some(db)) => match db.status { - Some(status) if status.observed_generation == db.metadata.generation => { - if status.applied == Some(true) { - info!("Extensions {:?} applied", self.config.extensions); - return Ok(()); - } - last_message = status.message; - } - _ => debug!("CNPG Database '{name}' not reconciled yet"), - }, - Ok(None) => debug!("CNPG Database '{name}' not found yet"), - Err(e) => debug!("Error checking CNPG Database '{name}': {e}"), - } - - if start.elapsed() > timeout { - return Err(InterpretError::new(format!( - "CNPG Database '{}' in '{}' did not apply extensions {:?} within {}s: {}", - name, - self.config.namespace, - self.config.extensions, - timeout.as_secs(), - last_message - .as_deref() - .unwrap_or("no status reported by CNPG") - ))); - } - - tokio::time::sleep(std::time::Duration::from_secs(2)).await; } } } @@ -409,15 +357,13 @@ impl Interpret for K8sPostgr } if !self.config.extensions.is_empty() { - let database = self.app_database(); - let name = database.metadata.name.clone().unwrap_or_default(); - K8sResourceScore::single(database, Some(self.config.namespace.clone())) - .create_interpret() - .execute(inventory, topology) - .await?; - if self.config.wait_for_ready { - self.wait_for_database(topology, &name).await?; - } + K8sResourceScore::single( + self.app_database(), + Some(self.config.namespace.clone()), + ) + .create_interpret() + .execute(inventory, topology) + .await?; } Ok(outcome) -- 2.39.5