fix: use a static kopia version, configurable

using :latest was unadvisable due to instability but also kubernetes's
pull policy for :latest images is Always, which puts more load on than
necessary and can even hit rate limits

Author: passcod

.github/workflows/integration.yml

@@ -123,11 +123,11 @@ jobs:
         run: |
           docker pull minio/minio:latest
           docker pull minio/mc:latest
-          docker pull kopia/kopia:latest
+          docker pull kopia/kopia:0.22.3
           docker pull postgres:16
           docker pull postgres:16-alpine
           docker pull alpine:latest
-          kind load docker-image minio/minio:latest minio/mc:latest kopia/kopia:latest postgres:16 postgres:16-alpine alpine:latest
+          kind load docker-image minio/minio:latest minio/mc:latest kopia/kopia:0.22.3 postgres:16 postgres:16-alpine alpine:latest
 
       - name: Pre-pull CNPG images
         if: matrix.needs_cnpg

operator.yaml

@@ -111,6 +111,7 @@ metadata:
     namespace: pgro-system
 data:
     maxConcurrentRestores: "2"
+    kopiaImage: "kopia/kopia:0.22.3"
 ---
 apiVersion: v1
 kind: Service

src/bin/operator.rs

@@ -16,7 +16,7 @@ use tower_http::trace::TraceLayer;
 use tracing::{debug, info, warn};
 
 use postgres_restore_operator::{
-	context::Context,
+	context::{Context, DEFAULT_KOPIA_IMAGE},
 	controllers,
 	types::{PostgresPhysicalReplica, PostgresPhysicalRestore},
 };
@@ -50,13 +50,28 @@ fn extract_max_concurrent_restores(cm: &ConfigMap) -> usize {
 		.unwrap_or(DEFAULT_MAX_CONCURRENT_RESTORES)
 }
 
-async fn read_max_concurrent_restores(client: &Client, namespace: &str) -> usize {
+fn extract_kopia_image(cm: &ConfigMap) -> String {
+	cm.data
+		.as_ref()
+		.and_then(|d| d.get("kopiaImage"))
+		.filter(|v| !v.is_empty())
+		.cloned()
+		.unwrap_or_else(|| DEFAULT_KOPIA_IMAGE.to_string())
+}
+
+async fn read_config(client: &Client, namespace: &str) -> (usize, String) {
 	let api: Api<ConfigMap> = Api::namespaced(client.clone(), namespace);
 	match api.get(CONFIGMAP_NAME).await {
-		Ok(cm) => extract_max_concurrent_restores(&cm),
+		Ok(cm) => (
+			extract_max_concurrent_restores(&cm),
+			extract_kopia_image(&cm),
+		),
 		Err(e) => {
-			warn!(error = %e, "failed to read ConfigMap {CONFIGMAP_NAME}, defaulting to {DEFAULT_MAX_CONCURRENT_RESTORES}");
-			DEFAULT_MAX_CONCURRENT_RESTORES
+			warn!(error = %e, "failed to read ConfigMap {CONFIGMAP_NAME}, using defaults");
+			(
+				DEFAULT_MAX_CONCURRENT_RESTORES,
+				DEFAULT_KOPIA_IMAGE.to_string(),
+			)
 		}
 	}
 }
@@ -77,17 +92,22 @@ async fn main() -> anyhow::Result<()> {
 	let client = Client::try_default().await?;
 
 	let namespace = operator_namespace();
-	let max_concurrent_restores = read_max_concurrent_restores(&client, &namespace).await;
+	let (max_concurrent_restores, kopia_image) = read_config(&client, &namespace).await;
 
 	info!(
 		max_concurrent_restores,
+		kopia_image,
 		metrics_addr,
 		operator_namespace = namespace,
 		version = env!("CARGO_PKG_VERSION"),
 		"starting postgres-restore-operator"
 	);
 
-	let ctx = Arc::new(Context::new(client.clone(), max_concurrent_restores));
+	let ctx = Arc::new(Context::new(
+		client.clone(),
+		max_concurrent_restores,
+		kopia_image,
+	));
 
 	// Heartbeat: a background task updates this timestamp every 5s.
 	// If the runtime is deadlocked, the timestamp goes stale and /livez fails.
@@ -106,6 +126,7 @@ async fn main() -> anyhow::Result<()> {
 	let config_watcher_config =
 		Config::default().fields(&format!("metadata.name={CONFIGMAP_NAME}"));
 	let max_concurrent_ref = ctx.max_concurrent_restores.clone();
+	let kopia_image_ref = ctx.kopia_image.clone();
 	tokio::spawn(async move {
 		let stream = watcher::watcher(config_api, config_watcher_config);
 		futures::pin_mut!(stream);
@@ -121,6 +142,17 @@ async fn main() -> anyhow::Result<()> {
 							"max_concurrent_restores updated from ConfigMap"
 						);
 					}
+
+					let new_image = extract_kopia_image(&cm);
+					let mut image = kopia_image_ref.write().unwrap();
+					if *image != new_image {
+						info!(
+							old = %*image,
+							new = new_image,
+							"kopia_image updated from ConfigMap"
+						);
+						*image = new_image;
+					}
 				}
 				Ok(watcher::Event::Delete(_)) => {
 					let old_val =
@@ -132,6 +164,16 @@ async fn main() -> anyhow::Result<()> {
 							"ConfigMap deleted, reverted max_concurrent_restores to default"
 						);
 					}
+
+					let mut image = kopia_image_ref.write().unwrap();
+					if *image != DEFAULT_KOPIA_IMAGE {
+						info!(
+							old = %*image,
+							new = DEFAULT_KOPIA_IMAGE,
+							"ConfigMap deleted, reverted kopia_image to default"
+						);
+						*image = DEFAULT_KOPIA_IMAGE.to_string();
+					}
 				}
 				Ok(watcher::Event::Init | watcher::Event::InitDone) => {}
 				Err(e) => {

src/context.rs

@@ -1,27 +1,30 @@
-use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, RwLock};
 
 use chrono::{DateTime, Utc};
 use kube::Client;
-use tokio::sync::RwLock;
 
 use crate::metrics::Metrics;
 
+pub const DEFAULT_KOPIA_IMAGE: &str = "kopia/kopia:0.22.3";
+
 pub struct Context {
 	pub client: Client,
 	pub metrics: Metrics,
-	pub restore_queue: Arc<RwLock<RestoreQueue>>,
+	pub restore_queue: Arc<tokio::sync::RwLock<RestoreQueue>>,
 	pub max_concurrent_restores: Arc<AtomicUsize>,
+	pub kopia_image: Arc<RwLock<String>>,
 	pub http_client: reqwest::Client,
 }
 
 impl Context {
-	pub fn new(client: Client, max_concurrent_restores: usize) -> Self {
+	pub fn new(client: Client, max_concurrent_restores: usize, kopia_image: String) -> Self {
 		Self {
 			client,
 			metrics: Metrics::new(),
-			restore_queue: Arc::new(RwLock::new(RestoreQueue::default())),
+			restore_queue: Arc::new(tokio::sync::RwLock::new(RestoreQueue::default())),
 			max_concurrent_restores: Arc::new(AtomicUsize::new(max_concurrent_restores)),
+			kopia_image: Arc::new(RwLock::new(kopia_image)),
 			http_client: reqwest::Client::new(),
 		}
 	}
@@ -30,6 +33,10 @@ impl Context {
 		self.max_concurrent_restores.load(Ordering::Relaxed)
 	}
 
+	pub fn kopia_image(&self) -> String {
+		self.kopia_image.read().unwrap().clone()
+	}
+
 	/// Remove a restore from the queue, promote the next pending one if there
 	/// is capacity, and update the related gauges. Returns the name of the
 	/// promoted restore, if any.

src/controllers/replica.rs

@@ -474,7 +474,12 @@ pub async fn reconcile(replica: Arc<PostgresPhysicalReplica>, ctx: Arc<Context>)
 		match jobs.get_opt(&snapshot_job_name).await? {
 			None => {
 				info!(replica = name, "creating snapshot list job");
-				let job = build_snapshot_list_job(&replica, &snapshot_job_name, &namespace)?;
+				let job = build_snapshot_list_job(
+					&replica,
+					&snapshot_job_name,
+					&namespace,
+					&ctx.kopia_image(),
+				)?;
 				jobs.create(&PostParams::default(), &job).await?;
 				return Ok(Action::requeue(Duration::from_secs(10)));
 			}

src/controllers/replica/resources.rs

@@ -60,6 +60,7 @@ pub fn build_snapshot_list_job(
 	replica: &PostgresPhysicalReplica,
 	job_name: &str,
 	namespace: &str,
+	kopia_image: &str,
 ) -> Result<Job> {
 	let kopia_secret = &replica.spec.kopia_secret_ref;
 	let replica_name = replica.name_any();
@@ -181,7 +182,7 @@ printf '{"id":"%s","size":%s}' "$ID" "$SIZE" > /dev/termination-log
 					restart_policy: Some("Never".to_string()),
 					containers: vec![Container {
 						name: "snapshot-list".to_string(),
-						image: Some("kopia/kopia:latest".to_string()),
+						image: Some(kopia_image.to_string()),
 						command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
 						args: Some(vec![script.to_string()]),
 						env: Some(env_vars),

src/controllers/restore.rs

@@ -191,7 +191,8 @@ async fn reconcile_restoring(
 			let replicas: Api<PostgresPhysicalReplica> = Api::namespaced(client.clone(), namespace);
 			let replica = replicas.get(&restore.spec.replica).await?;
 
-			let job = build_restore_job(restore, &job_name, namespace, &replica)?;
+			let job =
+				build_restore_job(restore, &job_name, namespace, &replica, &ctx.kopia_image())?;
 			jobs.create(&PostParams::default(), &job).await?
 		}
 	};

src/controllers/restore/builders.rs

@@ -213,6 +213,7 @@ pub fn build_restore_job(
 	job_name: &str,
 	namespace: &str,
 	replica: &PostgresPhysicalReplica,
+	kopia_image: &str,
 ) -> Result<Job> {
 	let kopia_secret = &replica.spec.kopia_secret_ref;
 	let pvc_name = format!("{}-data", restore.name_any());
@@ -321,7 +322,7 @@ echo -n "$VERSION" > /dev/termination-log
 
 					containers: vec![Container {
 						name: "restore".to_string(),
-						image: Some("kopia/kopia:latest".to_string()),
+						image: Some(kopia_image.to_string()),
 						command: Some(vec!["/bin/sh".to_string(), "-c".to_string()]),
 						args: Some(vec![restore_script.to_string()]),
 						env: Some(

tests/fixtures/setup-kopia-repo.yaml

@@ -53,7 +53,7 @@ spec:
                         mountPath: /pgdata
             containers:
                 - name: create-snapshot
-                  image: kopia/kopia:latest
+                  image: kopia/kopia:0.22.3
                   imagePullPolicy: IfNotPresent
                   command: ["/bin/sh", "-c"]
                   args:

tests/fixtures/setup-non-postgres-snapshot.yaml

@@ -21,7 +21,7 @@ spec:
                           echo "Bucket created"
             containers:
                 - name: create-snapshot
-                  image: kopia/kopia:latest
+                  image: kopia/kopia:0.22.3
                   imagePullPolicy: IfNotPresent
                   command: ["/bin/sh", "-c"]
                   args: