krabka_backup/
lib.rs

1//! Captures the inputs a point-in-time restore needs, and puts committed
2//! consumer-group offsets back after one.
3//!
4//! `krabka restore` rebuilds a log directory out of a KIP-405 archive, and it
5//! needs three things the archive does not hold. Two are files on a broker's
6//! own disk: `<log.dir>/remote-log-metadata/snapshot`, without which a segment
7//! the old cluster had already released is indistinguishable from a live one,
8//! and the controller's newest `<end-offset>-<epoch>.checkpoint`, which is
9//! where topic configuration, ACLs, client quotas, SCRAM credentials and
10//! finalized feature levels live. The third is not a file at all: the committed
11//! offsets of every consumer group, which sit in the compacted, never-tiered
12//! `__consumer_offsets` topic.
13//!
14//! All three are destroyed by the disaster that makes an operator reach for a
15//! restore, so a copy has to exist before it. This tool makes that copy and
16//! checks it.
17//!
18//! # Why a tool and not a `kubectl exec`
19//!
20//! The krabka container image is built from an apko base and has no shell, no
21//! `cp` and no `tar`, so `kubectl exec` and `kubectl cp` cannot take a file off
22//! a running broker. The supported copy is this binary, run beside the broker
23//! with the same volume mounted read-only — a `CronJob` in the same namespace —
24//! or on the node itself. Both snapshot files are written temp-then-rename, so
25//! a reader that opens one by name always sees a whole file.
26//!
27//! Group offsets need a cluster rather than a volume, so the same run takes
28//! them over the ordinary Kafka wire protocol, with `ListGroups` and
29//! `OffsetFetch`. Nothing here speaks a krabka-private API key, which is what
30//! lets the same capture run against a cluster that is not krabka.
31//!
32//! # The four subcommands
33//!
34//! `capture` copies the inputs into the archive under
35//! `restore-inputs/<capture-id>/`, beside the tiered segments the restore reads,
36//! with a `manifest.json` recording each artifact's size and SHA-256. `list`
37//! names the captures. `verify` re-reads one and checks it against those
38//! digests, which is the check a backup is worth nothing without. `restore-
39//! offsets` commits a capture's offsets into a restored cluster, so a group
40//! resumes where it stopped instead of at `auto.offset.reset`.
41//!
42//! # Exit codes
43//!
44//! `0` success, [`EXIT_BAD_ARGUMENT`], [`EXIT_UNREADABLE`], [`EXIT_INTEGRITY`]
45//! and [`EXIT_CLUSTER`]. They line up with `krabka restore`'s own where the two
46//! tools mean the same thing, because one runbook branches on both.
47//!
48//! The crate is a library as well as a binary, for the same reason
49//! `krabka-guard`, `krabka-barrier` and `krabka-format` are: a test that spawns
50//! the binary needs a Cargo working tree to build it from, and a Bazel test
51//! sandbox has none. Tests call [`run_from_args`] in process instead.
52//!
53//! The operations in [`mod@run`] return an [`OperationReport`] with an outcome and ordered diagnostics, including progress before an error. They do not print. [`run()`] renders those diagnostics for the CLI.
54
55use {clap::Parser as _, std::io};
56
57pub mod archive;
58pub mod capture;
59pub mod cli;
60pub mod connection;
61pub mod diskless;
62pub mod error;
63pub mod manifest;
64pub mod offsets;
65pub mod report;
66pub mod run;
67
68pub use self::{
69    cli::{Cli, Command},
70    error::{BackupError, EXIT_BAD_ARGUMENT, EXIT_CLUSTER, EXIT_INTEGRITY, EXIT_UNREADABLE},
71    report::{BackupEvent, OperationReport},
72};
73
74krabka_macros::parsed_cli_entrypoint!({
75    /// Run the tool from an argv-style iterator, returning its exit code.
76    ///
77    /// # Panics
78    ///
79    /// Panics if `argv` does not parse, which for a caller passing a literal
80    /// argument list is a bug in that list rather than a runtime condition.
81});
82
83/// Run one parsed command line, mapping its error onto an exit code.
84pub async fn run(cli: Cli) -> i32 {
85    let outcome = match &cli.command {
86        Command::Capture {
87            log_dir,
88            bootstrap_server,
89            security,
90            signing,
91            archive,
92        } => render_report(
93            run::capture_configured(
94                log_dir.as_deref(),
95                bootstrap_server.as_deref(),
96                security,
97                signing,
98                archive,
99            )
100            .await,
101        ),
102        Command::List { archive } => render_report(run::list(archive).await),
103        Command::Verify { capture, archive } => render_report(run::verify(capture, archive).await),
104        Command::RestoreOffsets {
105            capture,
106            bootstrap_server,
107            security,
108            dry_run,
109            trust,
110            archive,
111        } => render_report(
112            run::restore_offsets_configured(
113                capture,
114                bootstrap_server,
115                security,
116                trust,
117                *dry_run,
118                archive,
119            )
120            .await,
121        ),
122    };
123    match outcome {
124        Ok(()) => 0,
125        Err(error) => {
126            eprintln!("{error}");
127            error.exit_code()
128        }
129    }
130}
131
132fn render_report<T>(report: OperationReport<T>) -> Result<(), BackupError> {
133    report::render(&report.events, &mut io::stdout().lock())?;
134    report.result.map(|_| ())
135}
136
137#[cfg(test)]
138mod tests {
139    use {
140        super::{
141            Cli, Command, EXIT_BAD_ARGUMENT, EXIT_INTEGRITY, EXIT_UNREADABLE, run, run_from_args,
142        },
143        crate::{
144            capture::capture_key,
145            manifest::{MANIFEST, RLMM_SNAPSHOT},
146        },
147        assert2::check,
148        clap::Parser as _,
149        std::{fs, path::Path},
150    };
151
152    // A log directory holding the RLMM snapshot a capture takes off a node.
153    krabka_macros::snapshot_node_fixture!(node);
154
155    /// The one capture id the archive holds, which `latest` also resolves to.
156    fn only_capture(archive_root: &Path) -> String {
157        let mut ids: Vec<String> = fs::read_dir(archive_root.join("restore-inputs"))
158            .expect("read the capture root")
159            .flatten()
160            .filter_map(|entry| entry.file_name().to_str().map(ToOwned::to_owned))
161            .collect();
162        ids.sort();
163        ids.pop().expect("the archive holds a capture")
164    }
165
166    #[tokio::test(flavor = "multi_thread")]
167    async fn a_capture_and_the_checks_that_read_it_back_all_exit_zero() {
168        let node = node();
169        let archive_root = tempfile::tempdir().expect("archive root");
170        let log_dir = node.path().display().to_string();
171        let archive = archive_root.path().display().to_string();
172
173        check!(
174            run_from_args([
175                "krabka-backup",
176                "capture",
177                "--log-dir",
178                &log_dir,
179                "--archive-local",
180                &archive,
181            ])
182            .await
183                == 0
184        );
185        check!(run_from_args(["krabka-backup", "list", "--archive-local", &archive]).await == 0);
186        check!(run_from_args(["krabka-backup", "verify", "--archive-local", &archive]).await == 0);
187        check!(
188            run_from_args([
189                "krabka-backup",
190                "restore-offsets",
191                "--dry-run",
192                "-b",
193                "127.0.0.1:9092",
194                "--capture",
195                &only_capture(archive_root.path()),
196                "--archive-local",
197                &archive,
198            ])
199            .await
200                == EXIT_UNREADABLE,
201            "a capture of the on-disk inputs alone holds no offsets to restore",
202        );
203    }
204
205    #[tokio::test(flavor = "multi_thread")]
206    async fn a_damaged_capture_exits_with_the_integrity_code_a_runbook_branches_on() {
207        let node = node();
208        let archive_root = tempfile::tempdir().expect("archive root");
209        let archive = archive_root.path().display().to_string();
210        check!(
211            run_from_args([
212                "krabka-backup",
213                "capture",
214                "--log-dir",
215                &node.path().display().to_string(),
216                "--archive-local",
217                &archive,
218            ])
219            .await
220                == 0
221        );
222
223        let id = only_capture(archive_root.path());
224        fs::write(
225            archive_root.path().join(capture_key(&id, RLMM_SNAPSHOT)),
226            b"truncated",
227        )
228        .expect("truncate the captured snapshot");
229
230        check!(
231            run_from_args(["krabka-backup", "verify", "--archive-local", &archive]).await
232                == EXIT_INTEGRITY
233        );
234
235        // And an archive whose manifest is gone is unreadable rather than
236        // corrupt: the two codes send a runbook down different branches.
237        fs::remove_file(archive_root.path().join(capture_key(&id, MANIFEST)))
238            .expect("remove the manifest");
239        check!(
240            run_from_args(["krabka-backup", "verify", "--archive-local", &archive]).await
241                == EXIT_UNREADABLE
242        );
243    }
244
245    #[tokio::test(flavor = "multi_thread")]
246    async fn a_capture_that_finds_nothing_exits_unreadable() {
247        let empty = tempfile::tempdir().expect("an empty log dir");
248        let archive_root = tempfile::tempdir().expect("archive root");
249        check!(
250            run_from_args([
251                "krabka-backup",
252                "capture",
253                "--log-dir",
254                &empty.path().display().to_string(),
255                "--archive-local",
256                &archive_root.path().display().to_string(),
257            ])
258            .await
259                == EXIT_UNREADABLE
260        );
261    }
262
263    #[tokio::test(flavor = "multi_thread")]
264    async fn a_flag_that_names_a_backend_that_was_not_selected_exits_bad_argument() {
265        let archive_root = tempfile::tempdir().expect("archive root");
266        let cli = Cli::parse_from([
267            "krabka-backup",
268            "list",
269            "--archive-local",
270            &archive_root.path().display().to_string(),
271            "--archive-s3-region",
272            "eu-west-1",
273        ]);
274        check!(let Command::List { .. } = &cli.command);
275        check!(run(cli).await == EXIT_BAD_ARGUMENT);
276    }
277}
krabka_backup/lib.rs

krabka_backup/
lib.rs
