krabka_format/format/
args.rs

1//! The command line `krabka format` accepts, and the scalar value parsers that
2//! belong to no larger concern.
3//!
4//! [`FormatArgs`] is the one description of that command line: the binary and
5//! [`crate::run_from_args`] both parse into it, so a caller of either sees the
6//! same flags. The structured values behind `--feature`, `--add-scram`, and
7//! `--add-acl` parse in the module that owns each of those concerns, and this
8//! module names their parsers in the `value_parser` attributes.
9
10use {
11    super::{acl::parse_acl_spec, features::parse_feature_spec, scram::parse_scram_spec},
12    crate::ids::{ClusterId, DirectoryId},
13    clap::Args,
14    krabka_metadata::{AclEntry, FeatureLevelRecord},
15    krabka_security::SaslMechanism,
16    std::path::PathBuf,
17};
18
19#[derive(Args, Debug)]
20pub struct FormatArgs {
21    /// A log directory to format, an entry of Kafka's `log.dirs`. Repeat the
22    /// flag, or separate paths with commas, to format every directory of the
23    /// node in one run. Without `--metadata-log-dir`, the first directory is
24    /// also the metadata log directory, as Kafka's `metadata.log.dir` defaults
25    /// to the first entry of `log.dirs`. `kafka-storage format` reads the same
26    /// set from its `--config` file. Krabka takes no `server.properties`, so
27    /// there is no `--config`.
28    #[arg(
29        long = "log-dir",
30        value_name = "DIR",
31        value_delimiter = ',',
32        required = true
33    )]
34    pub(super) log_dirs: Vec<PathBuf>,
35    /// The metadata log directory, Kafka's `metadata.log.dir`. The run formats
36    /// it together with every `--log-dir`, as `kafka-storage format` adds
37    /// `metadata.log.dir` to the `log.dirs` set. Only this directory gets the
38    /// bootstrap files and the `__cluster_metadata-0` checkpoint. It can also
39    /// be one of the `--log-dir` entries. Give the broker the same directory
40    /// as its `--metadata-log-dir`. Defaults to the first `--log-dir`.
41    #[arg(long, value_name = "DIR")]
42    pub(super) metadata_log_dir: Option<PathBuf>,
43    /// Cluster id. Written and printed in Kafka's 22-character base64 form.
44    /// Accepts that form or the hyphenated form. When omitted, the id of an
45    /// already formatted directory in the set is kept, and otherwise a new id
46    /// is generated.
47    #[arg(long, value_parser = ClusterId::parse_cli)]
48    pub(super) cluster_id: Option<ClusterId>,
49    /// Bootstrap `metadata.version` (KIP-778), e.g. `4.0` or `4.0-IV3`. A
50    /// string with more than two dot-separated segments keeps the first two,
51    /// as Kafka's `MetadataVersion.fromVersionString` does, so `4.3.1` is
52    /// `4.3`. Defaults to Kafka 4.3's latest production level, `4.3-IV0`, when
53    /// omitted.
54    #[arg(long)]
55    pub(super) release_version: Option<String>,
56    /// Set an individual feature's finalized level at format time (KIP-1022),
57    /// e.g. `--feature transaction.version=2`. May be repeated. Combines with
58    /// `--release-version` (which sets the base release) for every feature
59    /// except `metadata.version`, where the two conflict.
60    #[arg(long = "feature", value_parser = parse_feature_spec)]
61    pub(super) feature: Vec<FeatureLevelRecord>,
62    /// Accept feature levels past the latest production ones: Kafka trunk's
63    /// `metadata.version` `4.4-IV0` to `4.4-IV2`, which a stock Kafka 4.3.1
64    /// does not know. It is Kafka's internal `unstable.feature.versions.enable`,
65    /// which `kafka-storage format` reads from its `--config` file; without it
66    /// such a release is refused as Kafka refuses it. A node formatted at one
67    /// of those levels needs the same setting in its `server_properties`.
68    #[arg(
69        long,
70        action = clap::ArgAction::SetTrue,
71        value_parser = clap::builder::TypedValueParser::map(
72            clap::builder::BoolValueParser::new(),
73            krabka_raft::UnstableFeatureVersions::from,
74        ),
75    )]
76    pub(super) unstable_feature_versions_enable: krabka_raft::UnstableFeatureVersions,
77    /// Seed a SCRAM credential. May be repeated.
78    /// Format: `SCRAM-SHA-256=[name=<u>,password=<p>,iterations=<n>]`
79    /// or `SCRAM-SHA-512=[name=<u>,password=<p>,iterations=<n>]`
80    /// (iterations defaults to 4096 when omitted)
81    #[arg(long, value_parser = parse_scram_spec)]
82    pub(super) add_scram: Vec<ScramSpec>,
83    /// Seed an ACL entry. May be repeated.
84    /// Format: `principal=User:<name>,host=<ip|*>,operation=<Op>,permission=<Allow|Deny>,resource=<Type>:<Name>[:<Pattern>]`
85    /// Pattern defaults to `Literal`.
86    #[arg(long, value_parser = parse_acl_spec)]
87    pub(super) add_acl: Vec<AclEntry>,
88    /// This node's id, Kafka's `node.id`: an integer from 0 to 2147483647.
89    /// Every `meta.properties` records it, and the broker refuses to start on
90    /// a directory that records another id. `kafka-storage format` reads it
91    /// from its `--config` file. Give the broker the same id as its
92    /// `--broker-id`.
93    #[arg(long, value_parser = parse_node_id)]
94    pub(super) node_id: krabka_metadata::NodeId,
95    /// The metadata log directory's stable id: the node's KIP-853 voter
96    /// identity. Intended for orchestrators that verify the exact node
97    /// incarnation before they declare it ready. Accepts Kafka's base64 form
98    /// or the hyphenated form. The other directories get generated ids.
99    #[arg(long, value_parser = parse_directory_id)]
100    pub(super) directory_id: Option<DirectoryId>,
101    /// Format this node as the sole initial controller voter.
102    #[arg(
103        long,
104        conflicts_with_all = ["initial_controllers", "no_initial_controllers"]
105    )]
106    pub(super) standalone: bool,
107    /// Explicit initial controllers: `id@host:port:directory-id`,
108    /// comma-separated. The directory id is in Kafka's base64 form or the
109    /// hyphenated form.
110    #[arg(
111        long,
112        value_delimiter = ',',
113        conflicts_with_all = ["standalone", "no_initial_controllers"]
114    )]
115    pub(super) initial_controllers: Vec<String>,
116    /// Format a dynamic controller that will join an existing quorum.
117    #[arg(
118        long,
119        conflicts_with_all = ["standalone", "initial_controllers"]
120    )]
121    pub(super) no_initial_controllers: bool,
122    /// This node's controller listener (`host:port`) — written into the
123    /// `VotersRecord` when `--standalone`.
124    #[arg(long)]
125    pub(super) controller_listener: Option<String>,
126    /// The name of the controller listener: the first entry of Kafka's
127    /// `controller.listener.names`. Each voter endpoint that `--standalone` or
128    /// `--initial-controllers` writes carries it, as `kafka-storage format`
129    /// names them. A leader refuses an `AddRaftVoter` whose endpoints do not
130    /// include its own listener name.
131    #[arg(long, default_value = "CONTROLLER", value_parser = parse_listener_name)]
132    pub(super) controller_listener_name: String,
133    /// Skip an already formatted directory, instead of refusing the run, and
134    /// format the others. Matches `kafka-storage format --ignore-formatted`.
135    ///
136    /// This is what makes the formatter safe to run unconditionally, which a
137    /// Kubernetes init container has to: the image carries no shell, so there
138    /// is nothing to test the directory with before the call, and a pod that
139    /// restarts against its existing volume would otherwise fail every
140    /// restart after the first.
141    #[arg(long)]
142    pub(super) ignore_formatted: bool,
143}
144
145#[derive(Debug, Clone, PartialEq, Eq)]
146pub struct ScramSpec {
147    pub(super) mechanism: SaslMechanism,
148    pub(super) name: String,
149    pub(super) password: String,
150    pub(super) iterations: u32,
151}
152
153/// Parses a node id as Kafka's `node.id`: an `int` that is not negative.
154///
155/// `meta.properties` holds the id as an `int`, which Kafka's tools read with
156/// `Integer.parseInt`, so a larger id would make a file they cannot read. The
157/// negative-id message is the one Kafka's `Formatter.run` throws.
158///
159/// # Errors
160///
161/// Returns a message for a value that is not an integer, is past the `int`
162/// range, or is negative.
163pub fn parse_node_id(s: &str) -> Result<krabka_metadata::NodeId, String> {
164    let id: i32 = s.trim().parse().map_err(|e| format!("node id: {e}"))?;
165    let id = u64::try_from(id)
166        .map_err(|_| "You must specify a valid non-negative node ID.".to_owned())?;
167    Ok(krabka_metadata::NodeId(id))
168}
169
170/// Parses `--directory-id`: Kafka's base64 form or the hyphenated form, and
171/// never one of the 100 ids that Kafka reserves as directory-id sentinels.
172///
173/// The broker refuses a reserved id at startup, as Kafka's
174/// `MetaPropertiesEnsemble.verify` does, so the format refuses it first and
175/// writes nothing.
176///
177/// # Errors
178///
179/// Returns the message of [`DirectoryId::parse_cli`], or one for a reserved
180/// id.
181pub fn parse_directory_id(s: &str) -> Result<DirectoryId, String> {
182    let id = DirectoryId::parse_cli(s).map_err(|e| e.to_string())?;
183    if id.is_reserved() {
184        return Err(format!(
185            "Invalid reserved directory ID {id}: Kafka reserves the 100 lowest directory ids"
186        ));
187    }
188    Ok(id)
189}
190
191/// Parse `--controller-listener-name` into Kafka's normalised form: upper
192/// case, as `ListenerName.normalised` makes it.
193///
194/// # Errors
195///
196/// Returns a message for an empty name, or one with leading or trailing white
197/// space, which Kafka's `RaftVoterEndpoint` refuses.
198pub fn parse_listener_name(s: &str) -> Result<String, String> {
199    if s.is_empty() || s.trim() != s {
200        return Err(format!(
201            "controller listener name {s:?} must be non-empty, without leading or trailing \
202             white space"
203        ));
204    }
205    Ok(s.to_ascii_uppercase())
206}
207
208#[cfg(test)]
209mod tests {
210
211    use {
212        super::*,
213        crate::Cli,
214        assert2::check,
215        clap::{Parser as _, error::ErrorKind},
216    };
217
218    /// `--log-dir` takes repeated flags and comma-separated lists alike, and
219    /// keeps the order they were given in.
220    #[yare::parameterized(
221        repeated_flags = { &["--log-dir", "/a", "--log-dir", "/b"] },
222        comma_separated = { &["--log-dir", "/a,/b"] },
223        equals_form = { &["--log-dir=/a", "--log-dir=/b"] },
224    )]
225    fn log_dir_is_repeatable_and_comma_separated(argv: &[&str]) {
226        let cli = Cli::try_parse_from(
227            ["krabka-format", "--node-id", "1"]
228                .into_iter()
229                .chain(argv.iter().copied()),
230        )
231        .expect("parse");
232        assert2::check!(
233            cli.args.log_dirs == vec![PathBuf::from("/a"), PathBuf::from("/b")],
234            "{argv:?}"
235        );
236    }
237
238    /// `--metadata-log-dir` takes one path, comma and all, as Kafka's
239    /// `metadata.log.dir` is one path. It is optional, it does not replace the
240    /// required `--log-dir`, and clap refuses it twice.
241    #[yare::parameterized(
242        log_only = { &["--log-dir", "/a"], None, &["/a"][..], None },
243        metadata_and_logs = {
244            &["--metadata-log-dir", "/m", "--log-dir", "/a,/b"],
245            Some("/m"),
246            &["/a", "/b"][..],
247            None,
248        },
249        same_metadata_and_log = {
250            &["--log-dir=/a", "--metadata-log-dir=/a"],
251            Some("/a"),
252            &["/a"][..],
253            None,
254        },
255        comma_in_metadata_is_path = {
256            &["--metadata-log-dir", "/m,/n", "--log-dir", "/a"],
257            Some("/m,/n"),
258            &["/a"][..],
259            None,
260        },
261        missing_required_log = {
262            &["--metadata-log-dir", "/m"],
263            None,
264            &[][..],
265            Some(ErrorKind::MissingRequiredArgument),
266        },
267        repeated_metadata_conflicts = {
268            &["--metadata-log-dir", "/m", "--metadata-log-dir", "/n", "--log-dir", "/a"],
269            None,
270            &[][..],
271            Some(ErrorKind::ArgumentConflict),
272        },
273    )]
274    fn metadata_log_dir_is_one_optional_path(
275        argv: &[&str],
276        metadata: Option<&str>,
277        log_dirs: &[&str],
278        error: Option<ErrorKind>,
279    ) {
280        let got = Cli::try_parse_from(
281            ["krabka-format", "--node-id", "1"]
282                .into_iter()
283                .chain(argv.iter().copied()),
284        )
285        .map(|cli| (cli.args.metadata_log_dir, cli.args.log_dirs))
286        .map_err(|error| error.kind());
287        let want = error.map_or_else(
288            || {
289                Ok((
290                    metadata.map(PathBuf::from),
291                    log_dirs.iter().map(PathBuf::from).collect(),
292                ))
293            },
294            Err,
295        );
296        assert2::check!(got == want, "{argv:?}");
297    }
298
299    /// A node id is Kafka's `int` without a sign: 0 through 2147483647.
300    /// Everything else is an error rather than a silent zero.
301    #[yare::parameterized(
302        ordinary = { "7", Ok(7) },
303        trims_spaces = { "  7  ", Ok(7) },
304        zero = { "0", Ok(0) },
305        max_i32 = { "2147483647", Ok(2_147_483_647) },
306        overflow = { "2147483648", Err("node id: number too large to fit in target type") },
307        negative = { "-1", Err("You must specify a valid non-negative node ID.") },
308        empty = { "", Err("node id: cannot parse integer from empty string") },
309        decimal = { "1.0", Err("node id: invalid digit found in string") },
310        hex = { "0x7", Err("node id: invalid digit found in string") },
311    )]
312    fn parse_node_id_takes_kafkas_node_id_range(input: &str, want: Result<u64, &str>) {
313        assert2::check!(
314            parse_node_id(input).map(|n| n.0) == want.map_err(str::to_owned),
315            "{input:?}"
316        );
317    }
318
319    /// `--directory-id` takes either id form and refuses the ids that Kafka
320    /// reserves, which the broker would refuse at startup.
321    #[yare::parameterized(
322        usable_uuid = {
323            uuid::Uuid::from_u64_pair(1, 1).to_string(),
324            Ok(DirectoryId::from(uuid::Uuid::from_u64_pair(1, 1))),
325        },
326        usable_kafka_form = {
327            DirectoryId::from(uuid::Uuid::from_u64_pair(1, 1)).to_string(),
328            Ok(DirectoryId::from(uuid::Uuid::from_u64_pair(1, 1))),
329        },
330        reserved_one = {
331            uuid::Uuid::from_u128(1).to_string(),
332            Err(format!(
333                "Invalid reserved directory ID {}: Kafka reserves the 100 lowest directory ids",
334                DirectoryId::from(uuid::Uuid::from_u128(1))
335            )),
336        },
337        reserved_ninety_nine = {
338            uuid::Uuid::from_u128(99).to_string(),
339            Err(format!(
340                "Invalid reserved directory ID {}: Kafka reserves the 100 lowest directory ids",
341                DirectoryId::from(uuid::Uuid::from_u128(99))
342            )),
343        },
344        first_usable = {
345            DirectoryId::from(uuid::Uuid::from_u128(100)).to_string(),
346            Ok(DirectoryId::from(uuid::Uuid::from_u128(100))),
347        },
348    )]
349    fn parse_directory_id_refuses_reserved_ids(input: String, want: Result<DirectoryId, String>) {
350        assert2::check!(parse_directory_id(&input) == want, "{input:?}");
351    }
352
353    /// `--node-id` is required, as Kafka's `node.id` is: every
354    /// `meta.properties` records it.
355    #[test]
356    fn node_id_is_required() {
357        let missing = Cli::try_parse_from(["krabka-format", "--log-dir", "/a"])
358            .map(|cli| cli.args.node_id)
359            .map_err(|error| error.kind());
360        check!(missing == Err(ErrorKind::MissingRequiredArgument));
361        let given = Cli::try_parse_from(["krabka-format", "--log-dir", "/a", "--node-id", "3"])
362            .map(|cli| cli.args.node_id)
363            .map_err(|error| error.kind());
364        check!(given == Ok(krabka_metadata::NodeId(3)));
365    }
366}
krabka_format/format/args.rs

krabka_format/format/
args.rs
