krabka_restore/
args.rs

1//! The `krabka restore` command line and the parsers for its compound values.
2//!
3//! Every flag is long-form. The target-side flags carry the same names as
4//! `krabka format`, because a restore formats the cluster it writes into and an
5//! operator must not have to learn two spellings for one concept.
6//!
7//! The flags are grouped into three flattened structs, one per stage of the
8//! command: where the archive is, what the target cluster is, and what the
9//! restore keeps. `#[command(flatten)]` keeps every flag top-level, so the
10//! grouping shows up in `--help` as headings and nowhere else.
11//!
12//! This file holds the clap definition itself. The values those flags take, and
13//! the parsers clap calls to build them, live in `value` and `timestamp`, and
14//! the cross-flag checks clap cannot express live in `validate`.
15
16use {
17    self::{
18        timestamp::parse_timestamp,
19        value::{
20            parse_cluster_id, parse_header_pattern, parse_node_id, parse_offset_bound,
21            parse_offset_range, parse_producer_id, parse_regex, parse_topic_name,
22        },
23    },
24    crate::report::ReportFormat,
25    clap::Args,
26    krabka_ids::ProducerId,
27    krabka_metadata::NodeId,
28    regex::Regex,
29    std::path::PathBuf,
30    uuid::Uuid,
31};
32
33mod timestamp;
34mod validate;
35mod value;
36
37#[cfg(test)]
38pub mod test_support;
39#[cfg(test)]
40mod tests;
41
42pub use self::value::{HeaderPattern, OffsetBound, OffsetRange, PartitionRef};
43
44/// Where the archive is, and the restore inputs that sit beside it.
45///
46/// The `--archive-*` flags are the ones `krabka backup` takes, from
47/// [`krabka_object_store::ArchiveArgs`]; the rest are restore's own.
48#[derive(Args, Debug)]
49#[command(next_help_heading = "Archive source")]
50pub struct ArchiveArgs {
51    /// The backend and key prefix the archive is read from.
52    #[command(flatten)]
53    pub location: krabka_object_store::ArchiveArgs,
54
55    /// A broker's `<log.dir>/remote-log-metadata/snapshot`.
56    ///
57    /// The snapshot is authoritative about segment lifecycle state for an
58    /// unauthenticated restore. Authenticated restore uses WORM manifests for
59    /// inventory and requires this snapshot's digest in the signed diskless
60    /// capture before seeding its chain receipts into the restored broker.
61    #[arg(long, value_name = "PATH")]
62    pub rlmm_snapshot: Option<PathBuf>,
63
64    /// A controller `<offset>-<epoch>.checkpoint` metadata snapshot.
65    ///
66    /// Topic configuration, ACLs, client quotas, SCRAM credentials, and
67    /// finalized feature levels are recovered from it. Authenticated restore
68    /// requires its digest to be bound into the signed diskless capture.
69    #[arg(long, value_name = "PATH")]
70    pub metadata_snapshot: Option<PathBuf>,
71
72    /// Committed diskless-WAL projection captured by `krabka backup`.
73    #[arg(long, value_name = "PATH")]
74    pub diskless_wal_capture: Option<PathBuf>,
75
76    /// Trusted WORM manifest signing-key id. Repeat with
77    /// `--worm-public-key` to trust archives spanning a key rotation.
78    #[arg(long, value_name = "ID", requires = "worm_public_key")]
79    pub worm_key_id: Vec<String>,
80
81    /// Raw 32-byte Ed25519 public key paired by position with
82    /// `--worm-key-id`. Supplying a pair enables authenticated restore.
83    #[arg(long, value_name = "PATH", requires = "worm_key_id")]
84    pub worm_public_key: Vec<PathBuf>,
85}
86
87/// The cluster the restore writes.
88///
89/// Every flag here carries the name `krabka format` gives it, and is forwarded
90/// to the formatter unchanged.
91#[derive(Args, Debug)]
92#[command(next_help_heading = "Target cluster")]
93pub struct TargetArgs {
94    /// Directory to restore into. Must be empty or absent.
95    #[arg(long, value_name = "DIR")]
96    pub log_dir: PathBuf,
97
98    /// Cluster id of the restored cluster. Generated if not provided.
99    /// Accepts Kafka's base64 `Uuid` form -- what `Metadata` and
100    /// `DescribeCluster` report (#1042) -- or `java.util.UUID`'s hyphenated
101    /// form.
102    #[arg(long, value_parser = parse_cluster_id)]
103    pub cluster_id: Option<Uuid>,
104
105    /// This node's id, Kafka's `node.id`: an integer from 0 to 2147483647.
106    /// Required: the formatter records it in `meta.properties`, and every
107    /// restored partition names this node as its leader and sole replica.
108    #[arg(long, value_parser = parse_node_id)]
109    pub node_id: Option<NodeId>,
110
111    /// Format the restored node as the sole initial controller voter.
112    #[arg(long, conflicts_with_all = ["initial_controllers", "no_initial_controllers"])]
113    pub standalone: bool,
114
115    /// Explicit initial controllers: `id@host:port:directory-id`,
116    /// comma-separated.
117    #[arg(
118        long,
119        value_delimiter = ',',
120        conflicts_with_all = ["standalone", "no_initial_controllers"]
121    )]
122    pub initial_controllers: Vec<String>,
123
124    /// Format a dynamic controller that will join an existing quorum.
125    #[arg(long, conflicts_with_all = ["standalone", "initial_controllers"])]
126    pub no_initial_controllers: bool,
127
128    /// This node's controller listener, as `host:port`.
129    #[arg(long, value_name = "HOST:PORT")]
130    pub controller_listener: Option<String>,
131}
132
133/// Arguments of an offline point-in-time restore.
134///
135/// The fields are public so a test or an embedding tool can build the struct
136/// directly. A `clap::Args` struct with private fields is reachable only
137/// through an argv, which forces every caller through string formatting.
138#[derive(Args, Debug)]
139pub struct RestoreArgs {
140    /// Where the archive is.
141    #[command(flatten)]
142    pub archive: ArchiveArgs,
143
144    /// Independently pinned WORM chain head, `PARTITION_DIR=64_HEX`.
145    #[arg(long, value_name = "PARTITION_DIR=HEX", requires = "worm_key_id")]
146    pub worm_expect_head: Vec<String>,
147
148    /// The cluster the restore writes.
149    #[command(flatten)]
150    pub target: TargetArgs,
151
152    /// Restore this topic. May be repeated. Every topic the archive holds is
153    /// restored when the flag is absent.
154    #[arg(
155        long = "topic",
156        value_name = "NAME",
157        value_parser = parse_topic_name,
158        help_heading = HEADING_BOUNDS
159    )]
160    pub topic: Vec<String>,
161
162    /// Keep offsets at or below `N` in one partition: `topic:partition=N`.
163    /// May be repeated.
164    #[arg(
165        long,
166        value_name = "TOPIC:PARTITION=N",
167        value_parser = parse_offset_bound,
168        help_heading = HEADING_BOUNDS
169    )]
170    pub to_offset: Vec<OffsetBound>,
171
172    /// Keep records whose timestamp is below this instant. Accepts RFC 3339
173    /// with an explicit zone, or bare epoch milliseconds.
174    #[arg(
175        long,
176        value_name = "RFC3339|EPOCH_MS",
177        value_parser = parse_timestamp,
178        help_heading = HEADING_BOUNDS
179    )]
180    pub to_timestamp: Option<i64>,
181
182    /// Drop records whose key matches this pattern. May be repeated.
183    #[arg(
184        long,
185        value_name = "REGEX",
186        value_parser = parse_regex,
187        help_heading = HEADING_BOUNDS
188    )]
189    pub exclude_key: Vec<Regex>,
190
191    /// Drop records that carry a header matching `NAME=REGEX`. May be
192    /// repeated.
193    #[arg(
194        long,
195        value_name = "NAME=REGEX",
196        value_parser = parse_header_pattern,
197        help_heading = HEADING_BOUNDS
198    )]
199    pub exclude_header: Vec<HeaderPattern>,
200
201    /// Drop records written by this producer id. May be repeated.
202    #[arg(
203        long,
204        value_name = "ID",
205        value_parser = parse_producer_id,
206        help_heading = HEADING_BOUNDS
207    )]
208    pub exclude_producer_id: Vec<ProducerId>,
209
210    /// Drop an offset range in one partition: `topic:partition=A..B`, with `B`
211    /// exclusive. Write `A..=B` to include `B`. May be repeated.
212    #[arg(
213        long,
214        value_name = "TOPIC:PARTITION=A..B",
215        value_parser = parse_offset_range,
216        help_heading = HEADING_BOUNDS
217    )]
218    pub exclude_offset: Vec<OffsetRange>,
219
220    /// Verify, check the format, and report without writing to --log-dir.
221    #[arg(long, help_heading = HEADING_BEHAVIOUR)]
222    pub dry_run: bool,
223
224    /// Report format.
225    #[arg(long, value_enum, default_value = "text", help_heading = HEADING_BEHAVIOUR)]
226    pub report: ReportFormat,
227
228    /// Skip a segment that fails verification instead of stopping. The report
229    /// names every segment that was skipped.
230    #[arg(long, help_heading = HEADING_BEHAVIOUR)]
231    pub continue_on_corrupt: bool,
232}
233
234/// Help headings for the flags that are not in a flattened group.
235const HEADING_BOUNDS: &str = "Selection and bounds";
236const HEADING_BEHAVIOUR: &str = "Behaviour";
237
238/// Partitions named by offset bounds or exclusions.
239pub fn offset_partitions(args: &RestoreArgs) -> impl Iterator<Item = &PartitionRef> {
240    args.to_offset
241        .iter()
242        .map(|bound| &bound.partition)
243        .chain(args.exclude_offset.iter().map(|range| &range.partition))
244}
245
246impl RestoreArgs {
247    /// Whether `topic` is in the restore set.
248    ///
249    /// An empty `--topic` list selects every topic the archive holds.
250    #[must_use]
251    pub fn selects_topic(&self, topic: &str) -> bool {
252        self.topic.is_empty() || self.topic.iter().any(|selected| selected == topic)
253    }
254}
krabka_restore/args.rs

krabka_restore/
args.rs
