krabka_backup/
cli.rs

1//! The `krabka-backup` command line.
2//!
3//! Four subcommands, and every one of them takes the same `--archive-*` flag
4//! group: a capture is written into the archive a restore reads, so the tool
5//! never introduces a second place to point at.
6
7use {
8    crate::{archive::ArchiveArgs, connection::SecurityArgs},
9    clap::{Args, Parser, Subcommand},
10    std::path::PathBuf,
11};
12
13/// The capture selector that means "the newest capture in the archive".
14pub const LATEST: &str = "latest";
15
16/// Optional signing key for an authenticated diskless capture boundary.
17#[derive(Args, Debug, Default)]
18pub struct CaptureSigningArgs {
19    /// Stable id of the Ed25519 key that signs the diskless capture.
20    #[arg(long, value_name = "ID", requires = "worm_signing_key")]
21    pub worm_signing_key_id: Option<String>,
22
23    /// PKCS#8 Ed25519 private key used to sign the diskless capture.
24    #[arg(long, value_name = "PATH", requires = "worm_signing_key_id")]
25    pub worm_signing_key: Option<PathBuf>,
26}
27
28/// Independently pinned trust material for a signed diskless capture.
29#[derive(Args, Debug, Default)]
30pub struct CaptureTrustArgs {
31    /// Trusted id of the Ed25519 key that signed the capture.
32    #[arg(long, value_name = "ID", requires_all = ["worm_public_key", "worm_expect_head"])]
33    pub worm_key_id: Option<String>,
34
35    /// Raw 32-byte Ed25519 public key for `--worm-key-id`.
36    #[arg(long, value_name = "PATH", requires_all = ["worm_key_id", "worm_expect_head"])]
37    pub worm_public_key: Option<PathBuf>,
38
39    /// Independently recorded head of the signed diskless capture.
40    #[arg(long, value_name = "HEX", requires_all = ["worm_key_id", "worm_public_key"])]
41    pub worm_expect_head: Option<String>,
42}
43
44/// Captures the inputs a point-in-time restore needs, and puts committed group
45/// offsets back after one.
46#[derive(Parser, Debug)]
47#[command(name = "krabka-backup", version, about, long_about = None)]
48pub struct Cli {
49    /// What to do.
50    #[command(subcommand)]
51    pub command: Command,
52}
53
54/// One thing the tool does.
55#[derive(Subcommand, Debug)]
56pub enum Command {
57    /// Copy this node's restore inputs into the archive.
58    Capture {
59        /// The broker's `log.dir`, mounted read-only. The capture reads the
60        /// metadata checkpoint from its `__cluster_metadata-0`, so it finds one
61        /// only when this is also the node's metadata log directory. Without
62        /// it the capture takes group offsets only.
63        #[arg(long, value_name = "DIR")]
64        log_dir: Option<PathBuf>,
65
66        /// A broker to read committed group offsets from. Without it the
67        /// capture takes the on-disk snapshots only.
68        #[arg(
69            long,
70            short = 'b',
71            value_name = "HOST:PORT",
72            env = "KRABKA_BOOTSTRAP_SERVER"
73        )]
74        bootstrap_server: Option<String>,
75
76        /// TLS and SASL properties for the broker connection.
77        #[command(flatten)]
78        security: SecurityArgs,
79
80        /// Optional authenticated boundary for diskless WAL recovery.
81        #[command(flatten)]
82        signing: CaptureSigningArgs,
83
84        /// Where the capture goes.
85        #[command(flatten, next_help_heading = "Archive")]
86        archive: ArchiveArgs,
87    },
88
89    /// List the captures the archive holds.
90    List {
91        /// Where the captures are.
92        #[command(flatten, next_help_heading = "Archive")]
93        archive: ArchiveArgs,
94    },
95
96    /// Re-read a capture and check every artifact against its recorded digest.
97    Verify {
98        /// Which capture, or `latest`.
99        #[arg(long, value_name = "ID", default_value = LATEST)]
100        capture: String,
101
102        /// Where the captures are.
103        #[command(flatten, next_help_heading = "Archive")]
104        archive: ArchiveArgs,
105    },
106
107    /// Commit a capture's group offsets into a restored cluster.
108    RestoreOffsets {
109        /// Which capture, or `latest`.
110        #[arg(long, value_name = "ID", default_value = LATEST)]
111        capture: String,
112
113        /// The restored cluster.
114        #[arg(
115            long,
116            short = 'b',
117            value_name = "HOST:PORT",
118            env = "KRABKA_BOOTSTRAP_SERVER"
119        )]
120        bootstrap_server: String,
121
122        /// TLS and SASL properties for the restored cluster connection.
123        #[command(flatten)]
124        security: SecurityArgs,
125
126        /// Report what would be committed and commit nothing.
127        #[arg(long)]
128        dry_run: bool,
129
130        /// Trust boundary for offsets bound into a signed diskless capture.
131        #[command(flatten)]
132        trust: CaptureTrustArgs,
133
134        /// Where the captures are.
135        #[command(flatten, next_help_heading = "Archive")]
136        archive: ArchiveArgs,
137    },
138}
139
140#[cfg(test)]
141mod tests {
142    use {
143        super::{Cli, Command},
144        assert2::{assert, check},
145        clap::Parser as _,
146        std::path::PathBuf,
147    };
148
149    #[test]
150    fn a_capture_takes_both_sources_and_one_archive() {
151        let cli = Cli::try_parse_from([
152            "krabka-backup",
153            "capture",
154            "--log-dir",
155            "/var/lib/krabka",
156            "--bootstrap-server",
157            "broker-1:9092",
158            "--archive-s3-bucket",
159            "krabka-tier",
160            "--archive-prefix",
161            "prod/",
162        ])
163        .expect("valid command line");
164
165        assert!(let Command::Capture { .. } = &cli.command);
166        let Command::Capture {
167            log_dir,
168            bootstrap_server,
169            archive,
170            ..
171        } = cli.command
172        else {
173            unreachable!("the assertion above rejected every other command")
174        };
175        check!(log_dir == Some(PathBuf::from("/var/lib/krabka")));
176        check!(bootstrap_server == Some("broker-1:9092".to_owned()));
177        check!(archive.s3_bucket == Some("krabka-tier".to_owned()));
178        check!(archive.prefix == Some("prod/".to_owned()));
179    }
180
181    #[yare::parameterized(
182        capture = { "capture" },
183        list = { "list" },
184        verify = { "verify" },
185        restore_offsets = { "restore-offsets" },
186    )]
187    fn every_subcommand_shows_the_archive_flags_under_one_heading(name: &str) {
188        use clap::CommandFactory as _;
189
190        let help = Cli::command()
191            .find_subcommand(name)
192            .expect("subcommand")
193            .clone()
194            .render_long_help()
195            .to_string();
196        let (_, archive) = help.split_once("\nArchive:\n").expect("an Archive heading");
197        check!(archive.contains("--archive-local"), "{name}");
198        check!(archive.contains("--archive-prefix"), "{name}");
199    }
200
201    #[test]
202    fn a_capture_without_an_archive_is_rejected() {
203        check!(Cli::try_parse_from(["krabka-backup", "capture", "--log-dir", "/data"]).is_err());
204    }
205
206    #[test]
207    fn verify_and_restore_offsets_default_to_the_newest_capture() {
208        let verify =
209            Cli::try_parse_from(["krabka-backup", "verify", "--archive-local", "/mnt/backups"])
210                .expect("valid command line");
211        assert!(let Command::Verify { .. } = &verify.command);
212        let Command::Verify { capture, .. } = verify.command else {
213            unreachable!("the assertion above rejected every other command")
214        };
215        check!(capture == "latest");
216
217        let restore = Cli::try_parse_from([
218            "krabka-backup",
219            "restore-offsets",
220            "-b",
221            "broker-1:9092",
222            "--archive-local",
223            "/mnt/backups",
224        ])
225        .expect("valid command line");
226        assert!(let Command::RestoreOffsets { .. } = &restore.command);
227        let Command::RestoreOffsets {
228            capture, dry_run, ..
229        } = restore.command
230        else {
231            unreachable!("the assertion above rejected every other command")
232        };
233        check!(capture == "latest");
234        check!(!dry_run);
235    }
236
237    #[test]
238    fn restoring_offsets_needs_a_cluster_to_commit_into() {
239        check!(
240            Cli::try_parse_from([
241                "krabka-backup",
242                "restore-offsets",
243                "--archive-local",
244                "/mnt/backups",
245            ])
246            .is_err()
247        );
248    }
249
250    #[test]
251    fn cluster_commands_accept_kafka_command_config() {
252        let capture = Cli::try_parse_from([
253            "krabka-backup",
254            "capture",
255            "-b",
256            "broker-1:9093",
257            "--command-config",
258            "/etc/krabka/backup-client.properties",
259            "--archive-local",
260            "/mnt/backups",
261        ])
262        .expect("capture command config");
263        assert!(let Command::Capture { .. } = &capture.command);
264
265        let restore = Cli::try_parse_from([
266            "krabka-backup",
267            "restore-offsets",
268            "-b",
269            "broker-1:9093",
270            "--command-config",
271            "/etc/krabka/backup-client.properties",
272            "--archive-local",
273            "/mnt/backups",
274        ])
275        .expect("restore command config");
276        assert!(let Command::RestoreOffsets { .. } = &restore.command);
277    }
278
279    #[test]
280    fn diskless_capture_signing_key_is_an_explicit_pair() {
281        check!(
282            Cli::try_parse_from([
283                "krabka-backup",
284                "capture",
285                "-b",
286                "broker:9092",
287                "--archive-local",
288                "/archive",
289                "--worm-signing-key-id",
290                "capture-key",
291                "--worm-signing-key",
292                "/keys/capture.pk8",
293            ])
294            .is_ok()
295        );
296        check!(
297            Cli::try_parse_from([
298                "krabka-backup",
299                "capture",
300                "-b",
301                "broker:9092",
302                "--archive-local",
303                "/archive",
304                "--worm-signing-key-id",
305                "capture-key",
306            ])
307            .is_err()
308        );
309    }
310
311    #[test]
312    fn signed_offset_restore_trust_is_an_explicit_triple() {
313        let base = [
314            "krabka-backup",
315            "restore-offsets",
316            "-b",
317            "broker:9092",
318            "--archive-local",
319            "/archive",
320        ];
321        check!(
322            Cli::try_parse_from(base.into_iter().chain([
323                "--worm-key-id",
324                "capture-key",
325                "--worm-public-key",
326                "/keys/capture.pub",
327                "--worm-expect-head",
328                "00",
329            ]))
330            .is_ok()
331        );
332        check!(
333            Cli::try_parse_from(base.into_iter().chain(["--worm-key-id", "capture-key"])).is_err()
334        );
335    }
336}
krabka_backup/cli.rs

krabka_backup/
cli.rs
