krabka_restore/args/
value.rs

1//! The compound values the bound flags take, and the parsers clap calls to
2//! build them.
3//!
4//! A `--to-offset`, `--exclude-offset`, or `--exclude-header` value is a small
5//! grammar of its own, and every one of these parsers is a clap `value_parser`
6//! that turns one such string into the type the restore later matches records
7//! against. They live beside those types because a parser and the value it
8//! builds are one contract: the parser is the only thing that constructs the
9//! type, and every rejection an operator can see is written here.
10
11use {
12    base64::{Engine as _, engine::general_purpose},
13    krabka_ids::{Offset, ProducerId},
14    krabka_metadata::NodeId,
15    regex::Regex,
16    uuid::Uuid,
17};
18
19/// Kafka's limit on a topic name, which a bound may not exceed either.
20const MAX_TOPIC_NAME_LEN: usize = 249;
21
22/// A topic partition an operator names in a bound.
23#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, derive_more::Display)]
24#[display("{topic}-{partition}")]
25pub struct PartitionRef {
26    /// Topic name.
27    pub topic: String,
28    /// Partition index.
29    pub partition: i32,
30}
31
32/// One `--to-offset` bound: the last offset the restore keeps in a partition.
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct OffsetBound {
35    /// The partition the bound applies to.
36    pub partition: PartitionRef,
37    /// The highest offset the restore keeps. It is inclusive.
38    pub last_offset: Offset,
39}
40
41/// One `--exclude-offset` range, normalized to a half-open interval.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct OffsetRange {
44    /// The partition the range applies to.
45    pub partition: PartitionRef,
46    /// First excluded offset.
47    pub start: Offset,
48    /// First offset past the range. It is not excluded.
49    pub end_exclusive: Offset,
50}
51
52/// One `--exclude-header` pattern.
53#[derive(Debug, Clone)]
54pub struct HeaderPattern {
55    /// The header name, matched byte for byte.
56    pub name: String,
57    /// The pattern the header value must match for the record to be dropped.
58    pub pattern: Regex,
59}
60
61impl PartialEq for HeaderPattern {
62    /// Two patterns are equal when they name the same header and hold the same
63    /// source pattern. `Regex` has no equality of its own, and the compiled
64    /// program is a function of the source.
65    fn eq(&self, other: &Self) -> bool {
66        self.name == other.name && self.pattern.as_str() == other.pattern.as_str()
67    }
68}
69
70impl Eq for HeaderPattern {}
71
72/// Parse a node id as `krabka-format --node-id` parses it: Kafka's
73/// `node.id`, an `int` that is not negative. The restore formats the target
74/// with the id, so an id that the formatter refuses is refused here first.
75pub(super) fn parse_node_id(s: &str) -> Result<NodeId, String> { krabka_format::parse_node_id(s) }
76
77/// Parses a `--cluster-id` value in Kafka's base64 `Uuid` form (what
78/// `Metadata` and `DescribeCluster` report, e.g. `AQIDBAUGBwgJCgsMDQ4PEA`) or
79/// `java.util.UUID`'s hyphenated form.
80pub(super) fn parse_cluster_id(s: &str) -> Result<Uuid, String> {
81    if let Ok(bytes) = general_purpose::URL_SAFE_NO_PAD.decode(s)
82        && let Ok(bytes) = <[u8; 16]>::try_from(bytes)
83    {
84        return Ok(Uuid::from_bytes(bytes));
85    }
86    Uuid::parse_str(s).map_err(|_| {
87        format!("{s:?} is not a cluster id: neither a base64 Uuid nor a hyphenated UUID")
88    })
89}
90
91/// Parse a producer id. A negative value is the "no producer" sentinel and
92/// never identifies a writer, so it cannot be excluded.
93pub(super) fn parse_producer_id(s: &str) -> Result<ProducerId, String> {
94    let id: i64 = s
95        .trim()
96        .parse()
97        .map_err(|error| format!("producer id: {error}"))?;
98    if id < 0 {
99        return Err(format!("producer id must not be negative, got {id}"));
100    }
101    Ok(ProducerId(id))
102}
103
104/// Compile one operator-supplied pattern.
105pub(super) fn parse_regex(s: &str) -> Result<Regex, String> {
106    Regex::new(s).map_err(|error| format!("pattern {s:?}: {error}"))
107}
108
109/// Check a Kafka topic name: it is not empty, it is at most 249 characters, it
110/// is neither `.` nor `..`, and it holds only `[a-zA-Z0-9._-]`.
111pub(super) fn parse_topic_name(s: &str) -> Result<String, String> {
112    if s.is_empty() {
113        return Err("topic name must not be empty".into());
114    }
115    if s.len() > MAX_TOPIC_NAME_LEN {
116        return Err(format!(
117            "topic name must be at most {MAX_TOPIC_NAME_LEN} characters, got {}",
118            s.len()
119        ));
120    }
121    if s == "." || s == ".." {
122        return Err(format!("topic name must not be {s:?}"));
123    }
124    if let Some(bad) = s
125        .chars()
126        .find(|c| !(c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-')))
127    {
128        return Err(format!(
129            "topic name may hold only [a-zA-Z0-9._-], found {bad:?}"
130        ));
131    }
132    Ok(s.to_owned())
133}
134
135/// Parse the `topic:partition` half of a bound.
136fn parse_partition_ref(s: &str) -> Result<PartitionRef, String> {
137    let (topic, partition) = s
138        .split_once(':')
139        .ok_or_else(|| format!("expected topic:partition, got {s:?}"))?;
140    let topic = parse_topic_name(topic)?;
141    let partition: i32 = partition
142        .parse()
143        .map_err(|error| format!("partition index: {error}"))?;
144    if partition < 0 {
145        return Err(format!(
146            "partition index must not be negative, got {partition}"
147        ));
148    }
149    Ok(PartitionRef { topic, partition })
150}
151
152/// Parse one `--to-offset topic:partition=N`.
153pub(super) fn parse_offset_bound(s: &str) -> Result<OffsetBound, String> {
154    let (partition, last) = s
155        .split_once('=')
156        .ok_or_else(|| format!("expected topic:partition=N, got {s:?}"))?;
157    let partition = parse_partition_ref(partition)?;
158    let last_offset: i64 = last
159        .parse()
160        .map_err(|error| format!("offset bound: {error}"))?;
161    if last_offset < 0 {
162        return Err(format!(
163            "offset bound must not be negative, got {last_offset}"
164        ));
165    }
166    Ok(OffsetBound {
167        partition,
168        last_offset: Offset(last_offset),
169    })
170}
171
172/// Parse one `--exclude-offset topic:partition=A..B` or `topic:partition=A..=B`.
173pub(super) fn parse_offset_range(s: &str) -> Result<OffsetRange, String> {
174    let (partition, range) = s
175        .split_once('=')
176        .ok_or_else(|| format!("expected topic:partition=A..B, got {s:?}"))?;
177    let partition = parse_partition_ref(partition)?;
178    let (start, end, end_included) = if let Some((start, end)) = range.split_once("..=") {
179        (start, end, true)
180    } else {
181        let (start, end) = range
182            .split_once("..")
183            .ok_or_else(|| format!("expected an offset range A..B, got {range:?}"))?;
184        (start, end, false)
185    };
186    let start: i64 = start
187        .parse()
188        .map_err(|error| format!("range start: {error}"))?;
189    let end: i64 = end.parse().map_err(|error| format!("range end: {error}"))?;
190    if start < 0 || end < 0 {
191        return Err(format!(
192            "offset range must not be negative, got {start}..{end}"
193        ));
194    }
195    let end_exclusive = if end_included {
196        end.checked_add(1)
197            .ok_or_else(|| format!("range end {end} overflows when made exclusive"))?
198    } else {
199        end
200    };
201    if start >= end_exclusive {
202        return Err(format!(
203            "offset range must exclude at least one offset, got {range:?}"
204        ));
205    }
206    Ok(OffsetRange {
207        partition,
208        start: Offset(start),
209        end_exclusive: Offset(end_exclusive),
210    })
211}
212
213/// Parse one `--exclude-header NAME=REGEX`.
214///
215/// The split is on the first `=`, so a pattern may hold `=` and a header name
216/// may not.
217pub(super) fn parse_header_pattern(s: &str) -> Result<HeaderPattern, String> {
218    let (name, pattern) = s
219        .split_once('=')
220        .ok_or_else(|| format!("expected NAME=REGEX, got {s:?}"))?;
221    if name.is_empty() {
222        return Err("header name must not be empty".into());
223    }
224    Ok(HeaderPattern {
225        name: name.to_owned(),
226        pattern: parse_regex(pattern)?,
227    })
228}
229
230#[cfg(test)]
231mod tests {
232    use {super::*, crate::args::test_support::partition, assert2::check};
233
234    /// `--cluster-id` accepts Kafka's base64 `Uuid` form -- what `Metadata`
235    /// and `DescribeCluster` report (#1082) -- as well as the hyphenated
236    /// `java.util.UUID` form.
237    #[test]
238    fn parse_cluster_id_accepts_kafka_base64_and_hyphenated_uuid_forms() {
239        let id = Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10);
240        check!(parse_cluster_id("AQIDBAUGBwgJCgsMDQ4PEA") == Ok(id));
241        check!(parse_cluster_id(&id.to_string()) == Ok(id));
242        check!(parse_cluster_id("not-a-cluster-id").is_err());
243    }
244
245    #[yare::parameterized(
246        common = { "orders".to_owned() },
247        one_character = { "a".to_owned() },
248        punctuation = { "a.b_c-d".to_owned() },
249        digit = { "0".to_owned() },
250        maximum_length = { "x".repeat(MAX_TOPIC_NAME_LEN) },
251    )]
252    fn topic_names_accept_kafka_names(good: String) {
253        check!(parse_topic_name(&good) == Ok(good.clone()), "{good:?}");
254    }
255
256    #[yare::parameterized(
257        empty = { String::new() },
258        dot = { ".".to_owned() },
259        double_dot = { "..".to_owned() },
260        whitespace = { "has space".to_owned() },
261        slash = { "has/slash".to_owned() },
262        colon = { "has:colon".to_owned() },
263        equals = { "has=equals".to_owned() },
264        non_ascii = { "café".to_owned() },
265        over_limit = { "x".repeat(MAX_TOPIC_NAME_LEN + 1) },
266    )]
267    fn topic_names_reject_invalid_kafka_names(bad: String) {
268        check!(parse_topic_name(&bad).is_err(), "{bad:?}");
269    }
270
271    #[test]
272    fn partition_refs_split_on_the_colon() {
273        check!(parse_partition_ref("orders:0") == Ok(partition("orders", 0)));
274        check!(parse_partition_ref("a.b-c:12") == Ok(partition("a.b-c", 12)));
275    }
276
277    #[yare::parameterized(
278        missing_colon = { "orders" },
279        missing_partition = { "orders:" },
280        missing_topic = { ":0" },
281        negative_partition = { "orders:-1" },
282        non_numeric_partition = { "orders:x" },
283        extra_colon = { "orders:0:1" },
284        invalid_topic = { "or ders:0" },
285        overflowing_partition = { "orders:99999999999" },
286    )]
287    fn partition_refs_reject_malformed_input(bad: &str) {
288        check!(parse_partition_ref(bad).is_err(), "{bad:?}");
289    }
290
291    #[test]
292    fn partition_ref_displays_the_kafka_spelling() {
293        check!(partition("orders", 3).to_string() == "orders-3");
294    }
295
296    #[test]
297    fn offset_bounds_parse_to_an_inclusive_last_offset() {
298        check!(
299            parse_offset_bound("orders:0=1000")
300                == Ok(OffsetBound {
301                    partition: partition("orders", 0),
302                    last_offset: Offset(1000),
303                })
304        );
305        check!(
306            parse_offset_bound("orders:7=0")
307                == Ok(OffsetBound {
308                    partition: partition("orders", 7),
309                    last_offset: Offset(0),
310                })
311        );
312    }
313
314    #[yare::parameterized(
315        missing_equals = { "orders:0" },
316        wrong_separator = { "orders=1000" },
317        missing_offset = { "orders:0=" },
318        negative_offset = { "orders:0=-1" },
319        non_numeric_offset = { "orders:0=x" },
320        missing_partition = { "=1000" },
321        repeated_equals = { "orders:0=1000=2000" },
322    )]
323    fn offset_bounds_reject_malformed_input(bad: &str) {
324        check!(parse_offset_bound(bad).is_err(), "{bad:?}");
325    }
326
327    #[test]
328    fn offset_ranges_normalize_to_a_half_open_interval() {
329        check!(
330            parse_offset_range("orders:0=100..200")
331                == Ok(OffsetRange {
332                    partition: partition("orders", 0),
333                    start: Offset(100),
334                    end_exclusive: Offset(200),
335                })
336        );
337        check!(
338            parse_offset_range("orders:0=100..=200")
339                == Ok(OffsetRange {
340                    partition: partition("orders", 0),
341                    start: Offset(100),
342                    end_exclusive: Offset(201),
343                })
344        );
345        check!(
346            parse_offset_range("orders:0=0..1")
347                == Ok(OffsetRange {
348                    partition: partition("orders", 0),
349                    start: Offset(0),
350                    end_exclusive: Offset(1),
351                })
352        );
353    }
354
355    #[yare::parameterized(
356        empty = { "orders:0=100..100".to_owned() },
357        inverted = { "orders:0=200..100".to_owned() },
358        inclusive_inverted = { "orders:0=100..=99".to_owned() },
359        missing_range = { "orders:0=100".to_owned() },
360        missing_start = { "orders:0=..200".to_owned() },
361        missing_end = { "orders:0=100..".to_owned() },
362        negative_start = { "orders:0=-1..5".to_owned() },
363        negative_end = { "orders:0=1..-5".to_owned() },
364        missing_offset = { "orders:0".to_owned() },
365        non_numeric = { "orders:0=a..b".to_owned() },
366        inclusive_end_overflow = { format!("orders:0=0..={}", i64::MAX) },
367    )]
368    fn offset_ranges_reject_empty_inverted_and_malformed_input(bad: String) {
369        check!(parse_offset_range(&bad).is_err(), "{bad:?}");
370    }
371
372    #[test]
373    fn header_patterns_split_on_the_first_equals() {
374        let parsed = parse_header_pattern("trace-id=^abc[0-9]+$").expect("pattern");
375        check!(
376            parsed
377                == HeaderPattern {
378                    name: "trace-id".to_owned(),
379                    pattern: Regex::new("^abc[0-9]+$").expect("pattern"),
380                }
381        );
382        let with_equals = parse_header_pattern("op=a=b").expect("pattern");
383        check!(with_equals.name == "op");
384        check!(with_equals.pattern.as_str() == "a=b");
385    }
386
387    #[yare::parameterized(
388        missing_equals = { "trace-id" },
389        missing_name = { "=^abc$" },
390        invalid_regex = { "trace-id=[unclosed" },
391    )]
392    fn header_patterns_reject_malformed_input(bad: &str) {
393        check!(parse_header_pattern(bad).is_err(), "{bad:?}");
394    }
395
396    #[yare::parameterized(
397        negative_sentinel = { "-1" },
398        text = { "x" },
399        empty = { "" },
400        fractional = { "1.5" },
401    )]
402    fn producer_ids_reject_malformed_values(bad: &str) {
403        check!(parse_producer_id(bad).is_err(), "{bad:?}");
404    }
405
406    #[test]
407    fn producer_ids_accept_nonnegative_values() {
408        check!(parse_producer_id("42") == Ok(ProducerId(42)));
409        check!(parse_producer_id(" 7 ") == Ok(ProducerId(7)));
410    }
411
412    #[yare::parameterized(
413        negative = { "-1" },
414        text = { "x" },
415        empty = { "" },
416        overflow = { "2147483648" },
417    )]
418    fn node_ids_reject_values_outside_kafkas_node_id_range(bad: &str) {
419        check!(parse_node_id(bad).is_err(), "{bad:?}");
420    }
421
422    #[test]
423    fn node_ids_accept_kafkas_node_id_range() {
424        check!(parse_node_id("3") == Ok(NodeId(3)));
425        check!(parse_node_id("2147483647") == Ok(NodeId(2_147_483_647)));
426    }
427
428    #[test]
429    fn regexes_compile_or_report_why_not() {
430        check!(parse_regex("^ok$").is_ok());
431        check!(parse_regex("[unclosed").is_err());
432    }
433}
krabka_restore/args/value.rs

krabka_restore/args/
value.rs
