From 45b662f50980064f41d944e11ead0c5227a446b2 Mon Sep 17 00:00:00 2001 From: perfectra1n Date: Thu, 17 Sep 2026 11:38:40 -0700 Subject: [PATCH 01/59] fix(controller): set KOPIA_LOG_DIR on the kopia server Deployment (#461) The server Deployment set KOPIA_CONFIG_PATH and KOPIA_CACHE_DIRECTORY but never KOPIA_LOG_DIR, so kopia fell back to its unwritable default log location and spammed stdout with log-write errors. Point it at /cache/logs, under the cache emptyDir already mounted at SERVER_CACHE_DIR (no new volume needed), and use the shared kopiur_kopia::env constants for all three instead of hand-written string literals. --- crates/controller/src/server/mod.rs | 11 ++++-- crates/controller/src/server/tests.rs | 52 +++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/crates/controller/src/server/mod.rs b/crates/controller/src/server/mod.rs index 1955b62f..5318bb1f 100644 --- a/crates/controller/src/server/mod.rs +++ b/crates/controller/src/server/mod.rs @@ -80,6 +80,8 @@ pub const SERVER_CONFIG_DIR: &str = "/config"; pub const SERVER_CONFIG_FILE: &str = "/config/repository.config"; /// Writable kopia cache dir (emptyDir). pub const SERVER_CACHE_DIR: &str = "/cache"; +/// kopia log dir, under [`SERVER_CACHE_DIR`] so it needs no volume of its own. +pub const SERVER_LOG_DIR: &str = "/cache/logs"; /// The `Deployment`/`Service`/`ConfigMap` name for a repository's server. pub fn server_object_name(instance: &str) -> String { @@ -380,15 +382,20 @@ pub fn build_server_deployment(inputs: &ServerBuildInputs<'_>) -> Deployment { value_from: None, }, EnvVar { - name: "KOPIA_CONFIG_PATH".to_string(), + name: kopiur_kopia::env::CONFIG_PATH_ENV.to_string(), value: Some(SERVER_CONFIG_FILE.to_string()), value_from: None, }, EnvVar { - name: "KOPIA_CACHE_DIRECTORY".to_string(), + name: kopiur_kopia::env::CACHE_DIRECTORY_ENV.to_string(), value: Some(SERVER_CACHE_DIR.to_string()), value_from: None, }, + EnvVar { + name: kopiur_kopia::env::LOG_DIR_ENV.to_string(), + value: Some(SERVER_LOG_DIR.to_string()), + value_from: None, + }, EnvVar { name: "KOPIA_CHECK_FOR_UPDATES".to_string(), value: Some("false".to_string()), diff --git a/crates/controller/src/server/tests.rs b/crates/controller/src/server/tests.rs index a36eb4ee..ca65b114 100644 --- a/crates/controller/src/server/tests.rs +++ b/crates/controller/src/server/tests.rs @@ -263,6 +263,58 @@ fn deployment_password_auth_injects_server_password_env_from_secret() { assert_eq!(env_from[0].secret_ref.as_ref().unwrap().name, "nas-creds"); } +#[test] +fn deployment_sets_kopia_config_cache_and_log_dir_env() { + // Regression (#461): the server Deployment set KOPIA_CONFIG_PATH and + // KOPIA_CACHE_DIRECTORY but never KOPIA_LOG_DIR, so kopia's file logs fell + // back to its unwritable default and spammed the container's stdout with + // write errors. All three must point under the writable `cache` emptyDir. + let dep = build_server_deployment(&inputs("ns", gen_auth())); + let spec = dep.spec.unwrap(); + let pod_spec = spec.template.spec.unwrap(); + let c = &pod_spec.containers[0]; + let env = c.env.as_ref().unwrap(); + + let get = |name: &str| { + env.iter() + .find(|e| e.name == name) + .map(|e| e.value.clone().unwrap()) + .unwrap_or_else(|| panic!("missing {name} env var")) + }; + assert_eq!(get(kopiur_kopia::env::CONFIG_PATH_ENV), SERVER_CONFIG_FILE); + assert_eq!( + get(kopiur_kopia::env::CACHE_DIRECTORY_ENV), + SERVER_CACHE_DIR + ); + assert_eq!(get(kopiur_kopia::env::LOG_DIR_ENV), SERVER_LOG_DIR); + + // /cache/logs needs no volume of its own: it lives under the `cache` + // emptyDir already mounted at SERVER_CACHE_DIR. + assert!( + SERVER_LOG_DIR.starts_with(&format!("{SERVER_CACHE_DIR}/")), + "log dir must live under the cache mount" + ); + let cache_mount = pod_spec + .volumes + .as_ref() + .unwrap() + .iter() + .find(|v| v.name == "cache") + .expect("cache volume present"); + assert!( + cache_mount.empty_dir.is_some(), + "cache volume must be an emptyDir" + ); + let cache_vm = c + .volume_mounts + .as_ref() + .unwrap() + .iter() + .find(|vm| vm.name == "cache") + .expect("cache volumeMount present"); + assert_eq!(cache_vm.mount_path, SERVER_CACHE_DIR); +} + #[test] fn deployment_env_from_includes_every_creds_secret_in_order() { // Regression (#416): the server used to env-inject only the encryption From bd5714cdfe01cae19261958e37b946dd49519abb Mon Sep 17 00:00:00 2001 From: Michael Elsdorfer Date: Mon, 7 Sep 2026 19:03:05 +0100 Subject: [PATCH 02/59] feat(api,kopia): stream source + stream restore types and the stdin snapshot path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the `stream` SnapshotPolicy source (capture a command's stdout as one virtual file) and the `streamExec` Restore target, plus the kopia plumbing that makes them safe. The load-bearing part is WHO decides to commit. kopia finalizes a stdin snapshot when — and only when — stdin reaches EOF, so `run_with_stdin` keeps ownership of the child's stdin and lets the caller return Commit or Abort AFTER the producer's exit status is known: Commit closes the pipe, Abort kills kopia with the pipe still open so no manifest is ever written. That is what makes "a producer that failed can never leave a retained, restorable snapshot" true by construction rather than by cleaning up afterwards. Two bugs the new integration tests caught, both of which would have shipped truncated dumps as successful backups: * a feed closure that OWNED ChildStdin dropped it on completion, closing the pipe and committing the snapshot before the outcome could be inspected; * draining stdout alongside the feed deadlocked — the drain waits for kopia to exit, kopia waits for stdin to close, and stdin closes after the drain. The drains are now background tasks and the runner owns the writer. Source keeps its sibling-Option wire shape (the forms share sourcePath*/readOnly keys, which an enum could only express via a flatten kube's schema rewriter cannot represent), so nothing on the wire breaks. Type-safety moves up a layer instead: source_shape() resolves it into the SourceShape enum and the reconcile paths match on that exhaustively, so a fifth source form cannot compile until every handler accounts for it. fileName is validated as ONE safe path segment because kopia stores it verbatim as the entry name without sanitizing: a value of '../x' makes a later 'kopia restore ' write outside (verified against 0.23.1), and an empty value makes kopia ignore stdin and snapshot the local filesystem instead. (cherry picked from commit 0763b759b5e6b40a6c6a2a4ce681e62c699c8a40) --- crates/api/src/expand.rs | 24 +- crates/api/src/expand/tests.rs | 7 +- crates/api/src/lib.rs | 5 +- crates/api/src/restore.rs | 24 ++ crates/api/src/snapshot_policy.rs | 183 ++++++++++- crates/api/src/validate/mod.rs | 197 +++++++++--- crates/api/src/validate/restore.rs | 13 + crates/api/src/validate/snapshot.rs | 39 +++ crates/controller/src/restore/mod.rs | 289 ++++++++++++----- crates/controller/src/restore/plan.rs | 6 +- crates/kopia/src/client/mod.rs | 299 +++++++++++++++++- crates/kopia/src/client/tests.rs | 13 +- crates/kopia/src/error.rs | 26 ++ crates/kopia/src/lib.rs | 4 +- crates/kopia/tests/integration_stdin.rs | 234 ++++++++++++++ crates/mover/src/status.rs | 4 + crates/webhook/src/handlers.rs | 5 +- deploy/crds/all-crds.yaml | 177 ++++++++++- deploy/crds/restores.yaml | 82 +++++ deploy/crds/snapshotpolicies.yaml | 95 +++++- deploy/helm/kopiur/crds/restores.yaml | 82 +++++ deploy/helm/kopiur/crds/snapshotpolicies.yaml | 95 +++++- docs/field-reference.md | 40 ++- 23 files changed, 1769 insertions(+), 174 deletions(-) create mode 100644 crates/kopia/tests/integration_stdin.rs diff --git a/crates/api/src/expand.rs b/crates/api/src/expand.rs index 110446aa..3db906b6 100644 --- a/crates/api/src/expand.rs +++ b/crates/api/src/expand.rs @@ -67,7 +67,9 @@ const FANOUT_MARKER: &str = "-pvc-"; /// The single source one `Snapshot` actually backs up, after resolving /// `spec.source` (a fanned-out child) against the policy's `sources[]`. -#[derive(Debug, Clone, PartialEq, Eq)] +// `PartialEq` only, never `Eq`: `stream` carries a `StreamExec`, which embeds +// k8s-openapi's `LabelSelector` — a `PartialEq`-only type (api-conventions §3). +#[derive(Debug, Clone, PartialEq)] pub struct EffectiveSource { /// Index into `policy.spec.sources` these knobs came from. pub index: usize, @@ -75,6 +77,10 @@ pub struct EffectiveSource { pub pvc: Option, /// The NFS export path, when this is an NFS source. pub nfs_path: Option, + /// The stream producer, when this is a `stream` source. Mounts nothing: the mover + /// execs a command and pipes its stdout into kopia, so no `pvc`/`nfs_path` is set + /// and `read_only` is meaningless (see [`EffectiveSource::read_only`]). + pub stream: Option, /// `sourcePathOverride` from the governing source. pub source_path_override: Option, /// Whether the mount is read-only. @@ -99,6 +105,12 @@ impl EffectiveSource { } }); } + // A stream source records `/stream/` — deliberately a different + // root from `/pvc/...` so a streamed artifact can never share a kopia + // identity with a volume backup. + if let Some(stream) = &self.stream { + return Some(crate::snapshot_policy::stream_source_path(stream)); + } self.nfs_path.clone() } } @@ -137,6 +149,7 @@ pub fn effective_source( index, pvc, nfs_path: source.nfs.as_ref().map(|n| n.path.clone()), + stream: source.stream.clone(), source_path_override: source.source_path_override.clone(), read_only, }; @@ -384,6 +397,8 @@ pub fn restore_source_path( index, pvc: Some(target.clone()), nfs_path: None, + // Matched by its own plain `pvc:` source, so PVC-shaped by construction. + stream: None, source_path_override: source.source_path_override.clone(), read_only: snapshot_policy::source_read_only(source), }; @@ -434,6 +449,9 @@ pub fn restore_source_path( index: 0, pvc: Some(target.clone()), nfs_path: None, + // A selector-derived path is PVC-shaped; a stream source never + // participates in a selector derivation. + stream: None, source_path_override: None, read_only: true, }; @@ -1024,6 +1042,10 @@ pub fn expand_sources( index, pvc: Some(target.clone()), nfs_path: None, + // A selector source is always PVC-shaped; a stream source never + // expands (it is one artifact per Snapshot, and admission requires + // it to be its policy's only source). + stream: None, source_path_override: source.source_path_override.clone(), read_only: snapshot_policy::source_read_only(source), }; diff --git a/crates/api/src/expand/tests.rs b/crates/api/src/expand/tests.rs index 300d11b7..d805d5da 100644 --- a/crates/api/src/expand/tests.rs +++ b/crates/api/src/expand/tests.rs @@ -37,6 +37,7 @@ fn selector_source(strategy: Option, namespaces: Vec<&str>) label_selector: None, }), nfs: None, + stream: None, read_only: None, acknowledge_live_mutation: None, source_path_override: None, @@ -51,6 +52,7 @@ fn pvc_source(name: &str) -> Source { }), pvc_selector: None, nfs: None, + stream: None, read_only: None, acknowledge_live_mutation: None, source_path_override: None, @@ -930,10 +932,7 @@ fn nfs_source(path: &str) -> Source { server: "nas.lan".into(), path: path.to_string(), }), - read_only: None, - acknowledge_live_mutation: None, - source_path_override: None, - source_path_strategy: None, + ..Default::default() } } diff --git a/crates/api/src/lib.rs b/crates/api/src/lib.rs index f0cbb53c..04422606 100644 --- a/crates/api/src/lib.rs +++ b/crates/api/src/lib.rs @@ -81,8 +81,9 @@ pub use snapshot::{ }; pub use snapshot_policy::{ CopyMethod, DeepVerification, GroupBy, Hook, PolicyRepositories, SnapshotPolicy, - SnapshotPolicySpec, SnapshotPolicyStatus, SourcePathStrategy, StagingSpec, Verification, - is_multi_repo, policy_repositories, repository_refs, single_repository_ref, + SnapshotPolicySpec, SnapshotPolicyStatus, SourcePathStrategy, SourceShape, StagingSpec, + StreamExec, StreamSource, Verification, is_multi_repo, policy_repositories, repository_refs, + single_repository_ref, source_shape, stream_source_path, }; pub use snapshot_replication::{ IdentityMatcher, IdentitySelection, MigrateOptions, MirrorSourcePruning, NoPruning, diff --git a/crates/api/src/restore.rs b/crates/api/src/restore.rs index fb066a8b..aa06392c 100644 --- a/crates/api/src/restore.rs +++ b/crates/api/src/restore.rs @@ -5,6 +5,7 @@ use crate::common::{ CredentialProjection, FailurePolicy, MoverSpec, ObjectRef, PvcAccessMode, RepositoryRef, ResolvedIdentity, }; +use crate::snapshot_policy::StreamExec; use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition; use kube::CustomResource; use schemars::JsonSchema; @@ -178,6 +179,10 @@ pub enum RestoreTarget { PvcRef(ObjectRef), /// Passive populator mode: the restore is claimed by a PVC's `spec.dataSourceRef`. Populator(PopulatorTarget), + /// Stream one virtual file out of the snapshot into a command's stdin in a + /// running Pod — the companion of a `stream` backup source (e.g. feeding a + /// `pg_dumpall` artifact back through `psql`). Writes no PVC. + StreamExec(StreamExecTarget), } impl RestoreTarget { @@ -200,10 +205,29 @@ impl RestoreTarget { RestoreTarget::Pvc(_) => "Pvc", RestoreTarget::PvcRef(_) => "PvcRef", RestoreTarget::Populator(_) => "Populator", + RestoreTarget::StreamExec(_) => "StreamExec", } } } +/// Feed one virtual file from the snapshot into a command's stdin, in exactly one +/// running Pod. +/// +/// The reverse of a `stream` backup source. Deliberately opt-in and deliberately +/// blunt: it runs whatever you name against whatever the selector matches, so point +/// it at a scratch or otherwise prepared database, never at a live production Pod +/// you are not willing to overwrite. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct StreamExecTarget { + /// Which virtual file inside the snapshot to read back — the `fileName` the + /// backup's stream source used (e.g. `postgres.sql`). + #[schemars(length(min = 1, max = 255))] + pub file_name: String, + /// Where to send it: the command receives the file's bytes on stdin. + pub workload_exec: StreamExec, +} + /// Passive-populator target marker; its presence selects populator mode. #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)] #[serde(rename_all = "camelCase")] diff --git a/crates/api/src/snapshot_policy.rs b/crates/api/src/snapshot_policy.rs index c3ae57a5..54f0c64b 100644 --- a/crates/api/src/snapshot_policy.rs +++ b/crates/api/src/snapshot_policy.rs @@ -308,7 +308,8 @@ pub fn effective_on_policy_delete( deletion.map(|d| d.on_policy_delete).unwrap_or_default() } -/// A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs` (webhook-enforced). +/// A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs`, `stream` +/// (webhook-enforced). // The exactly-one-of rule is written as an integer sum of `has()` ternaries rather // than `[...].filter(x,x).size()==1`: the apiserver estimates per-item CEL cost × // `maxItems`, and a list-construction + lambda `filter` blows the budget on the @@ -316,11 +317,18 @@ pub fn effective_on_policy_delete( // `Default` is derived purely for construction ergonomics: `Source` is built as an // exhaustive struct literal in ~20 places, and every added field would otherwise have // to be spelled out at each one. An all-`None` `Source` is not a valid spec (the CEL -// rule above demands exactly one of pvc/pvcSelector/nfs) and admission rejects it. +// rule above demands exactly one of pvc/pvcSelector/nfs/stream) and admission rejects it. +// +// The forms stay sibling `Option`s rather than an externally-tagged enum because they +// SHARE the `sourcePath*`/`readOnly` keys (see `crate::validate::validate_source`), which +// an enum could only express with a `#[serde(flatten)]` kube's structural-schema rewriter +// cannot represent. The type-safety thesis is upheld one layer up instead: [`source_shape`] +// resolves this struct into the [`SourceShape`] enum, and every reconcile path matches on +// THAT exhaustively — so a fifth form cannot compile until each handler accounts for it. #[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, JsonSchema)] #[schemars(extend("x-kubernetes-validations" = [{ - "rule": "(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) == 1", - "message": "exactly one of pvc, pvcSelector, nfs" + "rule": "(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) + (has(self.stream) ? 1 : 0) == 1", + "message": "exactly one of pvc, pvcSelector, nfs, stream" }]))] #[serde(rename_all = "camelCase")] pub struct Source { @@ -330,9 +338,15 @@ pub struct Source { /// Label/namespace selector matching many PVCs. Mutually exclusive with `pvc`/`nfs`. #[serde(default, skip_serializing_if = "Option::is_none")] pub pvc_selector: Option, - /// An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`. + /// An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`/`stream`. #[serde(default, skip_serializing_if = "Option::is_none")] pub nfs: Option, + /// Capture a command's standard output as one virtual file — a logical backup + /// (`pg_dumpall`, `mysqldump`, …) rather than a volume copy. Mutually exclusive + /// with `pvc`/`pvcSelector`/`nfs`. No volume is mounted; the mover execs the + /// command in a running workload Pod and streams its stdout straight into kopia. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stream: Option, /// Mount the source read-only (default `true`; kopia only ever reads it). /// /// Set `false` **only** to make `fsGroup` work on the source. The kubelet applies @@ -368,6 +382,165 @@ pub struct Source { pub source_path_strategy: Option, } +/// The virtual directory a stream source's single file lives under, so a streamed +/// artifact can never collide with a PVC source's `/pvc/` namespace. +pub const STREAM_SOURCE_ROOT: &str = "/stream"; + +/// Capture a command's standard output as ONE virtual file inside a normal kopia +/// snapshot — the logical-backup source (`pg_dumpall`, `mysqldump`, …). +/// +/// Nothing is mounted: the mover execs `workloadExec.command` in a running workload +/// Pod and pipes its stdout directly into `kopia snapshot create --stdin-file`. The +/// snapshot root is a virtual directory at `/stream/` (override with +/// `sourcePathOverride`) containing exactly that one file. +/// +/// The command runs in the WORKLOAD's container, so it uses the database credentials +/// already present there; the mover never hands it repository credentials. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct StreamSource { + /// Name of the single virtual file stored in the snapshot (e.g. `postgres.sql`). + /// + /// Must be ONE file name: no `/`, and not `.` or `..`. kopia stores this string + /// verbatim as the entry name and does not sanitize it, so a path-shaped value + /// would make a later `kopia restore` write OUTSIDE its destination directory. + #[schemars(length(min = 1, max = 255))] + pub file_name: String, + /// The producer: what to run, and where. Its stdout IS the backup data. + pub workload_exec: StreamExec, +} + +/// Exec a command in exactly one running workload Pod, streaming one of its +/// standard streams. Used as the producer of a `stream` backup source and as the +/// consumer of a `streamExec` restore target. +/// +/// Exactly one RUNNING Pod must match `podSelector`: zero, several, or only +/// not-running matches are named failures, never an arbitrary pick — a backup that +/// silently dumped a different replica than intended is worse than one that stops. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)] +#[serde(rename_all = "camelCase")] +pub struct StreamExec { + /// Standard label selector identifying the workload Pod, resolved in the + /// `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + /// empty selector matches every Pod in the namespace. + pub pod_selector: LabelSelector, + /// Container to exec in; absent uses the Pod's default container. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub container: Option, + /// The argv to execute. NOT a shell line: element 0 is the program, so use + /// `["sh", "-ec", "..."]` explicitly if you want shell semantics. + /// + /// Reference credentials through the container's existing environment or mounted + /// Secrets — never inline them here. This argv is copied into the mover Pod's + /// spec, so anyone with `pods:get` in the namespace can read it. + #[schemars(length(min = 1))] + pub command: Vec, + /// Go duration bounding the command (e.g. `2h`); absent uses + /// [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + /// fails, leaving no snapshot behind. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timeout: Option, +} + +/// Default bound on a stream producer/consumer command when `timeout` is unset. +/// Generous enough for a large logical dump, finite so a wedged command cannot pin +/// a mover Job forever. +pub const DEFAULT_STREAM_TIMEOUT_SECS: u64 = 3600; + +/// The four mutually-exclusive forms a [`Source`] can take, resolved from its +/// sibling `Option`s. +/// +/// THE point of this type: `Source` must stay a struct of `Option`s on the wire (the +/// forms share `sourcePath*`/`readOnly` keys), but every reconcile path matches on +/// this enum EXHAUSTIVELY — so a fifth source form cannot compile until backup Job +/// construction, validation, and identity resolution each account for it. Borrowed, +/// so callers match without cloning. +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum SourceShape<'a> { + /// One PVC named directly. + Pvc(&'a PvcSource), + /// A label/namespace selector matching many PVCs (fans out into child Snapshots). + PvcSelector(&'a PvcSelector), + /// An inline NFS export read directly. + Nfs(&'a NfsVolume), + /// A command whose stdout is captured as one virtual file. + Stream(&'a StreamSource), +} + +impl SourceShape<'_> { + /// Stable discriminant string for status, metrics, and messages. + pub fn kind_str(&self) -> &'static str { + match self { + SourceShape::Pvc(_) => "pvc", + SourceShape::PvcSelector(_) => "pvcSelector", + SourceShape::Nfs(_) => "nfs", + SourceShape::Stream(_) => "stream", + } + } +} + +/// THE exactly-one-of resolver for a [`Source`]'s form. +/// +/// Never panics: a stored CR can carry an invalid shape (none set, or several — e.g. a +/// write that raced an old CRD schema), so those come back as a +/// [`ValidationError`](crate::error::ValidationError) for the caller's error path rather +/// than as an `unwrap` in a reconciler. Mirrors +/// [`policy_repositories`](crate::snapshot_policy::policy_repositories). +/// +/// ``` +/// use kopiur_api::snapshot_policy::{Source, SourceShape, PvcSource, source_shape}; +/// +/// let s = Source { pvc: Some(PvcSource { name: "data".into() }), ..Default::default() }; +/// assert!(matches!(source_shape(&s), Ok(SourceShape::Pvc(p)) if p.name == "data")); +/// +/// // Nothing set is a named error, never a silent default. +/// assert!(source_shape(&Source::default()).is_err()); +/// ``` +pub fn source_shape(source: &Source) -> Result, crate::error::ValidationError> { + let set: Vec<&'static str> = [ + ("pvc", source.pvc.is_some()), + ("pvcSelector", source.pvc_selector.is_some()), + ("nfs", source.nfs.is_some()), + ("stream", source.stream.is_some()), + ] + .into_iter() + .filter_map(|(name, present)| present.then_some(name)) + .collect(); + + match set.as_slice() { + [] => Err(crate::error::ValidationError::MissingRequiredField { + field: "source.pvc, source.pvcSelector, source.nfs, or source.stream".to_string(), + }), + [first, second, ..] => Err(crate::error::ValidationError::MutuallyExclusive { + a: (*first).to_string(), + b: (*second).to_string(), + context: "snapshot source".to_string(), + }), + // Exactly one is set; return the one that is. The `expect`s cannot fire — + // each arm is reached only when its own `is_some()` put the name in `set`. + ["pvc"] => Ok(SourceShape::Pvc(source.pvc.as_ref().expect("pvc is set"))), + ["pvcSelector"] => Ok(SourceShape::PvcSelector( + source.pvc_selector.as_ref().expect("pvcSelector is set"), + )), + ["nfs"] => Ok(SourceShape::Nfs(source.nfs.as_ref().expect("nfs is set"))), + ["stream"] => Ok(SourceShape::Stream( + source.stream.as_ref().expect("stream is set"), + )), + // Unreachable: `set` is built from exactly the four names above. + [other] => Err(crate::error::ValidationError::InvalidFieldValue { + field: "spec.sources[]".to_string(), + reason: format!("unknown source form `{other}`"), + }), + } +} + +/// The kopia source path a stream source records by default: `/stream/`. +/// Deliberately distinct from a PVC source's `/pvc/` so a streamed artifact and +/// a volume backup can never share a kopia identity. +pub fn stream_source_path(stream: &StreamSource) -> String { + format!("{STREAM_SOURCE_ROOT}/{}", stream.file_name) +} + /// A single backup source addressed by PVC name. #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)] #[serde(rename_all = "camelCase")] diff --git a/crates/api/src/validate/mod.rs b/crates/api/src/validate/mod.rs index bbd19ddd..eb9dade0 100644 --- a/crates/api/src/validate/mod.rs +++ b/crates/api/src/validate/mod.rs @@ -18,7 +18,7 @@ use crate::backend::NfsVolume; use crate::common::{FailurePolicy, MoverSpec, PvcAccessMode, RepositoryMode}; use crate::error::{ValidationError, ValidationResult}; use crate::server::{ServerAuth, ServerSpec}; -use crate::snapshot_policy::Source; +use crate::snapshot_policy::{Source, SourceShape, StreamExec, StreamSource}; use k8s_openapi::api::core::v1::ResourceRequirements; use kube_quantity::ParsedQuantity; @@ -113,56 +113,157 @@ fn validate_pvc_selector(selector: &crate::snapshot_policy::PvcSelector) -> Vali } /// A single backup `Source` is well-formed: **exactly one** of `pvc`, -/// `pvcSelector`, or `nfs` is set (ADR §3.3 — modeled as sibling Options because -/// the forms share `sourcePath*` keys, so it's a webhook check, not an enum). When -/// the source is `nfs`, its server/path are also validated. +/// `pvcSelector`, `nfs`, or `stream` is set (ADR §3.3 — modeled as sibling Options +/// because the forms share `sourcePath*` keys, so it's a webhook check, not an +/// enum), and that form's own content is valid. +/// +/// The dispatch is an EXHAUSTIVE match over [`SourceShape`] rather than a chain of +/// `if let`s: a fifth source form then cannot compile until it is given content +/// validation here, instead of silently falling through with none. pub fn validate_source(source: &Source) -> ValidationResult { - let set: Vec<&str> = [ - ("pvc", source.pvc.is_some()), - ("pvcSelector", source.pvc_selector.is_some()), - ("nfs", source.nfs.is_some()), - ] - .into_iter() - .filter_map(|(name, present)| present.then_some(name)) - .collect(); - - match set.as_slice() { - [] => Err(ValidationError::MissingRequiredField { - field: "source.pvc, source.pvcSelector, or source.nfs".to_string(), - }), - [first, second, ..] => Err(ValidationError::MutuallyExclusive { - a: (*first).to_string(), - b: (*second).to_string(), - context: "snapshot source".to_string(), - }), - [_only] => match &source.nfs { - Some(nfs) => { - // `readOnly: false` exists for exactly one purpose — letting the kubelet - // apply `fsGroup` to the source — and the kubelet does not apply `fsGroup` - // to in-tree NFS volumes at all. So on NFS it buys nothing and only makes - // the export writable to the mover. Reject it rather than ship a knob that - // silently does the opposite of what its user wants. - if !crate::snapshot_policy::source_read_only(source) { - return Err(ValidationError::InvalidFieldValue { - field: "spec.sources[].readOnly".to_string(), - reason: "readOnly: false is not supported on an nfs source: the kubelet \ - does not apply fsGroup to in-tree NFS volumes, so a read-write \ - mount grants the mover no additional readability and only \ - exposes the export to writes. Remove readOnly (NFS is read \ - directly), and grant access with mover.podSecurityContext \ - supplementalGroups / mover.securityContext runAsUser matching \ - the export's ownership, or with a server-side ID remap" - .to_string(), - }); - } - validate_nfs_volume(nfs, "snapshot source") + match crate::snapshot_policy::source_shape(source)? { + SourceShape::Pvc(_) => Ok(()), + SourceShape::PvcSelector(selector) => validate_pvc_selector(selector), + SourceShape::Nfs(nfs) => { + // `readOnly: false` exists for exactly one purpose — letting the kubelet + // apply `fsGroup` to the source — and the kubelet does not apply `fsGroup` + // to in-tree NFS volumes at all. So on NFS it buys nothing and only makes + // the export writable to the mover. Reject it rather than ship a knob that + // silently does the opposite of what its user wants. + if !crate::snapshot_policy::source_read_only(source) { + return Err(ValidationError::InvalidFieldValue { + field: "spec.sources[].readOnly".to_string(), + reason: "readOnly: false is not supported on an nfs source: the kubelet \ + does not apply fsGroup to in-tree NFS volumes, so a read-write \ + mount grants the mover no additional readability and only \ + exposes the export to writes. Remove readOnly (NFS is read \ + directly), and grant access with mover.podSecurityContext \ + supplementalGroups / mover.securityContext runAsUser matching \ + the export's ownership, or with a server-side ID remap" + .to_string(), + }); } - None => match source.pvc_selector.as_ref() { - Some(selector) => validate_pvc_selector(selector), - None => Ok(()), - }, - }, + validate_nfs_volume(nfs, "snapshot source") + } + SourceShape::Stream(stream) => validate_stream_source(source, stream), + } +} + +/// A `stream` source is well-formed: a safe single-segment `fileName`, a runnable +/// `command`, a non-empty `podSelector`, a parseable `timeout`, and none of the +/// PVC-only knobs that cannot mean anything without a mounted volume. +fn validate_stream_source(source: &Source, stream: &StreamSource) -> ValidationResult { + validate_stream_file_name("spec.sources[].stream.fileName", &stream.file_name)?; + validate_stream_exec("spec.sources[].stream.workloadExec", &stream.workload_exec)?; + + // PVC-only knobs. A stream source mounts nothing, so `readOnly` has no volume to + // apply to, `acknowledgeLiveMutation` acknowledges a mutation that cannot happen, + // and `sourcePathStrategy` derives a path from a PVC that does not exist. Reject + // rather than ignore: silently accepting a field that does nothing is how a user + // comes to believe a protection is in place when it is not. + if source.read_only.is_some() { + return Err(ValidationError::InvalidFieldValue { + field: "spec.sources[].readOnly".to_string(), + reason: "readOnly does not apply to a stream source: nothing is mounted — the \ + mover execs a command and pipes its stdout into kopia. Remove readOnly" + .to_string(), + }); + } + if source.acknowledge_live_mutation.is_some() { + return Err(ValidationError::InvalidFieldValue { + field: "spec.sources[].acknowledgeLiveMutation".to_string(), + reason: "acknowledgeLiveMutation does not apply to a stream source: it \ + acknowledges the kubelet rewriting a mounted volume's ownership, and a \ + stream source mounts no volume. Remove acknowledgeLiveMutation" + .to_string(), + }); + } + if source.source_path_strategy.is_some() { + return Err(ValidationError::InvalidFieldValue { + field: "spec.sources[].sourcePathStrategy".to_string(), + reason: "sourcePathStrategy derives a kopia path from a matched PVC's name and \ + applies only to a pvcSelector source. A stream source records \ + /stream/; set sourcePathOverride to change it" + .to_string(), + }); + } + Ok(()) +} + +/// `fileName` is ONE safe path segment. +/// +/// Security-relevant, not cosmetic: kopia stores the `--stdin-file` value verbatim as +/// the snapshot entry's name and does not sanitize it, so an entry named `../x` makes +/// a later `kopia restore ` write OUTSIDE `` (verified against kopia +/// 0.23.1). An empty value is worse still — kopia then ignores stdin entirely and +/// tries to snapshot the source path from the mover's own filesystem. +pub fn validate_stream_file_name(field: &str, name: &str) -> ValidationResult { + let bad = name.is_empty() + || name.contains('/') + || name == "." + || name == ".." + || name.contains('\0') + || name.chars().any(|c| c.is_control()); + if bad { + return Err(ValidationError::InvalidFieldValue { + field: field.to_string(), + reason: format!( + "`{name}` is not a usable file name. Give ONE file name (e.g. `postgres.sql`): \ + no `/`, not `.` or `..`, no control characters, not empty. kopia stores this \ + value verbatim as the entry name inside the snapshot, so a path-shaped value \ + would make a later restore write outside its destination directory" + ), + }); + } + Ok(()) +} + +/// A [`StreamExec`] is runnable: a non-empty argv, a selector that identifies +/// something, and a parseable timeout. Shared by the backup source and the restore +/// target so both reject the same mistakes with the same words. +pub fn validate_stream_exec(field: &str, exec: &StreamExec) -> ValidationResult { + if exec.command.is_empty() { + return Err(ValidationError::InvalidFieldValue { + field: format!("{field}.command"), + reason: "give the argv to run, e.g. [\"sh\", \"-ec\", \"pg_dumpall -U postgres\"]. \ + This is exec'd directly, not through a shell, so element 0 is the program" + .to_string(), + }); } + if exec.command.iter().all(|a| a.trim().is_empty()) { + return Err(ValidationError::InvalidFieldValue { + field: format!("{field}.command"), + reason: "the command is entirely blank; element 0 must name the program to run" + .to_string(), + }); + } + let selector_empty = exec + .pod_selector + .match_labels + .as_ref() + .is_none_or(|m| m.is_empty()) + && exec + .pod_selector + .match_expressions + .as_ref() + .is_none_or(|e| e.is_empty()); + if selector_empty { + return Err(ValidationError::InvalidFieldValue { + field: format!("{field}.podSelector"), + reason: "the podSelector is empty, which matches EVERY pod in the namespace. Set \ + matchLabels/matchExpressions identifying the one workload pod to exec into" + .to_string(), + }); + } + if let Some(t) = &exec.timeout + && crate::duration::parse_go_duration(t).is_none() + { + return Err(ValidationError::InvalidFieldValue { + field: format!("{field}.timeout"), + reason: format!("`{t}` is not a Go duration. Use forms like `30m`, `2h`, `90s`"), + }); + } + Ok(()) } /// An inline [`NfsVolume`] is well-formed: a non-empty server and an absolute diff --git a/crates/api/src/validate/restore.rs b/crates/api/src/validate/restore.rs index 996b6783..0c232664 100644 --- a/crates/api/src/validate/restore.rs +++ b/crates/api/src/validate/restore.rs @@ -131,6 +131,19 @@ pub fn validate_restore(spec: &RestoreSpec) -> ValidationResult { } } } + // A stream target writes no PVC: it pipes ONE virtual file out of the + // snapshot into a command's stdin. Validate the same things the backup-side + // stream source validates, with the same words. + RestoreTarget::StreamExec(t) => { + crate::validate::validate_stream_file_name( + "restore.target.streamExec.fileName", + &t.file_name, + )?; + crate::validate::validate_stream_exec( + "restore.target.streamExec.workloadExec", + &t.workload_exec, + )?; + } RestoreTarget::Pvc(_) | RestoreTarget::PvcRef(_) => {} } // Access modes on a create-target PVC: canonical/unique/RWOP-sole. Fail-fast on diff --git a/crates/api/src/validate/snapshot.rs b/crates/api/src/validate/snapshot.rs index f60ebd8c..543ee5c6 100644 --- a/crates/api/src/validate/snapshot.rs +++ b/crates/api/src/validate/snapshot.rs @@ -156,6 +156,36 @@ pub fn validate_backup_config(spec: &SnapshotPolicySpec) -> Vec .to_string(), }); } + // Same reasoning as the NFS arm above: a stream source has no PVC to snapshot. + if spec.volume_snapshot_class_name.is_some() && spec.sources.iter().any(|s| s.stream.is_some()) + { + errs.push(ValidationError::InvalidFieldValue { + field: "spec.volumeSnapshotClassName".to_string(), + reason: "a stream source captures a command's stdout and mounts no volume, so \ + volumeSnapshotClassName is meaningless with it; remove \ + volumeSnapshotClassName, or use a PVC source for copyMethod: Snapshot/Clone" + .to_string(), + }); + } + // A stream source must be its policy's ONLY source. Expansion fans out selector + // sources only (`expand_sources` returns None when no `pvcSelector` is present), so + // a non-selector policy runs `sources[0]` and nothing else — pairing a stream with + // another source would silently back up just one of them. Refuse instead of + // dropping a backup the user believes is configured. + if spec.sources.len() > 1 + && let Some(i) = spec.sources.iter().position(|s| s.stream.is_some()) + { + errs.push(ValidationError::InvalidFieldValue { + field: format!("spec.sources[{i}].stream"), + reason: format!( + "a stream source must be the only source in its SnapshotPolicy, but this one \ + has {}. A stream source produces exactly one artifact per Snapshot and is \ + never expanded, so the other sources would not be backed up. Move the stream \ + source into its own SnapshotPolicy", + spec.sources.len() + ), + }); + } if let Some(m) = &spec.mover { // `inheritSecurityContextFrom.snapshot` replays a backup's RECORDED identity; // a backup has no recorded identity to replay — it is the run that records one. @@ -427,6 +457,15 @@ fn validate_staging(spec: &SnapshotPolicySpec) -> Vec { .to_string(), }); } + if spec.sources.iter().any(|s| s.stream.is_some()) { + errs.push(ValidationError::InvalidFieldValue { + field: overrides.clone(), + reason: "a stream source captures a command's stdout and mounts no volume, so \ + there is no staged PVC to override; remove the override(s) or use a PVC \ + source for copyMethod: Snapshot/Clone" + .to_string(), + }); + } // NOTE: `pvcSelector` sources used to be rejected here on the grounds that // they "are not CSI-staged". That was true only because the selector was // never implemented (#346). Each expanded member is now an ordinary diff --git a/crates/controller/src/restore/mod.rs b/crates/controller/src/restore/mod.rs index 17d6e8ac..224a8f4c 100644 --- a/crates/controller/src/restore/mod.rs +++ b/crates/controller/src/restore/mod.rs @@ -303,15 +303,19 @@ async fn drive_direct_target( // policy carried the SAME cross-volume hazard as the populator: a pathless // identity matches the newest snapshot of ANY member. Exhaustive over // `RestoreTarget`. - let source_target = match &restore.spec.target { - RestoreTarget::Pvc(t) => kopiur_api::snapshot::PvcTargetRef { + // `None` for a stream target: it fills no volume, so there is nothing to + // derive a per-PVC path from — the identity is the policy's own + // `/stream/` (see `from_policy_identity`). + let source_target: Option = match &restore.spec.target { + RestoreTarget::Pvc(t) => Some(kopiur_api::snapshot::PvcTargetRef { namespace: namespace.to_string(), name: t.name.clone(), - }, - RestoreTarget::PvcRef(r) => kopiur_api::snapshot::PvcTargetRef { + }), + RestoreTarget::PvcRef(r) => Some(kopiur_api::snapshot::PvcTargetRef { namespace: r.namespace.clone().unwrap_or_else(|| namespace.to_string()), name: r.name.clone(), - }, + }), + RestoreTarget::StreamExec(_) => None, RestoreTarget::Populator(_) => { return Err(Error::Invariant( "DirectTarget restore reached with a populator target (should route to \ @@ -357,7 +361,7 @@ async fn drive_direct_target( restore, namespace, wait_window.anchor(), - &source_target, + source_target.as_ref(), &mut cache, ) .await? @@ -507,7 +511,7 @@ async fn drive_direct_target( namespace, name, selection.as_ref(), - &source_target, + source_target.as_ref(), ) .await } @@ -2045,11 +2049,13 @@ async fn claim_populate( let mut mover_in_flight = false; let mut record = if let Some(selection) = selection.as_ref() { let dispatch = RestoreDispatch { - target_pvc: &cc.prime_name, - source_target: kopiur_api::snapshot::PvcTargetRef { + // The populator always writes into the prime PVC it just provisioned; + // `target.streamExec` never routes here (it is a DirectTarget). + destination: RestoreDestination::Pvc(&cc.prime_name), + source_target: Some(kopiur_api::snapshot::PvcTargetRef { namespace: namespace.to_string(), name: cc.consumer_name.clone(), - }, + }), claim_key: Some(&cc.consumer_name), job_reuse: cc.job_reuse, }; @@ -2227,25 +2233,33 @@ async fn resolve_claim_source( name: cc.consumer_name.clone(), }; let window = claim_wait_window(cc.prev, chrono::Utc::now().timestamp()); - let outcome = - match resolve_snapshot(ctx, restore, namespace, window.anchor(), &target, cache).await? { - SourceResolution::Resolved(outcome) => outcome, - // Fail CLOSED: the policy names no single volume for this claim, so - // restoring anything would be a guess — quite possibly with another - // volume's data. Scoped to THIS claim; siblings are unaffected. - SourceResolution::Ambiguous(msg) => { - return Ok(ClaimResolution::Parked(Box::new(ClaimOutcome { - record: cc.record( - RestoreClaimPhase::Failed, - ClaimReason::SourcePathAmbiguous, - msg, - ), - requeue: 300, - event: None, - after_patch: None, - }))); - } - }; + let outcome = match resolve_snapshot( + ctx, + restore, + namespace, + window.anchor(), + Some(&target), + cache, + ) + .await? + { + SourceResolution::Resolved(outcome) => outcome, + // Fail CLOSED: the policy names no single volume for this claim, so + // restoring anything would be a guess — quite possibly with another + // volume's data. Scoped to THIS claim; siblings are unaffected. + SourceResolution::Ambiguous(msg) => { + return Ok(ClaimResolution::Parked(Box::new(ClaimOutcome { + record: cc.record( + RestoreClaimPhase::Failed, + ClaimReason::SourcePathAmbiguous, + msg, + ), + requeue: 300, + event: None, + after_patch: None, + }))); + } + }; match outcome { Some(ResolveOutcome::Pinned(res)) => { *source_path = res @@ -2690,21 +2704,26 @@ async fn drive_direct_restore( namespace: &str, name: &str, selection: Option<&RestoreSelection>, - source_target: &kopiur_api::snapshot::PvcTargetRef, + source_target: Option<&kopiur_api::snapshot::PvcTargetRef>, ) -> Result { // Resolve the target PVC for the restore Job. DirectTarget is only reached for // an explicit PVC target (populator routes to AwaitingClaim in the reconcile // dispatch). Exhaustive over RestoreTarget so a new variant must be considered. - let target_pvc = match &restore.spec.target { - RestoreTarget::PvcRef(r) => r.name.clone(), + // Owned PVC name (when there is one) so the borrow in `destination` outlives + // the awaits below. + let target_pvc: Option = match &restore.spec.target { + RestoreTarget::PvcRef(r) => Some(r.name.clone()), // `target.pvc` means the operator CREATES the PVC (ADR §3.6) — without // this the mover Job references a claim nobody made and sits Pending // forever (FailedScheduling: persistentvolumeclaim not found). This also // provisions the empty volume for the deploy-or-restore (no-snapshot) case. RestoreTarget::Pvc(t) => { ensure_restore_target_pvc(ctx, namespace, t).await?; - t.name.clone() + Some(t.name.clone()) } + // A stream target creates nothing: the mover reads one virtual file out of + // the snapshot and pipes it into a command's stdin. + RestoreTarget::StreamExec(_) => None, RestoreTarget::Populator(_) => { return Err(Error::Invariant( "DirectTarget restore reached with a populator target (should route to \ @@ -2713,6 +2732,16 @@ async fn drive_direct_restore( )); } }; + let destination = match (&restore.spec.target, target_pvc.as_deref()) { + (RestoreTarget::StreamExec(t), _) => RestoreDestination::Stream(t), + (_, Some(pvc)) => RestoreDestination::Pvc(pvc), + // Unreachable: every non-stream direct arm above produced a name. + (_, None) => { + return Err(Error::Invariant( + "direct restore resolved no target PVC and no stream target".into(), + )); + } + }; let phase = restore.status.as_ref().and_then(|s| s.phase.as_ref()); @@ -2751,8 +2780,8 @@ async fn drive_direct_restore( // The Job is named after the Restore and writes into the explicit target PVC; // the helper creates/tracks it, the phase writes stay here. let dispatch = RestoreDispatch { - target_pvc: &target_pvc, - source_target: source_target.clone(), + destination, + source_target: source_target.cloned(), // A direct restore keeps the classic top-level status shape: it is the // only writer, so there is nothing to key by claim. claim_key: None, @@ -2862,18 +2891,20 @@ async fn drive_direct_restore( /// What ONE restore mover dispatch is filling, and how it reports. /// -/// A struct rather than four more positional parameters because -/// `target_pvc` and `source_target` are DIFFERENT PVCs on the populator path +/// A struct rather than four more positional parameters because the WRITTEN +/// destination and `source_target` are DIFFERENT PVCs on the populator path /// (the prime is written; the CLAIMANT is what the per-PVC kopia source path is /// derived from, #443) and swapping them silently restores the wrong volume. struct RestoreDispatch<'a> { - /// The PVC the mover WRITES into, mounted read-write at `/restore`: the - /// direct target, or a populator prime. - target_pvc: &'a str, - /// The PVC whose DATA this is. Same as `target_pvc` for a direct restore; - /// the claiming PVC (never the prime) for a populator. The per-PVC source - /// path and the recorded-identity catalog row are both keyed off it. - source_target: kopiur_api::snapshot::PvcTargetRef, + /// Where the mover puts the bytes: a PVC mounted read-write at `/restore` + /// (the direct target, or a populator prime), or a command's stdin. + destination: RestoreDestination<'a>, + /// The PVC whose DATA this is. Same as the destination PVC for a direct + /// restore; the claiming PVC (never the prime) for a populator. The per-PVC + /// source path and the recorded-identity catalog row are both keyed off it. + /// `None` for a stream destination, which fills no volume and therefore has + /// no per-PVC derivation — its path is the policy's own `/stream/`. + source_target: Option, /// The `status.claims.` the mover nests its status writes under; /// `None` keeps the classic top-level shape (direct restores). claim_key: Option<&'a str>, @@ -2881,6 +2912,15 @@ struct RestoreDispatch<'a> { job_reuse: JobNameReuse, } +impl<'a> RestoreDispatch<'a> { + /// The PVC this dispatch WRITES into, or `None` for a stream destination. + /// Reads at each call site as "this step needs a PVC" rather than + /// re-matching the destination enum. + fn pvc(&self) -> Option<&'a str> { + self.destination.pvc() + } +} + /// Observed state of a restore mover Job, returned by [`run_restore_mover`]. #[derive(Debug, Clone, PartialEq, Eq)] enum MoverOutcome { @@ -3064,6 +3104,33 @@ async fn reserve_restore_slot( .await } +/// Where a direct restore writes: a mounted PVC, or a command's stdin in a +/// running Pod. Borrowed and exhaustive, so a new [`RestoreTarget`] variant must +/// decide how the mover Job is shaped before it compiles. +#[derive(Debug, Clone, Copy)] +pub(super) enum RestoreDestination<'a> { + /// The classic path: the mover mounts this PVC read-write at `/restore` and + /// kopia writes the snapshot's files into it. + Pvc(&'a str), + /// The stream path: the mover reads ONE virtual file out of the snapshot and + /// pipes it into a command's stdin. No volume is mounted, so none of the + /// PVC-shaped machinery (colocation, target-PVC securityContext assessment) + /// applies. + Stream(&'a kopiur_api::restore::StreamExecTarget), +} + +impl<'a> RestoreDestination<'a> { + /// The target PVC name, or `None` for a stream destination. Used by the + /// PVC-only steps so each reads as "this step needs a PVC" rather than + /// re-matching the enum. + fn pvc(&self) -> Option<&'a str> { + match self { + RestoreDestination::Pvc(name) => Some(name), + RestoreDestination::Stream(_) => None, + } + } +} + #[allow(clippy::too_many_arguments)] async fn run_restore_mover( ctx: &Context, @@ -3075,7 +3142,6 @@ async fn run_restore_mover( dispatch: &RestoreDispatch<'_>, ) -> Result { use k8s_openapi::api::batch::v1::Job; - let target_pvc = dispatch.target_pvc; let job_api: Api = Api::namespaced(ctx.client.clone(), namespace); if let Some(job) = job_api.get_opt(job_name).await? { // A Job being DELETED is not THIS run's outcome — but only where the Job name is @@ -3217,7 +3283,7 @@ async fn run_restore_mover( restore, namespace, dispatch_pinned_id, - &dispatch.source_target, + dispatch.source_target.as_ref(), ) .await { @@ -3382,15 +3448,20 @@ async fn run_restore_mover( // consumer of the target PVC can read what the mover writes (matching UID / shared fsGroup // on the fresh volume). Never writes `False` — restore has no certain signal, so the // advisory negative lives in the admission warning. Never fatal. - assess_restore_security_context( - namespace, - restore, - target_pvc, - &resolved_mover.security_context, - resolved_mover.pod_security_context.as_ref(), - ctx, - ) - .await; + // PVC-only: the assessment compares the mover's identity against the FUTURE + // consumer of the target volume. A stream destination writes no volume, so + // there is no consumer and nothing to assess. + if let Some(target_pvc) = dispatch.pvc() { + assess_restore_security_context( + namespace, + restore, + target_pvc, + &resolved_mover.security_context, + resolved_mover.pod_security_context.as_ref(), + ctx, + ) + .await; + } let owner = io::owner_ref_for(restore, "Restore")?; let repo_ref = restore.spec.repository.as_ref(); @@ -3560,25 +3631,36 @@ async fn run_restore_mover( // MissingDependency retry, never the Snapshot-side terminal machinery // (Restore's parking precedent, `waitTimeout`/`onMissing`, is about // snapshots, not PVCs). - let decision = match io::resolve_source_colocation( - &ctx.client, - namespace, - target_pvc, - resolved_mover.source_colocation, - ) - .await? - { - io::ColocationOutcome::Resolved(decision) => decision, - io::ColocationOutcome::SourcePvcAbsent { - namespace: pvc_ns, - name: pvc_name, - } => return Err(restore_target_pvc_race_error(&pvc_ns, &pvc_name)), - }; - io::apply_colocation( - decision, - resolved_mover.affinity.clone(), - resolved_mover.tolerations.clone(), - )? + // Colocation places the mover next to the PVC it writes. A stream + // destination has no PVC to be near, so there is nothing to colocate on — + // yield the recipe's own affinity/tolerations unchanged. + match dispatch.pvc() { + None => ( + resolved_mover.affinity.clone(), + resolved_mover.tolerations.clone(), + ), + Some(target_pvc) => { + let decision = match io::resolve_source_colocation( + &ctx.client, + namespace, + target_pvc, + resolved_mover.source_colocation, + ) + .await? + { + io::ColocationOutcome::Resolved(decision) => decision, + io::ColocationOutcome::SourcePvcAbsent { + namespace: pvc_ns, + name: pvc_name, + } => return Err(restore_target_pvc_race_error(&pvc_ns, &pvc_name)), + }; + io::apply_colocation( + decision, + resolved_mover.affinity.clone(), + resolved_mover.tolerations.clone(), + )? + } + } }; let inputs = MoverJobInputs { name: job_name, @@ -3623,7 +3705,11 @@ async fn run_restore_mover( labels }, // Restore writes INTO the target PVC, mounted read-write at /restore. - source_volume: Some(VolumeMountSpec::pvc(target_pvc, target_path, false)), + // A stream destination mounts nothing — the bytes go straight from kopia + // into an exec stdin, never touching a filesystem. + source_volume: dispatch + .pvc() + .map(|pvc| VolumeMountSpec::pvc(pvc, target_path, false)), repo_volume, creds_secrets, result_configmap: None, @@ -3809,7 +3895,7 @@ async fn snapshot_recorded_source( restore: &Restore, namespace: &str, pinned_id: Option<&str>, - target: &kopiur_api::snapshot::PvcTargetRef, + target: Option<&kopiur_api::snapshot::PvcTargetRef>, ) -> Result> { if !snapshot_inherit_active(restore) { return Ok(None); @@ -4205,15 +4291,58 @@ async fn ensure_restore_target_pvc( /// [`Error`]: it is a DOMAIN outcome (a valid `Restore` plus a valid /// `SnapshotPolicy` that together name no single volume), reported on the /// affected claim's own record rather than as a reconcile failure. +/// The policy's OWN kopia source path, for a restore whose target is not a PVC +/// (today: `target.streamExec`). +/// +/// The per-PVC derivation in [`kopiur_api::expand::restore_source_path`] answers +/// "which member of a selector policy fills THIS volume" — a question a stream +/// restore does not ask, because it fills no volume. A stream policy is +/// admission-restricted to a single `stream` source, so its effective source is +/// unambiguous and `kopia_source_path` yields `/stream/`. An explicit +/// `source.fromPolicy.sourcePath` still wins, exactly as it does on the per-PVC +/// path. +/// +/// Returns the ready-to-surface message (not an [`Error`]) for the same reason +/// [`from_policy_identity`] does: it is a DOMAIN outcome about the spec pair. +fn policy_own_source_path( + config: &kopiur_api::SnapshotPolicy, + source_path_override: Option<&str>, +) -> std::result::Result { + if let Some(over) = source_path_override { + return Ok(kopiur_api::expand::RestoreSourcePath::Override( + over.to_string(), + )); + } + let eff = kopiur_api::expand::effective_source(config, None).map_err(|e| e.to_string())?; + // `sources[0]` exists whenever `effective_source(_, None)` succeeded. + let strategy = config + .spec + .sources + .first() + .map(kopiur_api::expand::strategy_for) + .unwrap_or(kopiur_api::snapshot_policy::SourcePathStrategy::PvcName); + Ok(kopiur_api::expand::RestoreSourcePath::PolicySource( + eff.kopia_source_path(strategy), + )) +} + fn from_policy_identity( config: &kopiur_api::SnapshotPolicy, config_namespace: &str, defaults: Option<&kopiur_api::IdentityDefaults>, source_path_override: Option<&str>, - target: &kopiur_api::snapshot::PvcTargetRef, + target: Option<&kopiur_api::snapshot::PvcTargetRef>, ) -> std::result::Result { - let path = - restore_source_path(config, source_path_override, target).map_err(|e| e.to_string())?; + let path = match target { + Some(target) => { + restore_source_path(config, source_path_override, target).map_err(|e| e.to_string())? + } + // No target PVC means a stream restore: it fills no volume, so there is + // nothing to derive a per-PVC path from. The identity is the policy's OWN + // path (`/stream/` for a stream source), with an explicit + // `sourcePath` override honoured exactly as `restore_source_path` does. + None => policy_own_source_path(config, source_path_override)?, + }; crate::snapshot_policy::config_identity_for_path( config, config_namespace, @@ -4240,7 +4369,7 @@ async fn resolve_snapshot( restore: &Restore, namespace: &str, wait_anchor: i64, - target: &kopiur_api::snapshot::PvcTargetRef, + target: Option<&kopiur_api::snapshot::PvcTargetRef>, cache: &mut PassCache, ) -> Result { use kopiur_api::common::{ObjectRef, ResolvedIdentity}; diff --git a/crates/controller/src/restore/plan.rs b/crates/controller/src/restore/plan.rs index c7bc63fd..11fbe22b 100644 --- a/crates/controller/src/restore/plan.rs +++ b/crates/controller/src/restore/plan.rs @@ -74,7 +74,11 @@ pub fn restore_job_duration_seconds(job: &k8s_openapi::api::batch::v1::Job) -> O pub fn populator_state(target: &RestoreTarget) -> PopulatorState { match target { RestoreTarget::Populator(_) => PopulatorState::AwaitingClaim, - RestoreTarget::Pvc(_) | RestoreTarget::PvcRef(_) => PopulatorState::DirectTarget, + // `streamExec` is operator-driven like pvc/pvcRef — the mover reads one + // virtual file and pipes it into a command. Nothing claims it. + RestoreTarget::Pvc(_) | RestoreTarget::PvcRef(_) | RestoreTarget::StreamExec(_) => { + PopulatorState::DirectTarget + } } } diff --git a/crates/kopia/src/client/mod.rs b/crates/kopia/src/client/mod.rs index c157735f..41750e83 100644 --- a/crates/kopia/src/client/mod.rs +++ b/crates/kopia/src/client/mod.rs @@ -1083,6 +1083,23 @@ async fn kill_and_reap(child: &mut tokio::process::Child) { } /// The raw outcome of running a kopia subprocess. +/// What a stdin producer decided, handed back from the `feed` closure of +/// [`KopiaClient::snapshot_create_stdin_outcome_with`]. +/// +/// An enum rather than a `bool` because the two arms are not "success/failure" — they +/// are two different LIFECYCLE instructions for a live child process, and getting them +/// backwards silently commits a truncated backup. Naming them forces the caller to say +/// which one it means. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StdinOutcome { + /// The producer finished successfully and has closed the writer. Let kopia see + /// EOF, finalize the snapshot, and exit. + Commit, + /// The producer did not finish successfully. Kill kopia with stdin still open so + /// it never writes a manifest — no partial snapshot is left behind. + Abort, +} + struct RawOutput { code: Option, stdout: String, @@ -1135,12 +1152,14 @@ impl KopiaClient { /// credentials (remapped from their `KOPIUR_DEST_`-prefixed copies) while /// clearing any source credential the destination does not set, so a stale /// source `AWS_SESSION_TOKEN` (etc.) cannot leak into the destination auth. - async fn run_with_env( - &self, - args: &[String], - env_overlay: &BTreeMap>, - ) -> Result { - let display_args = args.join(" "); + /// The `kopia` [`Command`] with this client's environment and common args + /// applied, and NO stdio configured yet — every runner sets its own. + /// + /// Extracted so the streaming runners share exactly the environment the + /// ordinary one uses: a stdin-fed snapshot that silently missed + /// `common_env`/`common_args` would connect to a different repository, or skip + /// `--no-check-for-updates`, in a way no test would obviously catch. + fn base_command(&self, args: &[String]) -> Command { let mut cmd = Command::new(&self.binary); // Do not inherit the ambient environment's KOPIA_* unless the caller // set it explicitly; but we *do* inherit PATH etc. by default, which is @@ -1153,6 +1172,20 @@ impl KopiaClient { for k in &self.common_env_remove { cmd.env_remove(k); } + cmd.args(args); + // Append common args (e.g. --no-check-for-updates) after the subcommand + // tokens the caller passed. + cmd.args(&self.common_args); + cmd + } + + async fn run_with_env( + &self, + args: &[String], + env_overlay: &BTreeMap>, + ) -> Result { + let display_args = args.join(" "); + let mut cmd = self.base_command(args); // Per-invocation overlay wins over both the inherited env and common_env. for (k, v) in env_overlay { match v { @@ -1160,10 +1193,6 @@ impl KopiaClient { None => cmd.env_remove(k), }; } - cmd.args(args); - // Append common args (e.g. --no-check-for-updates) after the subcommand - // tokens the caller passed. - cmd.args(&self.common_args); cmd.stdin(Stdio::null()); cmd.stdout(Stdio::piped()); cmd.stderr(Stdio::piped()); @@ -1290,6 +1319,156 @@ impl KopiaClient { /// /// Callers that genuinely only want stdout keep using `run_ok`; the two /// that need to reason about a silent success use this. + /// Run kopia with its stdin PIPED, handing the writer to `feed`. + /// + /// The whole point is that `feed` decides whether the run commits: kopia only + /// finalizes when stdin hits EOF, so returning [`StdinOutcome::Commit`] (having + /// dropped the writer) lets it finish, while [`StdinOutcome::Abort`] kills it + /// with stdin still open so it writes nothing. See + /// [`Self::snapshot_create_stdin_outcome_with`] for why that matters. + /// + /// stdout/stderr are captured as usual. `feed`'s own error is returned verbatim, + /// after the child has been killed and reaped. + async fn run_with_stdin(&self, args: &[String], feed: F) -> Result + where + F: AsyncFnOnce(&mut tokio::process::ChildStdin) -> Result, + { + let mut cmd = self.base_command(args); + cmd.stdin(Stdio::piped()); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + let mut child = cmd.spawn().map_err(|source| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source, + })?; + // The RUNNER owns the writer, and `feed` only borrows it. That ownership is + // load-bearing, not a style choice: if `feed` owned it, the writer would be + // dropped when `feed`'s future completes — closing the pipe, giving kopia EOF, + // and letting it COMMIT the snapshot before we ever got to inspect the + // producer's outcome. Keeping it here is what makes `Abort` able to kill kopia + // with stdin still open. (Caught by + // `integration_stdin::abort_after_full_payload_leaves_no_snapshot`.) + let mut stdin = child.stdin.take().ok_or_else(|| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source: std::io::Error::other("kopia stdin pipe was not created"), + })?; + + // Drain stdout/stderr in BACKGROUND TASKS, not alongside `feed` in a `join!`. + // Draining to EOF only completes when kopia EXITS, and kopia cannot exit until + // stdin closes — which happens after `feed` returns. Awaiting both together + // therefore deadlocks: `feed` finishes writing, the drain waits for an exit + // that waits for the close that waits for the drain. Spawning the drains keeps + // the pipes flowing (so kopia's progress chatter never blocks its stdin read) + // while leaving the close/kill decision to us. + let mut out = child.stdout.take(); + let mut err = child.stderr.take(); + let out_task = tokio::spawn(async move { + let mut so = String::new(); + if let Some(o) = out.as_mut() { + let _ = o.read_to_string(&mut so).await; + } + so + }); + let err_task = tokio::spawn(async move { + let mut se = String::new(); + if let Some(e) = err.as_mut() { + let _ = e.read_to_string(&mut se).await; + } + se + }); + + let fed = feed(&mut stdin).await; + + match fed { + Ok(StdinOutcome::Commit) => { + // Close the writer NOW — this is the commit point. kopia sees EOF and + // finalizes the manifest. + drop(stdin); + let status = child.wait().await.map_err(|source| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source, + })?; + Ok(RawOutput { + code: status.code(), + stdout: out_task.await.unwrap_or_default(), + stderr: err_task.await.unwrap_or_default(), + }) + } + Ok(StdinOutcome::Abort) => { + // Kill with stdin STILL OPEN — `stdin` is dropped only after this — so + // kopia never reaches EOF and never writes a manifest. + let _ = child.kill().await; + drop(stdin); + let stderr = err_task.await.unwrap_or_default(); + let _ = out_task.await; + Err(KopiaError::StdinProducerFailed { + detail: "the stdin producer did not complete successfully; the kopia \ + snapshot was aborted before any manifest was written" + .to_string(), + stderr_tail: tail_lines(&stderr), + }) + } + Err(e) => { + // Same ordering as `Abort`: kill first, close second. + let _ = child.kill().await; + drop(stdin); + let _ = err_task.await; + let _ = out_task.await; + Err(e) + } + } + } + + /// Run kopia streaming its stdout into `sink` (constant memory), capturing only + /// stderr for diagnostics. Used by [`Self::show_to`], where stdout IS user data + /// and must never be collected into a `String`. + async fn run_streaming_stdout(&self, args: &[String], mut sink: W) -> Result<(), KopiaError> + where + W: tokio::io::AsyncWrite + Unpin, + { + let mut cmd = self.base_command(args); + cmd.stdin(Stdio::null()); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + let mut child = cmd.spawn().map_err(|source| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source, + })?; + let mut out = child.stdout.take().ok_or_else(|| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source: std::io::Error::other("kopia stdout pipe was not created"), + })?; + let mut err = child.stderr.take(); + let mut stderr = String::new(); + let copied = { + let drain = async { + if let Some(e) = err.as_mut() { + let _ = e.read_to_string(&mut stderr).await; + } + }; + let (copied, ()) = tokio::join!(tokio::io::copy(&mut out, &mut sink), drain); + copied + }; + let status = child.wait().await.map_err(|source| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source, + })?; + copied.map_err(|source| KopiaError::Spawn { + binary: self.binary.display().to_string(), + source, + })?; + if status.code() == Some(0) { + Ok(()) + } else { + Err(KopiaError::NonZeroExit { + args: args.join(" "), + code: status.code(), + class: KopiaErrorClass::classify(&stderr), + stderr_tail: tail_lines(&stderr), + }) + } + } + async fn run_ok_full( &self, args: &[String], @@ -1555,10 +1734,96 @@ impl KopiaClient { override_source: Option<&str>, opts: &SnapshotCreateOptions, ) -> Result { - let args = snapshot_create_args(source_path, tags, override_source, opts); + let args = snapshot_create_args(source_path, None, tags, override_source, opts); self.run_json(&args, "snapshot create result").await } + /// Create a snapshot whose CONTENT is fed to kopia on stdin + /// (`kopia snapshot create --stdin-file `), storing it as one + /// virtual regular file inside a virtual directory at `source_path`. + /// + /// # Why the caller decides when to commit + /// + /// `feed` BORROWS kopia's `ChildStdin` for the transfer and returns [`StdinOutcome`]. + /// kopia finalizes the snapshot when — and only when — stdin reaches EOF, so + /// **holding stdin open is what keeps the snapshot uncommitted**: + /// + /// * [`StdinOutcome::Commit`] — this method closes the writer, kopia sees EOF, + /// writes the manifest, and exits 0. + /// * [`StdinOutcome::Abort`] — kopia is KILLED with stdin still open, so it never + /// writes a manifest; the repository keeps only orphan content blobs, which + /// maintenance reclaims. + /// + /// This is verified behaviour, not an assumption: with the complete payload + /// already written but stdin still open, killing kopia 0.23.1 leaves no snapshot + /// for that source at all. It is what lets a caller guarantee "a producer that + /// failed can never leave a retained, restorable snapshot" with no + /// delete-after-the-fact and no race — the commit decision happens strictly after + /// the producer's exit status is known. + /// + /// `feed` must never write the streamed bytes anywhere but the writer it is given: + /// they are the user's data (a database dump), and this crate deliberately offers + /// no path by which they could reach a log line or an error message. + pub async fn snapshot_create_stdin_outcome_with( + &self, + source_path: &str, + stdin_file: &str, + tags: &BTreeMap, + override_source: Option<&str>, + opts: &SnapshotCreateOptions, + feed: F, + ) -> Result + where + F: AsyncFnOnce(&mut tokio::process::ChildStdin) -> Result, + { + let args = snapshot_create_args(source_path, Some(stdin_file), tags, override_source, opts); + let out = self.run_with_stdin(&args, feed).await?; + if out.code != Some(0) { + return Err(KopiaError::NonZeroExit { + args: args.join(" "), + code: out.code, + class: KopiaErrorClass::classify(&out.stderr), + stderr_tail: tail_lines(&out.stderr), + }); + } + match extract_json(&out.stdout) { + Some(json) => serde_json::from_str::(json) + .map(|r| SnapshotCreateOutcome::Created(Box::new(r))) + .map_err(|source| KopiaError::Json { + context: "snapshot create result".to_string(), + source, + }), + // A stdin snapshot has no previous tree to compare against, so kopia's + // `ignoreIdenticalSnapshots` shortcut cannot really apply — but decode it + // the same way rather than mis-reporting it as an empty-output failure. + None if crate::error::snapshot_skipped_unchanged(&out.stderr) => { + Ok(SnapshotCreateOutcome::Unchanged) + } + None => Err(KopiaError::EmptyOutput { + context: "snapshot create result".to_string(), + stderr_tail: tail_lines(&out.stderr), + }), + } + } + + /// Stream one object's bytes out of the repository (`kopia show `), + /// writing them into `sink`. + /// + /// `object_id` addresses an entry inside a snapshot as `/`. + /// Note a snapshot MANIFEST id does NOT work in that sub-path form; the root + /// entry's object id (`rootEntry.obj` from `snapshot list --json`) does. + /// + /// kopia streams the object, so this is constant-memory regardless of file size, + /// and the bytes go straight into `sink` — never buffered here, never logged, + /// never attached to an error. + pub async fn show_to(&self, object_id: &str, sink: W) -> Result<(), KopiaError> + where + W: tokio::io::AsyncWrite + Unpin, + { + let args: Vec = vec!["show".into(), object_id.to_string()]; + self.run_streaming_stdout(&args, sink).await + } + /// [`Self::snapshot_create_with`], but able to say "kopia deliberately /// wrote no manifest". /// @@ -1584,7 +1849,7 @@ impl KopiaClient { override_source: Option<&str>, opts: &SnapshotCreateOptions, ) -> Result { - let args = snapshot_create_args(source_path, tags, override_source, opts); + let args = snapshot_create_args(source_path, None, tags, override_source, opts); let out = self.run_ok_full(&args, &BTreeMap::new()).await?; match extract_json(&out.stdout) { Some(json) => serde_json::from_str::(json) @@ -2139,6 +2404,7 @@ fn restore_args(id: &str, target_dir: &str, opts: &RestoreOptions) -> Vec, tags: &BTreeMap, override_source: Option<&str>, opts: &SnapshotCreateOptions, @@ -2149,6 +2415,15 @@ fn snapshot_create_args( source_path.to_string(), "--json".into(), ]; + // `--stdin-file ` switches kopia from walking `source_path` on disk to + // reading STDIN and storing it as ONE regular file of that name inside a virtual + // directory rooted at `source_path`. The value is stored VERBATIM as the entry + // name — kopia does not sanitize it — which is why callers must have validated it + // as a single safe path segment (`validate_stream_file_name`). + if let Some(name) = stdin_file { + args.push("--stdin-file".into()); + args.push(name.to_string()); + } if let Some(src) = override_source { args.push("--override-source".into()); args.push(src.to_string()); diff --git a/crates/kopia/src/client/tests.rs b/crates/kopia/src/client/tests.rs index bd2a1d96..851ac4a6 100644 --- a/crates/kopia/src/client/tests.rs +++ b/crates/kopia/src/client/tests.rs @@ -620,12 +620,19 @@ fn snapshot_create_args_default_is_todays_argv() { let mut tags = BTreeMap::new(); tags.insert("app".to_string(), "db".to_string()); assert_eq!( - snapshot_create_args("/data", &tags, None, &SnapshotCreateOptions::default()), + snapshot_create_args( + "/data", + None, + &tags, + None, + &SnapshotCreateOptions::default() + ), vec!["snapshot", "create", "/data", "--json", "--tags", "app:db"] ); assert_eq!( snapshot_create_args( "/data", + None, &BTreeMap::new(), Some("u@h:/data"), &SnapshotCreateOptions::default() @@ -651,7 +658,7 @@ fn snapshot_create_args_fail_fast_upload_limit_and_description() { description: Some("smoke test".to_string()), }; assert_eq!( - snapshot_create_args("/data", &BTreeMap::new(), None, &opts), + snapshot_create_args("/data", None, &BTreeMap::new(), None, &opts), vec![ "snapshot", "create", @@ -672,7 +679,7 @@ fn snapshot_create_args_fail_fast_upload_limit_and_description() { ..Default::default() }; assert_eq!( - snapshot_create_args("/data", &BTreeMap::new(), None, &opts_false), + snapshot_create_args("/data", None, &BTreeMap::new(), None, &opts_false), vec!["snapshot", "create", "/data", "--json", "--no-fail-fast"] ); } diff --git a/crates/kopia/src/error.rs b/crates/kopia/src/error.rs index 36dca63f..026e112f 100644 --- a/crates/kopia/src/error.rs +++ b/crates/kopia/src/error.rs @@ -387,6 +387,23 @@ pub enum KopiaError { source: serde_json::Error, }, + /// The process feeding kopia's stdin did not finish successfully, so the + /// snapshot was deliberately aborted before kopia wrote any manifest. + /// + /// Distinct from [`KopiaError::NonZeroExit`]: kopia itself was fine — it was + /// killed on purpose. Surfacing that difference is what tells an operator + /// "your dump command failed" rather than "the backup tool failed". + #[error( + "the snapshot was aborted because its stdin producer failed: {detail}{}", + if stderr_tail.is_empty() { String::new() } else { format!(" (kopia stderr: {stderr_tail})") } + )] + StdinProducerFailed { + /// What went wrong with the producer, already free of any streamed data. + detail: String, + /// Bounded tail of kopia's own stderr, for context. + stderr_tail: String, + }, + /// We expected a JSON object/array on stdout but found none (kopia printed /// only progress / nothing) even though it exited **0**. /// @@ -426,6 +443,10 @@ impl KopiaError { KopiaError::Json { .. } | KopiaError::EmptyOutput { .. } => KopiaErrorClass::Unknown, // Timeouts are usually a slow backend → worth a retry. KopiaError::Timeout { .. } => KopiaErrorClass::RepositoryUnavailable, + // The repository was never at fault — the user's dump command was. + // Retrying the same Job re-runs the same failing command, so this is + // NOT retryable; the fix is in the workload or the policy. + KopiaError::StdinProducerFailed { .. } => KopiaErrorClass::Unknown, } } @@ -438,6 +459,11 @@ impl KopiaError { KopiaError::EmptyOutput { stderr_tail, .. } if !stderr_tail.is_empty() => { Some(stderr_tail.as_str()) } + // Bounded kopia stderr only. The producer's own diagnostics ride in + // `detail`, and its STDOUT — the backup data — is never captured at all. + KopiaError::StdinProducerFailed { stderr_tail, .. } if !stderr_tail.is_empty() => { + Some(stderr_tail.as_str()) + } _ => None, } } diff --git a/crates/kopia/src/lib.rs b/crates/kopia/src/lib.rs index eae75faf..d3f9e781 100644 --- a/crates/kopia/src/lib.rs +++ b/crates/kopia/src/lib.rs @@ -12,8 +12,8 @@ pub mod session; pub use client::{ CacheTuning, ConnectOptions, ConnectSpec, CreateOptions, KopiaClient, KopiaClientBuilder, MaintenanceMode, MigratePolicies, MigrateSources, PolicyArgs, RestoreOptions, ServerAuthMode, - ServerStartSpec, SnapshotCreateOptions, SnapshotMigrateOptions, SyncToOptions, ThrottleArgs, - VerifyOptions, split_policy_scopes, + ServerStartSpec, SnapshotCreateOptions, SnapshotMigrateOptions, StdinOutcome, SyncToOptions, + ThrottleArgs, VerifyOptions, split_policy_scopes, }; pub use error::{ KopiaError, KopiaErrorClass, notfound_is_uninitialized, snapshot_skipped_unchanged, diff --git a/crates/kopia/tests/integration_stdin.rs b/crates/kopia/tests/integration_stdin.rs new file mode 100644 index 00000000..37082bcb --- /dev/null +++ b/crates/kopia/tests/integration_stdin.rs @@ -0,0 +1,234 @@ +//! Real-kopia integration coverage for the stdin-fed snapshot path. +//! +//! Gated behind the `integration` feature and `#[ignore]` by default so the +//! hermetic `cargo test` never invokes the real binary. Run with: +//! +//! ```text +//! cargo test -p kopiur-kopia --features integration --test integration_stdin +//! ``` +//! +//! These are the tests that pin the CORRECTNESS CLAIM the whole stream-source +//! feature rests on: a producer that fails leaves NO snapshot, even when it had +//! already written every byte. If kopia ever changed so that it committed a +//! manifest before stdin reached EOF, `abort_after_full_payload_leaves_no_snapshot` +//! is what would catch it — and a silent regression there would mean shipping +//! truncated database dumps as successful backups. + +#![cfg(unix)] + +use std::collections::BTreeMap; + +use kopiur_kopia::{ + ConnectSpec, KopiaClient, SnapshotCreateOptions, SnapshotCreateOutcome, SnapshotSource, + StdinOutcome, +}; +use tokio::io::AsyncWriteExt; + +fn isolated_client(config_dir: &std::path::Path) -> KopiaClient { + KopiaClient::builder() + .binary("kopia") + .env("KOPIA_PASSWORD", "test1234") + .env( + "KOPIA_CONFIG_PATH", + config_dir.join("repository.config").display().to_string(), + ) + .env( + "KOPIA_CACHE_DIRECTORY", + config_dir.join("cache").display().to_string(), + ) + .env( + "KOPIA_LOG_DIR", + config_dir.join("logs").display().to_string(), + ) + .env("KOPIA_CHECK_FOR_UPDATES", "false") + .build() +} + +async fn fresh_repo(repo: &std::path::Path, config: &std::path::Path) -> KopiaClient { + let client = isolated_client(config); + client + .repository_create( + &ConnectSpec::Filesystem { + path: repo.to_path_buf(), + }, + Default::default(), + &Default::default(), + ) + .await + .expect("create repository"); + client +} + +/// Commit path: bytes in, one virtual file out, byte-identical on the way back. +#[tokio::test] +#[cfg_attr(not(feature = "integration"), ignore)] +async fn stdin_snapshot_roundtrips_byte_identical() { + let repo_dir = tempfile::tempdir().unwrap(); + let config_dir = tempfile::tempdir().unwrap(); + let client = fresh_repo(repo_dir.path(), config_dir.path()).await; + + // Deliberately larger than any pipe buffer, and not newline-structured, so a + // chunking or text-mangling bug shows up as a hash mismatch. + let payload: Vec = (0..(3 * 1024 * 1024u32)).map(|i| (i % 251) as u8).collect(); + let expected = payload.clone(); + + let outcome = client + .snapshot_create_stdin_outcome_with( + "/stream/dump.sql", + "dump.sql", + &BTreeMap::new(), + Some("tester@host:/stream/dump.sql"), + &SnapshotCreateOptions::default(), + async |stdin: &mut tokio::process::ChildStdin| { + stdin.write_all(&payload).await.expect("write payload"); + Ok(StdinOutcome::Commit) + }, + ) + .await + .expect("stdin snapshot should succeed"); + + let result = match outcome { + SnapshotCreateOutcome::Created(r) => *r, + SnapshotCreateOutcome::Unchanged => panic!("a stdin snapshot is never 'unchanged'"), + }; + // One regular file inside a virtual directory — the shape the restore path relies on. + let root = result + .root_entry + .expect("a stdin snapshot always has a root entry"); + assert_eq!(root.summary.as_ref().map(|s| s.files), Some(1)); + + let root_obj = root.obj.clone(); + let mut got: Vec = Vec::new(); + client + .show_to(&format!("{root_obj}/dump.sql"), &mut got) + .await + .expect("show the stored file"); + assert_eq!(got.len(), expected.len(), "restored length differs"); + assert!( + got == expected, + "restored bytes differ from what was streamed in" + ); +} + +/// THE load-bearing test. The producer writes the COMPLETE payload and only then +/// reports failure. Because kopia's stdin is still open, aborting must leave no +/// manifest at all — "all bytes delivered" must never be the commit point. +#[tokio::test] +#[cfg_attr(not(feature = "integration"), ignore)] +async fn abort_after_full_payload_leaves_no_snapshot() { + let repo_dir = tempfile::tempdir().unwrap(); + let config_dir = tempfile::tempdir().unwrap(); + let client = fresh_repo(repo_dir.path(), config_dir.path()).await; + + let before = client.snapshot_list_all().await.expect("list before").len(); + + let err = client + .snapshot_create_stdin_outcome_with( + "/stream/dump.sql", + "dump.sql", + &BTreeMap::new(), + Some("tester@host:/stream/dump.sql"), + &SnapshotCreateOptions::default(), + async |stdin: &mut tokio::process::ChildStdin| { + stdin + .write_all(b"a complete and perfectly valid looking dump\n") + .await + .expect("write payload"); + // stdin is deliberately NOT closed: the runner owns it and closes it + // only AFTER killing kopia, which is what guarantees kopia never + // reaches EOF and never writes a manifest. + Ok(StdinOutcome::Abort) + }, + ) + .await + .expect_err("an aborted producer must fail the snapshot"); + + assert!( + matches!(err, kopiur_kopia::KopiaError::StdinProducerFailed { .. }), + "expected StdinProducerFailed, got {err:?}" + ); + + let after = client.snapshot_list_all().await.expect("list after"); + assert_eq!( + after.len(), + before, + "aborting the producer must leave NO snapshot manifest behind, found {after:#?}" + ); + assert!( + !after.iter().any(|e| e.source.path == "/stream/dump.sql"), + "no snapshot may exist for the aborted source" + ); +} + +/// A producer whose own error propagates: same guarantee, different arm of the +/// runner (`Err` rather than `Ok(Abort)`). +#[tokio::test] +#[cfg_attr(not(feature = "integration"), ignore)] +async fn producer_error_propagates_and_leaves_no_snapshot() { + let repo_dir = tempfile::tempdir().unwrap(); + let config_dir = tempfile::tempdir().unwrap(); + let client = fresh_repo(repo_dir.path(), config_dir.path()).await; + + let err = client + .snapshot_create_stdin_outcome_with( + "/stream/dump.sql", + "dump.sql", + &BTreeMap::new(), + Some("tester@host:/stream/dump.sql"), + &SnapshotCreateOptions::default(), + async |stdin: &mut tokio::process::ChildStdin| { + stdin.write_all(b"partial").await.expect("write"); + Err(kopiur_kopia::KopiaError::EmptyOutput { + context: "synthetic producer failure".to_string(), + stderr_tail: String::new(), + }) + }, + ) + .await + .expect_err("the producer's own error must fail the snapshot"); + assert!(matches!(err, kopiur_kopia::KopiaError::EmptyOutput { .. })); + + let after = client.snapshot_list_all().await.expect("list after"); + assert!(after.is_empty(), "no snapshot may exist, found {after:#?}"); +} + +/// `--stdin-file` must reach the argv, and the recorded identity must be the +/// operator-resolved one rather than the mover pod's ambient user/host. +#[tokio::test] +#[cfg_attr(not(feature = "integration"), ignore)] +async fn stdin_snapshot_records_the_overridden_identity() { + let repo_dir = tempfile::tempdir().unwrap(); + let config_dir = tempfile::tempdir().unwrap(); + let client = fresh_repo(repo_dir.path(), config_dir.path()).await; + + client + .snapshot_create_stdin_outcome_with( + "/stream/postgres.sql", + "postgres.sql", + &BTreeMap::new(), + Some("bundlecop@k7:/stream/postgres.sql"), + &SnapshotCreateOptions::default(), + async |stdin: &mut tokio::process::ChildStdin| { + stdin.write_all(b"SELECT 1;\n").await.unwrap(); + Ok(StdinOutcome::Commit) + }, + ) + .await + .expect("snapshot"); + + let listed = client + .snapshot_list(Some(&SnapshotSource { + user_name: "bundlecop".into(), + host: "k7".into(), + path: "/stream/postgres.sql".into(), + })) + .await + .expect("list by identity"); + assert_eq!( + listed.len(), + 1, + "expected exactly one snapshot: {listed:#?}" + ); + assert_eq!(listed[0].source.user_name, "bundlecop"); + assert_eq!(listed[0].source.host, "k7"); +} diff --git a/crates/mover/src/status.rs b/crates/mover/src/status.rs index ba62e9d8..e5b12e78 100644 --- a/crates/mover/src/status.rs +++ b/crates/mover/src/status.rs @@ -146,6 +146,10 @@ impl From<&crate::error::MoverError> for FailureBlock { KopiaError::Spawn { .. } | KopiaError::Json { .. } | KopiaError::EmptyOutput { .. } + // kopia was killed on purpose, so its exit code says nothing + // about the failure — the producer's does, and it is in the + // message. + | KopiaError::StdinProducerFailed { .. } | KopiaError::Timeout { .. } => None, }, Some(op.as_str().to_string()), diff --git a/crates/webhook/src/handlers.rs b/crates/webhook/src/handlers.rs index bba71b05..0bd4541e 100644 --- a/crates/webhook/src/handlers.rs +++ b/crates/webhook/src/handlers.rs @@ -701,7 +701,10 @@ async fn handle_restore( let target_pvc = match &spec.target { api::restore::RestoreTarget::PvcRef(r) => Some(r.name.as_str()), api::restore::RestoreTarget::Pvc(t) => Some(t.name.as_str()), - api::restore::RestoreTarget::Populator(_) => None, + // Neither writes a PVC, so there is no destination ownership to warn about. + api::restore::RestoreTarget::Populator(_) | api::restore::RestoreTarget::StreamExec(_) => { + None + } }; let warnings = crate::secctx::restore_warnings( client, diff --git a/deploy/crds/all-crds.yaml b/deploy/crds/all-crds.yaml index 699655d4..6d7f04ab 100644 --- a/deploy/crds/all-crds.yaml +++ b/deploy/crds/all-crds.yaml @@ -8868,7 +8868,9 @@ spec: sources: description: What to back up (at least one source; webhook-enforced). items: - description: A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs` (webhook-enforced). + description: |- + A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs`, `stream` + (webhook-enforced). properties: acknowledgeLiveMutation: description: |- @@ -8883,7 +8885,7 @@ spec: nullable: true type: boolean nfs: - description: An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`. + description: An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`/`stream`. nullable: true properties: path: @@ -8985,10 +8987,95 @@ spec: - PvcNamespacedName nullable: true type: string + stream: + description: |- + Capture a command's standard output as one virtual file — a logical backup + (`pg_dumpall`, `mysqldump`, …) rather than a volume copy. Mutually exclusive + with `pvc`/`pvcSelector`/`nfs`. No volume is mounted; the mover execs the + command in a running workload Pod and streams its stdout straight into kopia. + nullable: true + properties: + fileName: + description: |- + Name of the single virtual file stored in the snapshot (e.g. `postgres.sql`). + + Must be ONE file name: no `/`, and not `.` or `..`. kopia stores this string + verbatim as the entry name and does not sanitize it, so a path-shaped value + would make a later `kopia restore` write OUTSIDE its destination directory. + maxLength: 255 + minLength: 1 + type: string + workloadExec: + description: 'The producer: what to run, and where. Its stdout IS the backup data.' + properties: + command: + description: |- + The argv to execute. NOT a shell line: element 0 is the program, so use + `["sh", "-ec", "..."]` explicitly if you want shell semantics. + + Reference credentials through the container's existing environment or mounted + Secrets — never inline them here. This argv is copied into the mover Pod's + spec, so anyone with `pods:get` in the namespace can read it. + items: + type: string + minItems: 1 + type: array + container: + description: Container to exec in; absent uses the Pod's default container. + nullable: true + type: string + podSelector: + description: |- + Standard label selector identifying the workload Pod, resolved in the + `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + empty selector matches every Pod in the namespace. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. The requirements are ANDed. + items: + description: A label selector requirement is a selector that contains values, a key, and an operator that relates the key and values. + properties: + key: + description: key is the label key that the selector applies to. + type: string + operator: + description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch. + items: + type: string + type: array + required: + - key + - operator + type: object + type: array + matchLabels: + additionalProperties: + type: string + description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + timeout: + description: |- + Go duration bounding the command (e.g. `2h`); absent uses + [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + fails, leaving no snapshot behind. + nullable: true + type: string + required: + - command + - podSelector + type: object + required: + - fileName + - workloadExec + type: object type: object x-kubernetes-validations: - - message: exactly one of pvc, pvcSelector, nfs - rule: '(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) == 1' + - message: exactly one of pvc, pvcSelector, nfs, stream + rule: '(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) + (has(self.stream) ? 1 : 0) == 1' maxItems: 100 type: array staging: @@ -11269,6 +11356,8 @@ spec: - pvcRef - required: - populator + - required: + - streamExec properties: populator: description: 'Passive populator mode: the restore is claimed by a PVC''s `spec.dataSourceRef`.' @@ -11319,6 +11408,86 @@ spec: required: - name type: object + streamExec: + description: |- + Stream one virtual file out of the snapshot into a command's stdin in a + running Pod — the companion of a `stream` backup source (e.g. feeding a + `pg_dumpall` artifact back through `psql`). Writes no PVC. + properties: + fileName: + description: |- + Which virtual file inside the snapshot to read back — the `fileName` the + backup's stream source used (e.g. `postgres.sql`). + maxLength: 255 + minLength: 1 + type: string + workloadExec: + description: 'Where to send it: the command receives the file''s bytes on stdin.' + properties: + command: + description: |- + The argv to execute. NOT a shell line: element 0 is the program, so use + `["sh", "-ec", "..."]` explicitly if you want shell semantics. + + Reference credentials through the container's existing environment or mounted + Secrets — never inline them here. This argv is copied into the mover Pod's + spec, so anyone with `pods:get` in the namespace can read it. + items: + type: string + minItems: 1 + type: array + container: + description: Container to exec in; absent uses the Pod's default container. + nullable: true + type: string + podSelector: + description: |- + Standard label selector identifying the workload Pod, resolved in the + `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + empty selector matches every Pod in the namespace. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. The requirements are ANDed. + items: + description: A label selector requirement is a selector that contains values, a key, and an operator that relates the key and values. + properties: + key: + description: key is the label key that the selector applies to. + type: string + operator: + description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch. + items: + type: string + type: array + required: + - key + - operator + type: object + type: array + matchLabels: + additionalProperties: + type: string + description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + timeout: + description: |- + Go duration bounding the command (e.g. `2h`); absent uses + [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + fails, leaving no snapshot behind. + nullable: true + type: string + required: + - command + - podSelector + type: object + required: + - fileName + - workloadExec + type: object type: object required: - source diff --git a/deploy/crds/restores.yaml b/deploy/crds/restores.yaml index ff7aacd6..a38389d0 100644 --- a/deploy/crds/restores.yaml +++ b/deploy/crds/restores.yaml @@ -682,6 +682,8 @@ spec: - pvcRef - required: - populator + - required: + - streamExec properties: populator: description: 'Passive populator mode: the restore is claimed by a PVC''s `spec.dataSourceRef`.' @@ -732,6 +734,86 @@ spec: required: - name type: object + streamExec: + description: |- + Stream one virtual file out of the snapshot into a command's stdin in a + running Pod — the companion of a `stream` backup source (e.g. feeding a + `pg_dumpall` artifact back through `psql`). Writes no PVC. + properties: + fileName: + description: |- + Which virtual file inside the snapshot to read back — the `fileName` the + backup's stream source used (e.g. `postgres.sql`). + maxLength: 255 + minLength: 1 + type: string + workloadExec: + description: 'Where to send it: the command receives the file''s bytes on stdin.' + properties: + command: + description: |- + The argv to execute. NOT a shell line: element 0 is the program, so use + `["sh", "-ec", "..."]` explicitly if you want shell semantics. + + Reference credentials through the container's existing environment or mounted + Secrets — never inline them here. This argv is copied into the mover Pod's + spec, so anyone with `pods:get` in the namespace can read it. + items: + type: string + minItems: 1 + type: array + container: + description: Container to exec in; absent uses the Pod's default container. + nullable: true + type: string + podSelector: + description: |- + Standard label selector identifying the workload Pod, resolved in the + `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + empty selector matches every Pod in the namespace. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. The requirements are ANDed. + items: + description: A label selector requirement is a selector that contains values, a key, and an operator that relates the key and values. + properties: + key: + description: key is the label key that the selector applies to. + type: string + operator: + description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch. + items: + type: string + type: array + required: + - key + - operator + type: object + type: array + matchLabels: + additionalProperties: + type: string + description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + timeout: + description: |- + Go duration bounding the command (e.g. `2h`); absent uses + [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + fails, leaving no snapshot behind. + nullable: true + type: string + required: + - command + - podSelector + type: object + required: + - fileName + - workloadExec + type: object type: object required: - source diff --git a/deploy/crds/snapshotpolicies.yaml b/deploy/crds/snapshotpolicies.yaml index 384ec19f..afe145d7 100644 --- a/deploy/crds/snapshotpolicies.yaml +++ b/deploy/crds/snapshotpolicies.yaml @@ -992,7 +992,9 @@ spec: sources: description: What to back up (at least one source; webhook-enforced). items: - description: A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs` (webhook-enforced). + description: |- + A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs`, `stream` + (webhook-enforced). properties: acknowledgeLiveMutation: description: |- @@ -1007,7 +1009,7 @@ spec: nullable: true type: boolean nfs: - description: An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`. + description: An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`/`stream`. nullable: true properties: path: @@ -1109,10 +1111,95 @@ spec: - PvcNamespacedName nullable: true type: string + stream: + description: |- + Capture a command's standard output as one virtual file — a logical backup + (`pg_dumpall`, `mysqldump`, …) rather than a volume copy. Mutually exclusive + with `pvc`/`pvcSelector`/`nfs`. No volume is mounted; the mover execs the + command in a running workload Pod and streams its stdout straight into kopia. + nullable: true + properties: + fileName: + description: |- + Name of the single virtual file stored in the snapshot (e.g. `postgres.sql`). + + Must be ONE file name: no `/`, and not `.` or `..`. kopia stores this string + verbatim as the entry name and does not sanitize it, so a path-shaped value + would make a later `kopia restore` write OUTSIDE its destination directory. + maxLength: 255 + minLength: 1 + type: string + workloadExec: + description: 'The producer: what to run, and where. Its stdout IS the backup data.' + properties: + command: + description: |- + The argv to execute. NOT a shell line: element 0 is the program, so use + `["sh", "-ec", "..."]` explicitly if you want shell semantics. + + Reference credentials through the container's existing environment or mounted + Secrets — never inline them here. This argv is copied into the mover Pod's + spec, so anyone with `pods:get` in the namespace can read it. + items: + type: string + minItems: 1 + type: array + container: + description: Container to exec in; absent uses the Pod's default container. + nullable: true + type: string + podSelector: + description: |- + Standard label selector identifying the workload Pod, resolved in the + `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + empty selector matches every Pod in the namespace. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. The requirements are ANDed. + items: + description: A label selector requirement is a selector that contains values, a key, and an operator that relates the key and values. + properties: + key: + description: key is the label key that the selector applies to. + type: string + operator: + description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch. + items: + type: string + type: array + required: + - key + - operator + type: object + type: array + matchLabels: + additionalProperties: + type: string + description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + timeout: + description: |- + Go duration bounding the command (e.g. `2h`); absent uses + [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + fails, leaving no snapshot behind. + nullable: true + type: string + required: + - command + - podSelector + type: object + required: + - fileName + - workloadExec + type: object type: object x-kubernetes-validations: - - message: exactly one of pvc, pvcSelector, nfs - rule: '(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) == 1' + - message: exactly one of pvc, pvcSelector, nfs, stream + rule: '(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) + (has(self.stream) ? 1 : 0) == 1' maxItems: 100 type: array staging: diff --git a/deploy/helm/kopiur/crds/restores.yaml b/deploy/helm/kopiur/crds/restores.yaml index ff7aacd6..a38389d0 100644 --- a/deploy/helm/kopiur/crds/restores.yaml +++ b/deploy/helm/kopiur/crds/restores.yaml @@ -682,6 +682,8 @@ spec: - pvcRef - required: - populator + - required: + - streamExec properties: populator: description: 'Passive populator mode: the restore is claimed by a PVC''s `spec.dataSourceRef`.' @@ -732,6 +734,86 @@ spec: required: - name type: object + streamExec: + description: |- + Stream one virtual file out of the snapshot into a command's stdin in a + running Pod — the companion of a `stream` backup source (e.g. feeding a + `pg_dumpall` artifact back through `psql`). Writes no PVC. + properties: + fileName: + description: |- + Which virtual file inside the snapshot to read back — the `fileName` the + backup's stream source used (e.g. `postgres.sql`). + maxLength: 255 + minLength: 1 + type: string + workloadExec: + description: 'Where to send it: the command receives the file''s bytes on stdin.' + properties: + command: + description: |- + The argv to execute. NOT a shell line: element 0 is the program, so use + `["sh", "-ec", "..."]` explicitly if you want shell semantics. + + Reference credentials through the container's existing environment or mounted + Secrets — never inline them here. This argv is copied into the mover Pod's + spec, so anyone with `pods:get` in the namespace can read it. + items: + type: string + minItems: 1 + type: array + container: + description: Container to exec in; absent uses the Pod's default container. + nullable: true + type: string + podSelector: + description: |- + Standard label selector identifying the workload Pod, resolved in the + `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + empty selector matches every Pod in the namespace. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. The requirements are ANDed. + items: + description: A label selector requirement is a selector that contains values, a key, and an operator that relates the key and values. + properties: + key: + description: key is the label key that the selector applies to. + type: string + operator: + description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch. + items: + type: string + type: array + required: + - key + - operator + type: object + type: array + matchLabels: + additionalProperties: + type: string + description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + timeout: + description: |- + Go duration bounding the command (e.g. `2h`); absent uses + [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + fails, leaving no snapshot behind. + nullable: true + type: string + required: + - command + - podSelector + type: object + required: + - fileName + - workloadExec + type: object type: object required: - source diff --git a/deploy/helm/kopiur/crds/snapshotpolicies.yaml b/deploy/helm/kopiur/crds/snapshotpolicies.yaml index 384ec19f..afe145d7 100644 --- a/deploy/helm/kopiur/crds/snapshotpolicies.yaml +++ b/deploy/helm/kopiur/crds/snapshotpolicies.yaml @@ -992,7 +992,9 @@ spec: sources: description: What to back up (at least one source; webhook-enforced). items: - description: A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs` (webhook-enforced). + description: |- + A single backup source; exactly one of `pvc`, `pvcSelector`, `nfs`, `stream` + (webhook-enforced). properties: acknowledgeLiveMutation: description: |- @@ -1007,7 +1009,7 @@ spec: nullable: true type: boolean nfs: - description: An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`. + description: An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`/`stream`. nullable: true properties: path: @@ -1109,10 +1111,95 @@ spec: - PvcNamespacedName nullable: true type: string + stream: + description: |- + Capture a command's standard output as one virtual file — a logical backup + (`pg_dumpall`, `mysqldump`, …) rather than a volume copy. Mutually exclusive + with `pvc`/`pvcSelector`/`nfs`. No volume is mounted; the mover execs the + command in a running workload Pod and streams its stdout straight into kopia. + nullable: true + properties: + fileName: + description: |- + Name of the single virtual file stored in the snapshot (e.g. `postgres.sql`). + + Must be ONE file name: no `/`, and not `.` or `..`. kopia stores this string + verbatim as the entry name and does not sanitize it, so a path-shaped value + would make a later `kopia restore` write OUTSIDE its destination directory. + maxLength: 255 + minLength: 1 + type: string + workloadExec: + description: 'The producer: what to run, and where. Its stdout IS the backup data.' + properties: + command: + description: |- + The argv to execute. NOT a shell line: element 0 is the program, so use + `["sh", "-ec", "..."]` explicitly if you want shell semantics. + + Reference credentials through the container's existing environment or mounted + Secrets — never inline them here. This argv is copied into the mover Pod's + spec, so anyone with `pods:get` in the namespace can read it. + items: + type: string + minItems: 1 + type: array + container: + description: Container to exec in; absent uses the Pod's default container. + nullable: true + type: string + podSelector: + description: |- + Standard label selector identifying the workload Pod, resolved in the + `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an + empty selector matches every Pod in the namespace. + properties: + matchExpressions: + description: matchExpressions is a list of label selector requirements. The requirements are ANDed. + items: + description: A label selector requirement is a selector that contains values, a key, and an operator that relates the key and values. + properties: + key: + description: key is the label key that the selector applies to. + type: string + operator: + description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist. + type: string + values: + description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch. + items: + type: string + type: array + required: + - key + - operator + type: object + type: array + matchLabels: + additionalProperties: + type: string + description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed. + type: object + type: object + timeout: + description: |- + Go duration bounding the command (e.g. `2h`); absent uses + [`DEFAULT_STREAM_TIMEOUT`]. On expiry the command is abandoned and the run + fails, leaving no snapshot behind. + nullable: true + type: string + required: + - command + - podSelector + type: object + required: + - fileName + - workloadExec + type: object type: object x-kubernetes-validations: - - message: exactly one of pvc, pvcSelector, nfs - rule: '(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) == 1' + - message: exactly one of pvc, pvcSelector, nfs, stream + rule: '(has(self.pvc) ? 1 : 0) + (has(self.pvcSelector) ? 1 : 0) + (has(self.nfs) ? 1 : 0) + (has(self.stream) ? 1 : 0) == 1' maxItems: 100 type: array staging: diff --git a/docs/field-reference.md b/docs/field-reference.md index ef7e08a7..bf8fe82b 100644 --- a/docs/field-reference.md +++ b/docs/field-reference.md @@ -2404,17 +2404,18 @@ Externally tagged — set **exactly one** of: `pvcConsumer` · `snapshot` · `wo #### `spec.sources[]` { #snapshotpolicy-spec-sources } -Externally tagged — set **exactly one** of: `nfs` · `pvc` · `pvcSelector`. +Externally tagged — set **exactly one** of: `nfs` · `pvc` · `pvcSelector` · `stream`. | Field | Type | Default | Description | | --- | --- | --- | --- | | `acknowledgeLiveMutation` | boolean | — | Acknowledges that `copyMethod: Direct` + `readOnly: false` lets the kubelet recursively `chgrp` the **live** volume to the mover's `fsGroup` and make it group-writable — permanently, while the workload is running. Required for that combination alone.
Ignored (not rejected) otherwise: it is an acknowledgement, never harmful to carry, and rejecting a stale one would make switching `copyMethod` between `Direct` and `Snapshot`/`Clone` a two-step edit in both directions. | -| `nfs` | [object](#snapshotpolicy-spec-sources-nfs) | — | An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`. | +| `nfs` | [object](#snapshotpolicy-spec-sources-nfs) | — | An inline NFS export to back up directly. Mutually exclusive with `pvc`/`pvcSelector`/`stream`. | | `pvc` | [object](#snapshotpolicy-spec-sources-pvc) | — | Single PVC by name. Mutually exclusive with `pvcSelector`/`nfs`. | | `pvcSelector` | [object](#snapshotpolicy-spec-sources-pvcselector) | — | Label/namespace selector matching many PVCs. Mutually exclusive with `pvc`/`nfs`. | | `readOnly` | boolean | `true` | Mount the source read-only (default `true`; kopia only ever reads it).
Set `false` **only** to make `fsGroup` work on the source. The kubelet applies `fsGroup` by recursively `chgrp`-ing the volume and adding group-write — and it skips that walk entirely on a read-only mount, which is why a mover `fsGroup`/`fsGroupChangePolicy` otherwise has no effect here. Under `copyMethod: Snapshot`/`Clone` the walk rewrites the throwaway staged PVC and never touches your data. Under `copyMethod: Direct` it rewrites the LIVE volume, which requires `acknowledgeLiveMutation`.
Not supported on an `nfs` source: the kubelet does not apply `fsGroup` to in-tree NFS volumes at all, so a read-write mount would grant nothing. | | `sourcePathOverride` | string | —
maxLength 4096 | What kopia records as the source path (default `/pvc/<name>`, or the NFS export `path`). | | `sourcePathStrategy` | enum: PvcName \| PvcNamespacedName | `PvcName` | How a selector-matched PVC's source path is derived. Only relevant for `pvcSelector` sources, where one recipe expands to many PVCs and each needs a distinct kopia source path. Defaults to `PvcName`. | +| `stream` | [object](#snapshotpolicy-spec-sources-stream) | — | Capture a command's standard output as one virtual file — a logical backup (`pg_dumpall`, `mysqldump`, …) rather than a volume copy. Mutually exclusive with `pvc`/`pvcSelector`/`nfs`. No volume is mounted; the mover execs the command in a running workload Pod and streams its stdout straight into kopia. | ##### `spec.sources[].nfs` { #snapshotpolicy-spec-sources-nfs } @@ -2442,6 +2443,22 @@ Externally tagged — set **exactly one** of: `nfs` · `pvc` · `pvcSelector`. | --- | --- | --- | --- | | `matchNames` | []string | — | Exact namespace names to search; empty means the policy's own namespace. | +##### `spec.sources[].stream` { #snapshotpolicy-spec-sources-stream } + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `fileName` | string | **required**
minLength 1; maxLength 255 | Name of the single virtual file stored in the snapshot (e.g. `postgres.sql`).
Must be ONE file name: no `/`, and not `.` or `..`. kopia stores this string verbatim as the entry name and does not sanitize it, so a path-shaped value would make a later `kopia restore` write OUTSIDE its destination directory. | +| `workloadExec` | [object](#snapshotpolicy-spec-sources-stream-workloadexec) | **required** | The producer: what to run, and where. Its stdout IS the backup data. | + +###### `spec.sources[].stream.workloadExec` { #snapshotpolicy-spec-sources-stream-workloadexec } + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `command` | []string | **required**
minItems 1 | The argv to execute. NOT a shell line: element 0 is the program, so use `["sh", "-ec", "..."]` explicitly if you want shell semantics.
Reference credentials through the container's existing environment or mounted Secrets — never inline them here. This argv is copied into the mover Pod's spec, so anyone with `pods:get` in the namespace can read it. | +| `podSelector` | core/v1 LabelSelector | **required** | Standard label selector identifying the workload Pod, resolved in the `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an empty selector matches every Pod in the namespace. | +| `container` | string | — | Container to exec in; absent uses the Pod's default container. | +| `timeout` | string | — | Go duration bounding the command (e.g. `2h`); absent uses `DEFAULT_STREAM_TIMEOUT`. On expiry the command is abandoned and the run fails, leaving no snapshot behind. | + #### `spec.staging` { #snapshotpolicy-spec-staging } | Field | Type | Default | Description | @@ -3025,13 +3042,14 @@ Externally tagged — set **exactly one** of: `fromPolicy` · `identity` · `sna #### `spec.target` { #restore-spec-target } -Externally tagged — set **exactly one** of: `populator` · `pvc` · `pvcRef`. +Externally tagged — set **exactly one** of: `populator` · `pvc` · `pvcRef` · `streamExec`. | Field | Type | Default | Description | | --- | --- | --- | --- | | `populator` | object | — | Passive populator mode: the restore is claimed by a PVC's `spec.dataSourceRef`. | | `pvc` | [object](#restore-spec-target-pvc) | — | Operator creates the PVC. | | `pvcRef` | [object](#restore-spec-target-pvcref) | — | Write into an existing PVC. | +| `streamExec` | [object](#restore-spec-target-streamexec) | — | Stream one virtual file out of the snapshot into a command's stdin in a running Pod — the companion of a `stream` backup source (e.g. feeding a `pg_dumpall` artifact back through `psql`). Writes no PVC. | ##### `spec.target.pvc` { #restore-spec-target-pvc } @@ -3049,6 +3067,22 @@ Externally tagged — set **exactly one** of: `populator` · `pvc` · `pvcRef`. | `name` | string | **required** | Name of the referenced object. | | `namespace` | string | — | Namespace of the referenced object; absent = same namespace as the referrer. | +##### `spec.target.streamExec` { #restore-spec-target-streamexec } + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `fileName` | string | **required**
minLength 1; maxLength 255 | Which virtual file inside the snapshot to read back — the `fileName` the backup's stream source used (e.g. `postgres.sql`). | +| `workloadExec` | [object](#restore-spec-target-streamexec-workloadexec) | **required** | Where to send it: the command receives the file's bytes on stdin. | + +###### `spec.target.streamExec.workloadExec` { #restore-spec-target-streamexec-workloadexec } + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `command` | []string | **required**
minItems 1 | The argv to execute. NOT a shell line: element 0 is the program, so use `["sh", "-ec", "..."]` explicitly if you want shell semantics.
Reference credentials through the container's existing environment or mounted Secrets — never inline them here. This argv is copied into the mover Pod's spec, so anyone with `pods:get` in the namespace can read it. | +| `podSelector` | core/v1 LabelSelector | **required** | Standard label selector identifying the workload Pod, resolved in the `SnapshotPolicy`'s (or `Restore`'s) own namespace. Must not be empty — an empty selector matches every Pod in the namespace. | +| `container` | string | — | Container to exec in; absent uses the Pod's default container. | +| `timeout` | string | — | Go duration bounding the command (e.g. `2h`); absent uses `DEFAULT_STREAM_TIMEOUT`. On expiry the command is abandoned and the run fails, leaving no snapshot behind. | + #### `spec.credentialProjection` { #restore-spec-credentialprojection } | Field | Type | Default | Description | From 01e1da3fa41e27ee6b91705b16f4a0f3b5446196 Mon Sep 17 00:00:00 2001 From: Michael Elsdorfer Date: Mon, 7 Sep 2026 19:14:52 +0100 Subject: [PATCH 03/59] feat(mover): exec a workload command and stream its stdout into kopia MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mover resolves the one running pod matching the selector, opens a pods/exec websocket, and pumps its stdout straight into kopia's stdin. The controller turns a `stream` source into a StreamProducerSpec on the work spec; build_backup_run is now an exhaustive match over SourceShape, and the stream arm mounts no volume at all. Correctness rests on the exec STATUS, never on the byte stream. If the websocket dies mid-dump the stdout reader returns a perfectly clean EOF — indistinguishable from a finished command — so committing on EOF would silently store a truncated database dump as a successful backup. Only `Success` commits; a non-zero exit, a closed status channel (NoStatus), a timeout, or an IO error all abort, which kills kopia with stdin still open so no manifest is ever written. Buffers are raised to 4 MiB: kube-rs defaults its exec duplex pipes to 1 KiB (remote_command.rs MAX_BUF_SIZE), which would move a multi-gigabyte dump one kilobyte per wakeup. Backpressure still works end to end without extra machinery — a full duplex blocks the message loop, which stops reading the socket, which TCP-backpressures the kubelet, which blocks the producer's write. The dump bytes are only ever copied pipe-to-pipe: never collected into a String, logged, attached to an error, or written to a status. Only stderr is captured, capped at 8 KiB. The restore direction mirrors this and additionally requests no stdout channel at all, so restored data has no path back into the mover. Zero, several, unready, or terminating selector matches are each a distinct named failure rather than an arbitrary pick: a backup that quietly dumped a different replica than intended is worse than one that stops and says why. (cherry picked from commit aedc81b37878cca6d7df26c41a48fd4a318c221f) --- crates/controller/src/restore/mod.rs | 1 + crates/controller/src/restore/tests.rs | 1 + crates/controller/src/snapshot/build.rs | 131 +++--- crates/controller/src/snapshot/tests.rs | 1 + crates/mover/Cargo.toml | 4 +- crates/mover/README.md | 1 + crates/mover/src/error.rs | 24 ++ crates/mover/src/jobs.rs | 1 + crates/mover/src/lib.rs | 1 + crates/mover/src/main.rs | 151 ++++++- crates/mover/src/status.rs | 4 + crates/mover/src/stream.rs | 506 ++++++++++++++++++++++++ crates/mover/src/stream/tests.rs | 184 +++++++++ crates/mover/src/workspec/mod.rs | 104 +++++ crates/mover/src/workspec/tests.rs | 6 + 15 files changed, 1065 insertions(+), 55 deletions(-) create mode 100644 crates/mover/src/stream.rs create mode 100644 crates/mover/src/stream/tests.rs diff --git a/crates/controller/src/restore/mod.rs b/crates/controller/src/restore/mod.rs index 224a8f4c..0881208b 100644 --- a/crates/controller/src/restore/mod.rs +++ b/crates/controller/src/restore/mod.rs @@ -3567,6 +3567,7 @@ async fn run_restore_mover( let work_spec = MoverWorkSpec { version: 2, operation: Operation::Restore(RestoreOp { + stdout: None, source: selection.clone(), target_path: target_path.clone(), anchor, diff --git a/crates/controller/src/restore/tests.rs b/crates/controller/src/restore/tests.rs index 058b8c39..b9b92b43 100644 --- a/crates/controller/src/restore/tests.rs +++ b/crates/controller/src/restore/tests.rs @@ -2288,6 +2288,7 @@ fn restore_flags_enable_file_deletion_regression() { // End-to-end through the mover's RestoreOp -> kopia client RestoreOptions -> // argv, proving the whole chain (not just this one hop). let op = RestoreOp { + stdout: None, source: RestoreSelection::Snapshot("s".into()), target_path: "/data".into(), anchor: Default::default(), diff --git a/crates/controller/src/snapshot/build.rs b/crates/controller/src/snapshot/build.rs index 4142bf77..8f4bf273 100644 --- a/crates/controller/src/snapshot/build.rs +++ b/crates/controller/src/snapshot/build.rs @@ -61,55 +61,96 @@ pub(super) fn build_backup_run( // `/pvc//` under `sourcePathStrategy: PvcNamespacedName` for a // selector source); NFS: the export path by default; either overridable by // `sourcePathOverride`. - let (source_path, source_volume) = match (&eff.pvc, &source.nfs) { - (Some(pvc), None) => { - let path = eff - .kopia_source_path(strategy) - .unwrap_or_else(|| format!("/pvc/{}", pvc.name)); - ( - path.clone(), - VolumeMountSpec::pvc(pvc.name.clone(), path, read_only), - ) - } - (None, Some(nfs)) => { - // The export's server-side path (`nfs.path`) is what the volume is - // mounted FROM; it is NOT necessarily a valid in-container mount - // path. An NFSv4 pseudo-root export ("/") must not be mounted at "/" - // in the container — that mounts over the rootfs and the pod fails to - // start. Remap a "/" target to a safe path; kopia snapshots there. - let requested = source + // Exhaustive over the source form (no `_ =>`): a new source kind cannot compile + // until it decides what path kopia records and what — if anything — is mounted. + let (source_path, source_volume, stdin) = match kopiur_api::source_shape(source) + .map_err(|e| Error::Validation(e.to_string()))? + { + // A stream source mounts NOTHING. `source_path` is the virtual root + // (`/stream/` unless overridden) that kopia records the artifact + // under, and the bytes arrive on the mover's stdin from a pod exec. + kopiur_api::SourceShape::Stream(stream) => { + let path = source .source_path_override .clone() - .unwrap_or_else(|| nfs.path.clone()); - let mount_path = if requested == "/" { - crate::consts::NFS_SOURCE_MOUNT_PATH.to_string() - } else { - requested - }; + .unwrap_or_else(|| kopiur_api::stream_source_path(stream)); + let selector = io::label_selector_to_string(&stream.workload_exec.pod_selector); ( - mount_path.clone(), - VolumeMountSpec::nfs(nfs.server.clone(), nfs.path.clone(), mount_path, read_only), + path, + None, + Some(kopiur_mover::workspec::StreamProducerSpec { + namespace: namespace.to_string(), + pod_selector: selector, + container: stream.workload_exec.container.clone(), + command: stream.workload_exec.command.clone(), + file_name: stream.file_name.clone(), + timeout_seconds: stream + .workload_exec + .timeout + .as_deref() + .and_then(kopiur_api::duration::parse_go_duration) + .map(|d| d.as_secs()) + .unwrap_or(kopiur_api::snapshot_policy::DEFAULT_STREAM_TIMEOUT_SECS), + }), ) } - // A `pvcSelector` source reaches here ONLY when this Snapshot carries no - // `spec.source` pin — i.e. someone applied a bare `policyRef` Snapshot - // against a selector policy, or edited a `pvc:` policy into a selector - // after its Snapshots existed (which admission never sees). Refuse with - // the fix rather than the old `invariant violated … please report it`, - // which is what #346 actually looked like to users. - (None, None) if source.pvc_selector.is_some() => { - return Err(Error::Validation(format!( - "SnapshotPolicy `{}` source #{} is a `pvcSelector`, which expands to one Snapshot per matched PVC — but this Snapshot carries no `spec.source` saying which PVC it covers, so there is nothing unambiguous to back up. Create it with `kubectl kopiur snapshot now --policy {0}` or let a SnapshotSchedule fire it; both expand the selector for you.", - config.name_any(), - eff.index, - ))); - } - // Genuinely mutually-exclusive/empty shapes, which admission does reject. - _ => { - return Err(Error::Invariant( - "backup mover path requires exactly one of source.pvc or source.nfs".into(), - )); - } + _ => match (&eff.pvc, &source.nfs) { + (Some(pvc), None) => { + let path = eff + .kopia_source_path(strategy) + .unwrap_or_else(|| format!("/pvc/{}", pvc.name)); + ( + path.clone(), + Some(VolumeMountSpec::pvc(pvc.name.clone(), path, read_only)), + None, + ) + } + (None, Some(nfs)) => { + // The export's server-side path (`nfs.path`) is what the volume is + // mounted FROM; it is NOT necessarily a valid in-container mount + // path. An NFSv4 pseudo-root export ("/") must not be mounted at "/" + // in the container — that mounts over the rootfs and the pod fails to + // start. Remap a "/" target to a safe path; kopia snapshots there. + let requested = source + .source_path_override + .clone() + .unwrap_or_else(|| nfs.path.clone()); + let mount_path = if requested == "/" { + crate::consts::NFS_SOURCE_MOUNT_PATH.to_string() + } else { + requested + }; + ( + mount_path.clone(), + Some(VolumeMountSpec::nfs( + nfs.server.clone(), + nfs.path.clone(), + mount_path, + read_only, + )), + None, + ) + } + // A `pvcSelector` source reaches here ONLY when this Snapshot carries no + // `spec.source` pin — i.e. someone applied a bare `policyRef` Snapshot + // against a selector policy, or edited a `pvc:` policy into a selector + // after its Snapshots existed (which admission never sees). Refuse with + // the fix rather than the old `invariant violated … please report it`, + // which is what #346 actually looked like to users. + (None, None) if source.pvc_selector.is_some() => { + return Err(Error::Validation(format!( + "SnapshotPolicy `{}` source #{} is a `pvcSelector`, which expands to one Snapshot per matched PVC — but this Snapshot carries no `spec.source` saying which PVC it covers, so there is nothing unambiguous to back up. Create it with `kubectl kopiur snapshot now --policy {0}` or let a SnapshotSchedule fire it; both expand the selector for you.", + config.name_any(), + eff.index, + ))); + } + // Genuinely mutually-exclusive/empty shapes, which admission does reject. + _ => { + return Err(Error::Invariant( + "backup mover path requires exactly one of source.pvc or source.nfs".into(), + )); + } + }, }; let creds_secrets = io::plain_creds(io::mover_creds_secrets(&repo.backend, &repo.encryption)); @@ -127,6 +168,7 @@ pub(super) fn build_backup_run( version: 1, operation: Operation::Snapshot(SnapshotOp { source_path: source_path.clone(), + stdin, tags: tags_for(backup, config), // Flattened SnapshotPolicy policy knobs → kopia `policy set` flags // (compression / files / errorHandling / upload / extraArgs). Wired so @@ -163,7 +205,6 @@ pub(super) fn build_backup_run( throttle: crate::io::throttle_spec(repo.mover_defaults.as_ref()), }; - let source_volume = Some(source_volume); let repo_volume = io::filesystem_repo_mount_source(&repo.backend).map(|source| VolumeMountSpec { source, diff --git a/crates/controller/src/snapshot/tests.rs b/crates/controller/src/snapshot/tests.rs index 1a588166..909ca9c9 100644 --- a/crates/controller/src/snapshot/tests.rs +++ b/crates/controller/src/snapshot/tests.rs @@ -4371,6 +4371,7 @@ fn sample_backup_work_spec() -> kopiur_mover::workspec::MoverWorkSpec { MoverWorkSpec { version: 1, operation: Operation::Snapshot(SnapshotOp { + stdin: None, source_path: "/data".into(), tags: BTreeMap::new(), policy: Default::default(), diff --git a/crates/mover/Cargo.toml b/crates/mover/Cargo.toml index 84d626bf..1f563d83 100644 --- a/crates/mover/Cargo.toml +++ b/crates/mover/Cargo.toml @@ -12,7 +12,9 @@ authors.workspace = true kopiur-api = { workspace = true } kopiur-kopia = { workspace = true } kopiur-telemetry = { workspace = true } -kube = { workspace = true, features = ["client", "rustls-tls"] } +# `ws` adds the websocket transport `Api::exec` needs: a stream source execs a +# command in a workload pod and pipes its stdout into kopia (crates/mover/src/stream.rs). +kube = { workspace = true, features = ["client", "rustls-tls", "ws"] } rustls = { workspace = true } k8s-openapi = { workspace = true, features = ["v1_33"] } serde = { workspace = true } diff --git a/crates/mover/README.md b/crates/mover/README.md index 79fb16b4..6aaea124 100644 --- a/crates/mover/README.md +++ b/crates/mover/README.md @@ -67,6 +67,7 @@ use kopiur_mover::workspec::*; let spec = MoverWorkSpec { version: 1, operation: Operation::Snapshot(SnapshotOp { + stdin: None, source_path: "/data".into(), tags: BTreeMap::new(), policy: Default::default(), diff --git a/crates/mover/src/error.rs b/crates/mover/src/error.rs index 0b9efc18..a02a89b2 100644 --- a/crates/mover/src/error.rs +++ b/crates/mover/src/error.rs @@ -366,6 +366,26 @@ pub enum MoverError { source: kopiur_api::ValidationError, }, + /// The single workload pod a stream source must exec into could not be + /// resolved: nothing matched, several matched, or none was running. + /// + /// Its own variant rather than a generic kube error because the fix is always + /// in the user's selector or workload, never in the operator — and because + /// "which pod" is exactly the question an operator asks first. + #[error("{detail}")] + StreamPodResolve { + /// What the selector matched and what to do about it. + detail: String, + }, + + /// A stream producer/consumer exec failed, timed out, or lost its connection. + /// Carries only bounded stderr — never the streamed data. + #[error("{detail}")] + StreamExecFailed { + /// What went wrong, already free of any streamed bytes. + detail: String, + }, + /// A kube client could not be built (the side-channel status PATCHes need /// in-cluster ServiceAccount credentials). #[error( @@ -596,6 +616,10 @@ impl MoverError { | MoverError::RestoreNoSnapshot { .. } | MoverError::RestoreAsOfInvalid { .. } | MoverError::ScratchNotWritable { .. } + // A failed dump command / unresolvable pod is the user's config, not a + // transient fault: re-running the same Job re-runs the same command. + | MoverError::StreamPodResolve { .. } + | MoverError::StreamExecFailed { .. } | MoverError::SuccessExprFalse { .. } | MoverError::SuccessExprEval { .. } | MoverError::KubeClient { .. } diff --git a/crates/mover/src/jobs.rs b/crates/mover/src/jobs.rs index fd0f393b..cacb06b8 100644 --- a/crates/mover/src/jobs.rs +++ b/crates/mover/src/jobs.rs @@ -911,6 +911,7 @@ mod tests { MoverWorkSpec { version: 1, operation: Operation::Snapshot(SnapshotOp { + stdin: None, source_path: "/data".into(), tags: BTreeMap::new(), policy: Default::default(), diff --git a/crates/mover/src/lib.rs b/crates/mover/src/lib.rs index 07a7a3bd..a7527cd0 100644 --- a/crates/mover/src/lib.rs +++ b/crates/mover/src/lib.rs @@ -12,4 +12,5 @@ pub mod repo_meta; pub mod resolve; pub mod serve; pub mod status; +pub mod stream; pub mod workspec; diff --git a/crates/mover/src/main.rs b/crates/mover/src/main.rs index bbab8cd3..deaaad5e 100644 --- a/crates/mover/src/main.rs +++ b/crates/mover/src/main.rs @@ -589,15 +589,65 @@ async fn run_operation( // real duration — kopia hashes the whole tree even when it decides // not to write a manifest, so this is not free and is worth showing. let started_at = chrono::Utc::now(); - let outcome = client - .snapshot_create_outcome_with( - &op.source_path, - &op.tags, - Some(&override_source), - &op.create_options(), - ) - .await - .map_err(kopia(KopiaOp::SnapshotCreate))?; + // Exhaustive: where the bytes come from decides the whole execution. + // A filesystem run walks the mounted path; a stream run execs a command + // in a workload pod and pipes its stdout in. A new input kind cannot + // compile until it is given one. + let outcome = match kopiur_mover::workspec::snapshot_input(op) { + kopiur_mover::workspec::SnapshotInput::Filesystem => client + .snapshot_create_outcome_with( + &op.source_path, + &op.tags, + Some(&override_source), + &op.create_options(), + ) + .await + .map_err(kopia(KopiaOp::SnapshotCreate))?, + kopiur_mover::workspec::SnapshotInput::Stream(producer) => { + let kube_client = + kube::Client::try_default() + .await + .map_err(|e| MoverError::KubeClient { + source: Box::new(e), + })?; + // `failure` is how the producer's verdict escapes the closure: + // the kopia runner only learns Commit/Abort, and the actionable + // text has to reach the status. It holds ONLY the verdict and a + // bounded stderr tail — never the dumped bytes. + let mut failure: Option = None; + let res = client + .snapshot_create_stdin_outcome_with( + &op.source_path, + &producer.file_name, + &op.tags, + Some(&override_source), + &op.create_options(), + async |stdin: &mut tokio::process::ChildStdin| { + Ok(kopiur_mover::stream::feed_from_pod( + &kube_client, + producer, + stdin, + &mut failure, + ) + .await) + }, + ) + .await; + match res { + Ok(o) => o, + Err(e) => { + // A producer failure is the user's dump command, not + // kopia: report it as such so the status names the right + // thing to fix. kopia aborted before writing a manifest, + // so nothing partial survives either way. + return Err(match failure { + Some(detail) => MoverError::StreamExecFailed { detail }, + None => kopia(KopiaOp::SnapshotCreate)(e), + }); + } + } + } + }; // Exhaustive: a deduped run and a real one are BOTH successes but are // not interchangeable, and the difference is invisible in the happy // path. Matching here is what stops a run that owns no manifest from @@ -647,6 +697,13 @@ async fn run_operation( Ok(StatusUpdate::succeeded_backup(&result, chrono::Utc::now())) } Operation::Restore(op) => { + // Where the bytes GO is decided first: a stream restore never touches a + // filesystem, so it does not share the mounted-target machinery below. + if let kopiur_mover::workspec::RestoreOutput::Stream(consumer) = + kopiur_mover::workspec::restore_output(op) + { + return restore_stream(client, op, consumer).await; + } // Exactly one source kind (externally tagged): a controller-resolved id, // or an in-Job selector to resolve here. Exhaustive — a new variant // can't compile until handled. @@ -755,6 +812,82 @@ async fn run_operation( /// `NotFound`, re-resolve the live id from the snapshot's stable anchors /// ([`RestoreOp::anchor`]) and retry once. Returns the id actually restored (for /// `status.logTail`). +/// Restore ONE virtual file out of a snapshot straight into a command's stdin. +/// +/// Resolves the snapshot to its ROOT OBJECT id (not the manifest id — kopia's +/// `/` sub-path form only accepts the former; a manifest id fails with +/// "parent is not a directory") and streams `kopia show /` into the +/// exec. Both halves must succeed: a `psql` that died halfway leaves a half-loaded +/// database, and calling that a completed restore would be worse than failing. +async fn restore_stream( + client: &KopiaClient, + op: &RestoreOp, + consumer: &kopiur_mover::workspec::StreamConsumerSpec, +) -> Result { + // Only a controller-resolved id is supported: the deferred `Resolve` path pins + // its choice through the StatusReporter, which this path does not carry. The + // controller resolves a streamExec restore to a concrete snapshot before the Job. + let snapshot_id = match &op.source { + RestoreSelection::Snapshot(id) => id.clone(), + RestoreSelection::Resolve(sel) => { + let filter = SnapshotSource { + host: sel.hostname.clone(), + user_name: sel.username.clone(), + path: sel.source_path.clone().unwrap_or_default(), + }; + let mut list = client + .snapshot_list(Some(&filter)) + .await + .map_err(|source| MoverError::Kopia { + op: KopiaOp::RestoreSnapshotList, + source, + })?; + list.sort_by_key(|e| std::cmp::Reverse(e.end_time)); + let cutoff = sel + .as_of + .as_deref() + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) + .map(|t| t.with_timezone(&chrono::Utc)); + let candidates = filter_as_of(list, cutoff); + pick_offset(candidates, sel.offset) + .ok_or_else(|| MoverError::RestoreNoSnapshot { + identity: filter.identity(), + })? + .id + } + }; + + // The sub-path form needs the ROOT ENTRY object id, so look the snapshot up. + let listed = client + .snapshot_list_all() + .await + .map_err(|source| MoverError::Kopia { + op: KopiaOp::RestoreSnapshotList, + source, + })?; + let root_obj = listed + .iter() + .find(|e| e.id == snapshot_id) + .and_then(|e| e.root_entry.as_ref()) + .map(|r| r.obj.clone()) + .ok_or_else(|| MoverError::StreamExecFailed { + detail: format!( + "snapshot `{snapshot_id}` could not be resolved to a root object, so the file \ + `{}` inside it cannot be addressed", + consumer.file_name + ), + })?; + let object_id = format!("{root_obj}/{}", consumer.file_name); + + let kube_client = kube::Client::try_default() + .await + .map_err(|e| MoverError::KubeClient { + source: Box::new(e), + })?; + kopiur_mover::stream::restore_into_pod(&kube_client, client, consumer, &object_id).await?; + Ok(StatusUpdate::completed(&snapshot_id, chrono::Utc::now())) +} + async fn restore_with_heal( client: &KopiaClient, op: &RestoreOp, diff --git a/crates/mover/src/status.rs b/crates/mover/src/status.rs index e5b12e78..a0c76bfc 100644 --- a/crates/mover/src/status.rs +++ b/crates/mover/src/status.rs @@ -168,6 +168,10 @@ impl From<&crate::error::MoverError> for FailureBlock { | MoverError::RestoreNoSnapshot { .. } | MoverError::RestoreAsOfInvalid { .. } | MoverError::ScratchNotWritable { .. } + // A failed dump command / unresolvable pod is the user's config, not a + // transient fault: re-running the same Job re-runs the same command. + | MoverError::StreamPodResolve { .. } + | MoverError::StreamExecFailed { .. } | MoverError::SuccessExprFalse { .. } | MoverError::SuccessExprEval { .. } | MoverError::KubeClient { .. } diff --git a/crates/mover/src/stream.rs b/crates/mover/src/stream.rs new file mode 100644 index 00000000..88bb9913 --- /dev/null +++ b/crates/mover/src/stream.rs @@ -0,0 +1,506 @@ +//! Streamed command sources: exec a command in a running workload pod and move its +//! bytes to or from kopia without ever touching a filesystem. +//! +//! # Where the correctness lives +//! +//! A backup here has TWO ways to fail that a filesystem backup does not: the +//! producer can fail after writing bytes, and the exec connection can drop +//! mid-transfer. Both must be impossible to mistake for success, because kopia +//! cannot tell the difference on its own — from its side, a dropped connection and +//! a clean finish both look like EOF on stdin. +//! +//! The guard is the exec STATUS, not the byte stream: +//! +//! * `Some(status == "Success")` — the command exited 0. Only then do we let kopia +//! see EOF and commit ([`kopiur_kopia::StdinOutcome::Commit`]). +//! * `Some(other)` — the command failed. Abort. +//! * `None` — the status channel closed without a verdict, i.e. the websocket died. +//! The stdout reader will have returned a perfectly clean EOF, so this is the ONLY +//! signal distinguishing "finished" from "connection dropped mid-dump". Abort. +//! * timeout / IO error — abort. +//! +//! Aborting kills kopia with its stdin still open, so no manifest is ever written +//! (see [`kopiur_kopia::KopiaClient::snapshot_create_stdin_outcome_with`]). +//! +//! # Where the data does NOT go +//! +//! The producer's stdout is a database dump. It is copied pipe-to-pipe and is never +//! collected into a `String`, logged, attached to an error, or written to a CR +//! status. Only stderr is captured, and only a bounded tail of it. + +use std::time::Duration; + +use k8s_openapi::api::core::v1::Pod; +use kube::api::{AttachParams, ListParams}; +use kube::{Api, ResourceExt}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use crate::error::{MoverError, Result}; +use crate::workspec::{StreamConsumerSpec, StreamProducerSpec}; +use kopiur_kopia::StdinOutcome; + +/// Bytes of exec stderr kept for diagnostics. Bounded because the point of the cap +/// is that an operator sees WHY a dump failed without the failure message becoming +/// unbounded — and because stderr is the one stream from a database command that +/// could plausibly echo data. +pub const EXEC_STDERR_CAP: usize = 8 * 1024; + +/// Buffer for the exec stdout/stdin duplex pipes. +/// +/// kube-rs defaults these to 1 KiB (`remote_command.rs: MAX_BUF_SIZE`), which would +/// make a multi-gigabyte logical dump crawl — one wakeup per kilobyte. 4 MiB keeps +/// the websocket saturated. Backpressure still works end to end: a full duplex +/// blocks the message loop, which stops reading the socket, which TCP-backpressures +/// the kubelet, which blocks the producer's `write(2)`. +pub const EXEC_STREAM_BUF: usize = 4 * 1024 * 1024; + +/// Exec stderr never exceeds a line or two of diagnostics; keep its buffer small. +pub const EXEC_STDERR_BUF: usize = 64 * 1024; + +/// Pick the one pod a stream source may exec into. +/// +/// Requires EXACTLY ONE `Running`, non-terminating match. Zero, several, or +/// only-not-running are each a distinct named failure rather than an arbitrary +/// pick: a backup that silently dumped a different replica than the operator +/// intended is worse than a backup that stops and says why. +/// +/// Pure over the listed pods so every message is unit-testable without a cluster. +pub fn pick_stream_pod<'a>( + pods: &'a [Pod], + selector: &str, + namespace: &str, +) -> std::result::Result<&'a Pod, String> { + if pods.is_empty() { + return Err(format!( + "no pod matches podSelector `{selector}` in namespace `{namespace}`. The workload \ + must be running for a stream source to dump from it — scale it up, or fix the \ + selector" + )); + } + let live: Vec<&Pod> = pods + .iter() + .filter(|p| pod_is_running(p) && p.metadata.deletion_timestamp.is_none()) + .collect(); + match live.as_slice() { + [] => Err(format!( + "podSelector `{selector}` matched {} pod(s) in namespace `{namespace}`, but none is \ + Running and not terminating. A stream source needs a live container to exec into; \ + wait for the workload to become ready, or fix the selector", + pods.len() + )), + [one] => Ok(one), + many => Err(format!( + "podSelector `{selector}` matched {} RUNNING pods in namespace `{namespace}` ({}). A \ + stream source must identify exactly one pod — dumping an arbitrary replica would \ + make the backup's contents depend on scheduling. Narrow the selector (e.g. add a \ + role/primary label) so it matches only the pod you mean to dump", + many.len(), + many.iter() + .map(|p| p.name_any()) + .collect::>() + .join(", ") + )), + } +} + +/// Whether a pod is in the `Running` phase. +fn pod_is_running(p: &Pod) -> bool { + p.status + .as_ref() + .and_then(|s| s.phase.as_deref()) + .is_some_and(|ph| ph == "Running") +} + +/// The verdict an exec ended with, derived from the websocket status channel. +/// +/// `None` from `take_status()` is deliberately its own arm rather than folded into +/// failure-with-no-message: it is the connection-dropped case, and telling an +/// operator "the exec connection closed without reporting an exit status" instead of +/// "the command failed" is the difference between suspecting the network and +/// suspecting their dump command. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ExecVerdict { + /// The command exited 0. + Success, + /// The command exited non-zero (or the API server reported a failure). + Failed(String), + /// The status channel closed without a verdict — the connection dropped. + NoStatus, +} + +/// Turn a websocket exec status into an [`ExecVerdict`]. Pure, so the mapping is +/// unit-testable without a cluster. +pub fn verdict_of( + status: Option, +) -> ExecVerdict { + match status { + Some(s) if s.status.as_deref() == Some("Success") => ExecVerdict::Success, + Some(s) => ExecVerdict::Failed(s.message.unwrap_or_else(|| "non-zero exit".to_string())), + None => ExecVerdict::NoStatus, + } +} + +/// The actionable failure message for a producer that did not succeed. +pub fn producer_failure_message( + verdict: &ExecVerdict, + pod: &str, + command: &[String], + stderr_tail: &str, +) -> String { + let head = match verdict { + ExecVerdict::Success => "the producer succeeded".to_string(), + ExecVerdict::Failed(detail) => { + format!("the stream command {command:?} in pod `{pod}` failed: {detail}") + } + ExecVerdict::NoStatus => format!( + "the exec connection to pod `{pod}` closed without reporting an exit status, so the \ + dump cannot be assumed complete (a dropped connection is indistinguishable from a \ + clean end-of-output on its own). The snapshot was discarded" + ), + }; + if stderr_tail.trim().is_empty() { + head + } else { + format!("{head}; stderr: {}", stderr_tail.trim()) + } +} + +/// Timeout message for a producer/consumer that overran its budget. +pub fn timeout_message(pod: &str, command: &[String], timeout: Duration, field: &str) -> String { + format!( + "the stream command {command:?} in pod `{pod}` did not finish within {}s. Raise \ + `{field}.timeout`, or make the command faster. The snapshot was discarded, so no \ + partial dump was kept", + timeout.as_secs() + ) +} + +/// Resolve the single workload pod for `selector` in `namespace`. +pub async fn resolve_pod(client: &kube::Client, namespace: &str, selector: &str) -> Result { + let api: Api = Api::namespaced(client.clone(), namespace); + let listed = api + .list(&ListParams::default().labels(selector)) + .await + .map_err(|e| MoverError::StreamPodResolve { + detail: format!("listing pods with selector `{selector}` in `{namespace}` failed: {e}"), + })?; + pick_stream_pod(&listed.items, selector, namespace) + .map(|p| p.clone()) + .map_err(|detail| MoverError::StreamPodResolve { detail }) +} + +/// [`AttachParams`] for a producer exec: read stdout and stderr, no stdin, with +/// buffers sized for bulk transfer rather than kube-rs's 1 KiB default. +pub fn producer_attach_params(container: Option<&str>) -> AttachParams { + let mut p = AttachParams::default() + .stdin(false) + .stdout(true) + .stderr(true) + .max_stdout_buf_size(EXEC_STREAM_BUF) + .max_stderr_buf_size(EXEC_STDERR_BUF); + if let Some(c) = container { + p = p.container(c.to_string()); + } + p +} + +/// [`AttachParams`] for a consumer exec: write stdin, read stderr only. +/// +/// `stdout(false)` is deliberate. The restore target is something like `psql`, whose +/// stdout is chatter we have no use for — and not requesting it means kube-rs never +/// even opens that channel, so there is no path by which restored data could be +/// echoed back into our process. +pub fn consumer_attach_params(container: Option<&str>) -> AttachParams { + let mut p = AttachParams::default() + .stdin(true) + .stdout(false) + .stderr(true) + .max_stdin_buf_size(EXEC_STREAM_BUF) + .max_stderr_buf_size(EXEC_STDERR_BUF); + if let Some(c) = container { + p = p.container(c.to_string()); + } + p +} + +/// Read at most [`EXEC_STDERR_CAP`] bytes of `reader` into a string. +/// +/// Bounded so a chatty command can neither stall on a full pipe nor turn a failure +/// message into a log dump. +pub async fn drain_stderr(reader: &mut R) -> String +where + R: tokio::io::AsyncRead + Unpin, +{ + let mut buf = Vec::new(); + let _ = reader + .take(EXEC_STDERR_CAP as u64) + .read_to_end(&mut buf) + .await; + String::from_utf8_lossy(&buf).into_owned() +} + +/// Copy `reader` into `writer` until EOF, returning the byte count. +/// +/// A thin wrapper so the backup path has ONE place where the dump bytes are handled, +/// and it is obvious by inspection that they only ever move between the two pipes. +pub async fn pump(reader: &mut R, writer: &mut W) -> std::io::Result +where + R: tokio::io::AsyncRead + Unpin, + W: tokio::io::AsyncWrite + Unpin, +{ + let n = tokio::io::copy(reader, writer).await?; + writer.flush().await?; + Ok(n) +} + +/// Run a stream producer and feed its stdout into `stdin`, returning the outcome +/// kopia's runner needs. +/// +/// The contract that matters: this returns [`StdinOutcome::Commit`] ONLY when the +/// exec reported `Success`. Everything else — non-zero exit, a dropped connection +/// (`NoStatus`), a timeout, or an IO error — returns `Abort`, which kills kopia with +/// stdin still open so no manifest is ever written. +/// +/// `stdout` is copied pipe-to-pipe and never inspected. `on_failure` receives the +/// operator-facing message, which contains only the verdict and a bounded stderr +/// tail. +pub async fn feed_from_pod( + client: &kube::Client, + spec: &StreamProducerSpec, + stdin: &mut tokio::process::ChildStdin, + failure: &mut Option, +) -> StdinOutcome { + let timeout = Duration::from_secs(spec.timeout_seconds); + let pod = match resolve_pod(client, &spec.namespace, &spec.pod_selector).await { + Ok(p) => p, + Err(e) => { + *failure = Some(e.to_string()); + return StdinOutcome::Abort; + } + }; + let pod_name = pod.name_any(); + + let api: Api = Api::namespaced(client.clone(), &spec.namespace); + let params = producer_attach_params(spec.container.as_deref()); + let attach = api.exec(&pod_name, spec.command.clone(), ¶ms); + let mut attached = match tokio::time::timeout(timeout, attach).await { + Ok(Ok(a)) => a, + Ok(Err(e)) => { + *failure = Some(format!( + "could not exec into pod `{pod_name}`: {e}. Check that the kopiur mover \ + ServiceAccount is allowed `pods/exec` in namespace `{}`", + spec.namespace + )); + return StdinOutcome::Abort; + } + Err(_) => { + *failure = Some(format!( + "the exec into pod `{pod_name}` did not start within {}s", + timeout.as_secs() + )); + return StdinOutcome::Abort; + } + }; + + let Some(mut out) = attached.stdout() else { + *failure = Some("the exec stream provided no stdout channel".to_string()); + return StdinOutcome::Abort; + }; + let mut errs = attached.stderr(); + let status_fut = attached.take_status(); + + // Pump stdout→kopia and drain stderr TOGETHER. Both channels ride one websocket, + // so leaving stderr unread would eventually block the message loop and stall the + // dump — a hang that would look exactly like a slow database. + let mut stderr_tail = String::new(); + let transfer = async { + let pumped = match errs.as_mut() { + Some(e) => { + let (pumped, tail) = tokio::join!(pump(&mut out, stdin), drain_stderr(e)); + stderr_tail = tail; + pumped + } + None => pump(&mut out, stdin).await, + }; + pumped + }; + + let pumped = match tokio::time::timeout(timeout, transfer).await { + Ok(r) => r, + Err(_) => { + *failure = Some(timeout_message( + &pod_name, + &spec.command, + timeout, + "spec.sources[].stream.workloadExec", + )); + return StdinOutcome::Abort; + } + }; + if let Err(e) = pumped { + *failure = Some(format!( + "the dump stream from pod `{pod_name}` broke mid-transfer: {e}. The snapshot was \ + discarded" + )); + return StdinOutcome::Abort; + } + + // THE decision point. stdout reaching EOF proves nothing — a dropped websocket + // ends the stream just as cleanly as a finished command — so the verdict comes + // from the status channel and nowhere else. + let verdict = match status_fut { + Some(f) => match tokio::time::timeout(timeout, f).await { + Ok(s) => verdict_of(s), + Err(_) => { + *failure = Some(timeout_message( + &pod_name, + &spec.command, + timeout, + "spec.sources[].stream.workloadExec", + )); + return StdinOutcome::Abort; + } + }, + None => ExecVerdict::NoStatus, + }; + + match verdict { + ExecVerdict::Success => { + tracing::info!( + pod = %pod_name, + bytes = pumped.unwrap_or(0), + file = %spec.file_name, + "stream producer finished; committing the snapshot" + ); + StdinOutcome::Commit + } + other => { + *failure = Some(producer_failure_message( + &other, + &pod_name, + &spec.command, + &stderr_tail, + )); + StdinOutcome::Abort + } + } +} + +/// Feed a restored virtual file into a command's stdin in a running pod. +/// +/// The mirror of [`feed_from_pod`]. Both halves must succeed: kopia must read the +/// object cleanly AND the consuming command must exit 0 — a `psql` that failed +/// halfway leaves a half-loaded database, and reporting that as a completed restore +/// would be worse than reporting nothing. +pub async fn restore_into_pod( + client: &kube::Client, + kopia: &kopiur_kopia::KopiaClient, + spec: &StreamConsumerSpec, + object_id: &str, +) -> Result<()> { + let timeout = Duration::from_secs(spec.timeout_seconds); + let pod = resolve_pod(client, &spec.namespace, &spec.pod_selector).await?; + let pod_name = pod.name_any(); + + let api: Api = Api::namespaced(client.clone(), &spec.namespace); + let params = consumer_attach_params(spec.container.as_deref()); + let attach = api.exec(&pod_name, spec.command.clone(), ¶ms); + let mut attached = tokio::time::timeout(timeout, attach) + .await + .map_err(|_| MoverError::StreamExecFailed { + detail: format!( + "the exec into pod `{pod_name}` did not start within {}s", + timeout.as_secs() + ), + })? + .map_err(|e| MoverError::StreamExecFailed { + detail: format!( + "could not exec into pod `{pod_name}`: {e}. Check that the kopiur mover \ + ServiceAccount is allowed `pods/exec` in namespace `{}`", + spec.namespace + ), + })?; + + let mut stdin = attached + .stdin() + .ok_or_else(|| MoverError::StreamExecFailed { + detail: "the exec stream provided no stdin channel".to_string(), + })?; + let mut errs = attached.stderr(); + let status_fut = attached.take_status(); + + let mut stderr_tail = String::new(); + let shipped = { + let ship = async { + // `show_to` streams the object straight into the exec's stdin. + let r = kopia.show_to(object_id, &mut stdin).await; + // Close stdin so the consumer sees EOF and can finish (psql will not exit + // while its input is still open). + let _ = stdin.shutdown().await; + drop(stdin); + r + }; + match errs.as_mut() { + Some(e) => { + let (r, tail) = + tokio::time::timeout(timeout, async { tokio::join!(ship, drain_stderr(e)) }) + .await + .map_err(|_| MoverError::StreamExecFailed { + detail: timeout_message( + &pod_name, + &spec.command, + timeout, + "spec.target.streamExec.workloadExec", + ), + })?; + stderr_tail = tail; + r + } + None => tokio::time::timeout(timeout, ship).await.map_err(|_| { + MoverError::StreamExecFailed { + detail: timeout_message( + &pod_name, + &spec.command, + timeout, + "spec.target.streamExec.workloadExec", + ), + } + })?, + } + }; + shipped.map_err(|e| MoverError::StreamExecFailed { + detail: format!( + "reading `{}` out of the snapshot failed: {e}", + spec.file_name + ), + })?; + + let verdict = match status_fut { + Some(f) => match tokio::time::timeout(timeout, f).await { + Ok(s) => verdict_of(s), + Err(_) => { + return Err(MoverError::StreamExecFailed { + detail: timeout_message( + &pod_name, + &spec.command, + timeout, + "spec.target.streamExec.workloadExec", + ), + }); + } + }, + None => ExecVerdict::NoStatus, + }; + match verdict { + ExecVerdict::Success => { + tracing::info!(pod = %pod_name, file = %spec.file_name, "stream restore consumer finished"); + Ok(()) + } + other => Err(MoverError::StreamExecFailed { + detail: producer_failure_message(&other, &pod_name, &spec.command, &stderr_tail), + }), + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/mover/src/stream/tests.rs b/crates/mover/src/stream/tests.rs new file mode 100644 index 00000000..55952ed5 --- /dev/null +++ b/crates/mover/src/stream/tests.rs @@ -0,0 +1,184 @@ +//! Unit coverage for the pure halves of the stream path: pod selection, exec +//! verdicts, and the messages an operator actually reads when it goes wrong. + +use super::*; +use k8s_openapi::api::core::v1::{Pod, PodStatus}; +use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, Status, Time}; + +fn pod(name: &str, phase: &str, terminating: bool) -> Pod { + Pod { + metadata: ObjectMeta { + name: Some(name.to_string()), + // A `Time` without adding a date-library dev-dependency: it deserializes + // from an RFC3339 string, which is exactly what the API server sends. + deletion_timestamp: terminating.then(|| { + serde_json::from_str::