krabka_raft/connection_limiter/
settings.rs

1//! The connection quotas of Kafka's `ConnectionQuotas`, read from the dynamic
2//! broker configs of the metadata image.
3//!
4//! Kafka layers a node's own dynamic config over the cluster default, and a
5//! listener's `listener.name.<listener>.` value over the broker-wide one
6//! (`KafkaConfig.valuesWithPrefixOverride`). A value Kafka would not apply
7//! leaves the value in force before it: `DynamicBrokerConfig` keeps its
8//! previous config when the new one fails `KafkaConfig` validation or a
9//! reconfigurable's `validateReconfiguration`. Kafka drops the whole update;
10//! krabka keeps the previous value of the refused key alone. With no previous
11//! value, at startup, the refused value is skipped for the next layer, as
12//! `DynamicBrokerConfig.initialize` drops an invalid value from its layer.
13
14use {crate::types::NodeId, krabka_metadata::MetadataImage, std::collections::BTreeMap};
15
16/// Kafka's `SocketServerConfigs.MAX_CONNECTIONS_CONFIG`.
17pub const MAX_CONNECTIONS: &str = "max.connections";
18
19/// Kafka's `SocketServerConfigs.MAX_CONNECTIONS_PER_IP_CONFIG`.
20pub const MAX_CONNECTIONS_PER_IP: &str = "max.connections.per.ip";
21
22/// Kafka's `SocketServerConfigs.MAX_CONNECTIONS_PER_IP_OVERRIDES_CONFIG`.
23pub const MAX_CONNECTIONS_PER_IP_OVERRIDES: &str = "max.connections.per.ip.overrides";
24
25/// Kafka's `SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG`.
26pub const MAX_CONNECTION_CREATION_RATE: &str = "max.connection.creation.rate";
27
28/// The two caps a node is started with. `usize::MAX` means unlimited.
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub struct ConnectionLimits {
31    /// `max.connections`: live connections across the node's listeners.
32    pub max_connections: usize,
33    /// `max.connections.per.ip`: live connections from one client IP.
34    pub max_connections_per_ip: usize,
35}
36
37/// The quotas in force for one listener, before the hosts of
38/// `max.connections.per.ip.overrides` are resolved.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct ConnectionQuotaSettings {
41    /// `max.connections`: the broker-wide cap.
42    pub max_connections: usize,
43    /// `max.connections.per.ip`: the cap for an IP with no override.
44    pub max_connections_per_ip: usize,
45    /// `max.connections.per.ip.overrides`: the cap of each host, before the
46    /// host is resolved to an address.
47    pub per_ip_overrides: BTreeMap<String, usize>,
48    /// `listener.name.<listener>.max.connections`, else `max.connections`.
49    pub listener_max_connections: usize,
50    /// `max.connection.creation.rate` per second; `None` is Kafka's
51    /// `Int.MaxValue` default, no limit.
52    pub max_connection_creation_rate: Option<u32>,
53    /// `listener.name.<listener>.max.connection.creation.rate`, else the
54    /// broker-wide rate.
55    pub listener_max_connection_creation_rate: Option<u32>,
56}
57
58/// Kafka's `ListenerName.configPrefix` applied to `key`: the listener name is
59/// lowercased.
60fn listener_key(listener: &str, key: &str) -> String {
61    format!("listener.name.{}.{key}", listener.to_ascii_lowercase())
62}
63
64/// `String.trim`, which `ConfigDef` applies to every value.
65fn java_trim(value: &str) -> &str { value.trim_matches(|c: char| c <= ' ') }
66
67/// An `INT` as `ConfigDef` parses it.
68fn parse_int(value: &str) -> Option<i32> { java_trim(value).parse().ok() }
69
70/// A connection cap of at least `min`.
71fn parse_cap(value: &str, min: i32) -> Option<usize> {
72    let cap = parse_int(value).filter(|cap| *cap >= min)?;
73    usize::try_from(cap).ok()
74}
75
76/// A connection creation rate: an `INT` of at least 1, as
77/// `ListenerConnectionQuota.validateReconfiguration` requires of every
78/// listener, which inherits the broker-wide value.
79fn parse_rate(value: &str) -> Option<u32> {
80    parse_int(value)
81        .filter(|rate| *rate >= 1)
82        .map(i32::unsigned_abs)
83}
84
85/// A rate as a limit: Kafka's `Int.MaxValue` default is no limit.
86fn limited(rate: u32) -> Option<u32> { (rate != i32::MAX.unsigned_abs()).then_some(rate) }
87
88/// `Utils.validHostPattern`: `[0-9a-zA-Z\-%._:]*`.
89fn valid_host(host: &str) -> bool {
90    host.chars()
91        .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '%' | '.' | '_' | ':'))
92}
93
94/// `max.connections.per.ip.overrides` as Kafka reads it: `Csv.parseCsvMap`
95/// splits on commas and each entry on its last colon, `toInt` reads the
96/// count, and `KafkaConfig.validateValues` refuses a host outside
97/// `validHostPattern`. `None` is a value Kafka refuses. A negative count
98/// refuses every connection from the host, as `count >= max` always holds.
99pub(super) fn parse_overrides(value: &str) -> Option<BTreeMap<String, usize>> {
100    let value = java_trim(value);
101    if value.is_empty() {
102        return Some(BTreeMap::new());
103    }
104    // Java's `String.split` drops trailing empty entries and keeps the rest.
105    let mut entries: Vec<&str> = value.split(',').map(java_trim).collect();
106    while entries.last().is_some_and(|entry| entry.is_empty()) {
107        entries.pop();
108    }
109    entries
110        .into_iter()
111        .map(|entry| {
112            let (host, count) = entry.rsplit_once(':')?;
113            let host = java_trim(host);
114            let count: i32 = java_trim(count).parse().ok()?;
115            valid_host(host).then(|| (host.to_owned(), usize::try_from(count).unwrap_or(0)))
116        })
117        .collect()
118}
119
120/// The value of a key in force: the most preferred layer's value when it
121/// parses, else the `previous` value, else the next layer that parses, else
122/// `fallback`.
123fn pick<T>(
124    values: &[&str],
125    parse: impl Fn(&str) -> Option<T>,
126    previous: Option<T>,
127    fallback: T,
128) -> T {
129    let Some((first, rest)) = values.split_first() else {
130        return fallback;
131    };
132    parse(first)
133        .or(previous)
134        .or_else(|| rest.iter().find_map(|value| parse(value)))
135        .unwrap_or(fallback)
136}
137
138impl ConnectionQuotaSettings {
139    /// The quotas in force on `listener` of `node_id` as of `image`, over the
140    /// caps the node was started with and the `previous` settings of the
141    /// listener, which a refused value leaves in force.
142    #[must_use]
143    pub fn resolve(
144        image: &MetadataImage,
145        node_id: NodeId,
146        listener: &str,
147        configured: ConnectionLimits,
148        previous: Option<&Self>,
149    ) -> Self {
150        let layers = [image.broker_config(node_id), image.default_broker_config()];
151        // The values of `key`, most preferred layer first.
152        let value = |key: &str| -> Vec<&str> {
153            layers
154                .iter()
155                .filter_map(|configs| configs.and_then(|configs| configs.get(key)))
156                .map(String::as_str)
157                .collect()
158        };
159        // Every listener inherits `max.connections`, and Kafka refuses a
160        // listener cap below 1, so the broker-wide value needs 1 as well.
161        let max_connections = pick(
162            &value(MAX_CONNECTIONS),
163            |v| parse_cap(v, 1),
164            previous.map(|p| p.max_connections),
165            configured.max_connections,
166        );
167        let per_ip_overrides = pick(
168            &value(MAX_CONNECTIONS_PER_IP_OVERRIDES),
169            parse_overrides,
170            previous.map(|p| p.per_ip_overrides.clone()),
171            BTreeMap::new(),
172        );
173        // `KafkaConfig.validateValues` allows a per-IP cap of 0 only beside
174        // overrides.
175        let per_ip_min = i32::from(per_ip_overrides.is_empty());
176        let max_connections_per_ip = pick(
177            &value(MAX_CONNECTIONS_PER_IP),
178            |v| parse_cap(v, per_ip_min),
179            previous.map(|p| p.max_connections_per_ip),
180            configured.max_connections_per_ip,
181        );
182        let listener_max_connections = pick(
183            &value(&listener_key(listener, MAX_CONNECTIONS)),
184            |v| parse_cap(v, 1),
185            previous.map(|p| p.listener_max_connections),
186            max_connections,
187        );
188        let max_connection_creation_rate = pick(
189            &value(MAX_CONNECTION_CREATION_RATE),
190            |v| parse_rate(v).map(limited),
191            previous.map(|p| p.max_connection_creation_rate),
192            None,
193        );
194        let listener_max_connection_creation_rate = pick(
195            &value(&listener_key(listener, MAX_CONNECTION_CREATION_RATE)),
196            |v| parse_rate(v).map(limited),
197            previous.map(|p| p.listener_max_connection_creation_rate),
198            max_connection_creation_rate,
199        );
200        Self {
201            max_connections,
202            max_connections_per_ip,
203            per_ip_overrides,
204            listener_max_connections,
205            max_connection_creation_rate,
206            listener_max_connection_creation_rate,
207        }
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214
215    /// Apply one dynamic broker config, for node 1 or the cluster default.
216    fn set_broker_config(image: &mut MetadataImage, node: Option<u64>, name: &str, value: &str) {
217        image.apply(&krabka_metadata::MetadataRecord::V1BrokerConfig(
218            krabka_metadata::BrokerConfigRecord {
219                node_id: node.map_or(krabka_metadata::DEFAULT_BROKER_CONFIG_NODE_ID, NodeId),
220                config_name: name.into(),
221                config_value: Some(value.into()),
222            },
223        ));
224    }
225
226    const CONFIGURED: ConnectionLimits = ConnectionLimits {
227        max_connections: 10,
228        max_connections_per_ip: 5,
229    };
230
231    /// The settings with nothing dynamic, over [`CONFIGURED`].
232    fn unset() -> ConnectionQuotaSettings {
233        ConnectionQuotaSettings {
234            max_connections: 10,
235            max_connections_per_ip: 5,
236            per_ip_overrides: BTreeMap::new(),
237            listener_max_connections: 10,
238            max_connection_creation_rate: None,
239            listener_max_connection_creation_rate: None,
240        }
241    }
242
243    fn overrides(entries: &[(&str, usize)]) -> BTreeMap<String, usize> {
244        entries
245            .iter()
246            .map(|(host, count)| ((*host).to_owned(), *count))
247            .collect()
248    }
249
250    #[yare::parameterized(
251        static_only = { &[], unset() },
252        cluster_default = {
253            &[(None, MAX_CONNECTIONS, "3")],
254            ConnectionQuotaSettings { max_connections: 3, listener_max_connections: 3, ..unset() }
255        },
256        per_broker_over_default = {
257            &[(None, MAX_CONNECTIONS_PER_IP, "2"), (Some(1), MAX_CONNECTIONS_PER_IP, "4")],
258            ConnectionQuotaSettings { max_connections_per_ip: 4, ..unset() }
259        },
260        other_broker_ignored = { &[(Some(2), MAX_CONNECTIONS, "1")], unset() },
261        zero_global_not_applied = { &[(None, MAX_CONNECTIONS, "0")], unset() },
262        trimmed = {
263            &[(None, MAX_CONNECTIONS, " 7 ")],
264            ConnectionQuotaSettings { max_connections: 7, listener_max_connections: 7, ..unset() }
265        },
266        unparsable_falls_back_a_layer = {
267            &[(None, MAX_CONNECTIONS, "2"), (Some(1), MAX_CONNECTIONS, "lots")],
268            ConnectionQuotaSettings { max_connections: 2, listener_max_connections: 2, ..unset() }
269        },
270        zero_per_ip_needs_overrides = { &[(None, MAX_CONNECTIONS_PER_IP, "0")], unset() },
271        zero_per_ip_beside_overrides = {
272            &[(None, MAX_CONNECTIONS_PER_IP, "0"), (None, MAX_CONNECTIONS_PER_IP_OVERRIDES, "127.0.0.1:2")],
273            ConnectionQuotaSettings {
274                max_connections_per_ip: 0,
275                per_ip_overrides: overrides(&[("127.0.0.1", 2)]),
276                ..unset()
277            }
278        },
279        listener_cap_over_broker_wide = {
280            &[(None, MAX_CONNECTIONS, "8"), (Some(1), "listener.name.plaintext.max.connections", "3")],
281            ConnectionQuotaSettings { max_connections: 8, listener_max_connections: 3, ..unset() }
282        },
283        other_listener_ignored = {
284            &[(None, "listener.name.external.max.connections", "3")],
285            unset()
286        },
287        zero_listener_cap_not_applied = {
288            &[(None, "listener.name.plaintext.max.connections", "0")],
289            unset()
290        },
291        broker_rate = {
292            &[(None, MAX_CONNECTION_CREATION_RATE, "20")],
293            ConnectionQuotaSettings {
294                max_connection_creation_rate: Some(20),
295                listener_max_connection_creation_rate: Some(20),
296                ..unset()
297            }
298        },
299        listener_rate_over_broker_rate = {
300            &[
301                (None, MAX_CONNECTION_CREATION_RATE, "20"),
302                (None, "listener.name.plaintext.max.connection.creation.rate", "5"),
303            ],
304            ConnectionQuotaSettings {
305                max_connection_creation_rate: Some(20),
306                listener_max_connection_creation_rate: Some(5),
307                ..unset()
308            }
309        },
310        int_max_rate_is_unlimited = {
311            &[(None, MAX_CONNECTION_CREATION_RATE, "2147483647")],
312            unset()
313        },
314        zero_rate_not_applied = { &[(None, MAX_CONNECTION_CREATION_RATE, "0")], unset() },
315    )]
316    fn settings_layer_broker_default_listener_and_static(
317        configs: &[(Option<u64>, &str, &str)],
318        want: ConnectionQuotaSettings,
319    ) {
320        let mut image = MetadataImage::new(uuid::Uuid::nil());
321        for (node, name, value) in configs {
322            set_broker_config(&mut image, *node, name, value);
323        }
324        assert2::assert!(
325            ConnectionQuotaSettings::resolve(&image, NodeId(1), "PLAINTEXT", CONFIGURED, None)
326                == want
327        );
328    }
329
330    /// A refused value keeps the value in force before it, which an unset
331    /// key does not: after `max.connections=2`, a stored 0 keeps 2, and a
332    /// deleted key returns to the static cap.
333    #[yare::parameterized(
334        refused_keeps_previous = { Some("0"), 2 },
335        unparsable_keeps_previous = { Some("lots"), 2 },
336        deleted_returns_to_static = { None, 10 },
337        valid_replaces_previous = { Some("4"), 4 },
338    )]
339    fn a_refused_value_keeps_the_previous_one(next: Option<&str>, want: usize) {
340        let mut image = MetadataImage::new(uuid::Uuid::nil());
341        set_broker_config(&mut image, None, MAX_CONNECTIONS, "2");
342        let previous =
343            ConnectionQuotaSettings::resolve(&image, NodeId(1), "PLAINTEXT", CONFIGURED, None);
344        let mut image = MetadataImage::new(uuid::Uuid::nil());
345        if let Some(value) = next {
346            set_broker_config(&mut image, None, MAX_CONNECTIONS, value);
347        }
348        let got = ConnectionQuotaSettings::resolve(
349            &image,
350            NodeId(1),
351            "PLAINTEXT",
352            CONFIGURED,
353            Some(&previous),
354        );
355        assert2::assert!((got.max_connections, got.listener_max_connections) == (want, want));
356    }
357
358    #[yare::parameterized(
359        empty = { "", Some(overrides(&[])) },
360        one = { "127.0.0.1:2", Some(overrides(&[("127.0.0.1", 2)])) },
361        spaced = { " a.example : 3 , b:4 ", Some(overrides(&[("a.example", 3), ("b", 4)])) },
362        ipv6_splits_on_last_colon = { "::1:5", Some(overrides(&[("::1", 5)])) },
363        trailing_comma = { "a:1,", Some(overrides(&[("a", 1)])) },
364        last_duplicate_wins = { "a:1,a:2", Some(overrides(&[("a", 2)])) },
365        negative_refuses_all = { "a:-1", Some(overrides(&[("a", 0)])) },
366        leading_comma = { ",a:1", None },
367        no_colon = { "a", None },
368        not_a_number = { "a:x", None },
369        bad_host = { "a/b:1", None },
370    )]
371    fn overrides_parse_as_kafka_parses_them(value: &str, want: Option<BTreeMap<String, usize>>) {
372        assert2::assert!(parse_overrides(value) == want);
373    }
374}
krabka_raft/connection_limiter/settings.rs

krabka_raft/connection_limiter/
settings.rs
