krabka_replicator/
config.rs

1//! Replicator configuration model (clusters / flows / policies).
2
3use std::{collections::BTreeMap, num::NonZeroUsize, str::FromStr};
4
5use clap::Args;
6use krabka_units::{Time, millis, secs};
7use refined_type::rule::GreaterI16;
8use serde::{Deserialize, Serialize};
9
10use crate::error::ReplicatorError;
11
12/// Kafka client resource policy owned by the replicator process.
13#[derive(Debug, Clone, Copy, Default)]
14pub struct ClientResourcePolicy {
15    pub dispatch_queue_capacity: krabka_client_core::ConnectionDispatchQueueCapacity,
16    pub frame_max: krabka_client_core::ClientFrameMax,
17}
18
19type RefinedReplicationFactor = GreaterI16<0>;
20
21/// Positive Kafka replication factor accepted by the replicator.
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub struct ReplicationFactor(i16);
24
25impl Default for ReplicationFactor {
26    fn default() -> Self {
27        Self(1)
28    }
29}
30
31impl ReplicationFactor {
32    /// Validate a Kafka replication factor.
33    ///
34    /// # Errors
35    /// Returns an error when the value is not positive.
36    pub fn new(value: i16) -> Result<Self, String> {
37        RefinedReplicationFactor::new(value)
38            .map(|value| Self(value.into_value()))
39            .map_err(|error| format!("replication factor: {error}"))
40    }
41
42    /// Return the Kafka protocol value.
43    #[must_use]
44    pub const fn get(self) -> i16 {
45        self.0
46    }
47}
48
49impl FromStr for ReplicationFactor {
50    type Err = String;
51
52    fn from_str(value: &str) -> Result<Self, Self::Err> {
53        Self::new(value.parse::<i16>().map_err(|error| error.to_string())?)
54    }
55}
56
57/// Process-owned replicator runtime and topic policy.
58#[derive(Args, Debug, Clone, PartialEq)]
59pub struct ReplicatorRuntimePolicy {
60    #[arg(long, env = "KRABKA_REPLICATOR_TOPIC_CREATE_TIMEOUT", default_value = "10s", value_parser = krabka_units::parse::positive_time)]
61    pub topic_create_timeout: Time,
62    #[arg(long, env = "KRABKA_REPLICATOR_SOURCE_POLL_TIMEOUT", default_value = "500ms", value_parser = krabka_units::parse::positive_time)]
63    pub source_poll_timeout: Time,
64    #[arg(long, env = "KRABKA_REPLICATOR_INTERNAL_DRAIN_POLL_TIMEOUT", default_value = "500ms", value_parser = krabka_units::parse::positive_time)]
65    pub internal_drain_poll_timeout: Time,
66    #[arg(
67        long,
68        env = "KRABKA_REPLICATOR_INTERNAL_DRAIN_EMPTY_POLLS",
69        default_value = "3"
70    )]
71    pub internal_drain_empty_polls: NonZeroUsize,
72    #[arg(long, env = "KRABKA_REPLICATOR_WORKER_BUILD_RETRY_BUDGET", default_value = "30s", value_parser = krabka_units::parse::positive_time)]
73    pub worker_build_retry_budget: Time,
74    #[arg(long, env = "KRABKA_REPLICATOR_WORKER_BUILD_INITIAL_BACKOFF", default_value = "250ms", value_parser = krabka_units::parse::positive_time)]
75    pub worker_build_initial_backoff: Time,
76    #[arg(long, env = "KRABKA_REPLICATOR_WORKER_BUILD_MAX_BACKOFF", default_value = "8s", value_parser = krabka_units::parse::positive_time)]
77    pub worker_build_max_backoff: Time,
78    #[arg(long, env = "KRABKA_REPLICATOR_CONNECT_COMMIT_INTERVAL", default_value = "500ms", value_parser = krabka_units::parse::positive_time)]
79    pub connect_commit_interval: Time,
80    #[arg(
81        long,
82        env = "KRABKA_REPLICATOR_CONNECT_MAX_BATCH_RECORDS",
83        default_value = "500"
84    )]
85    pub connect_max_batch_records: NonZeroUsize,
86    #[arg(long, env = "KRABKA_REPLICATOR_SUPERVISOR_INTERVAL", default_value = "3s", value_parser = krabka_units::parse::positive_time)]
87    pub supervisor_interval: Time,
88    #[arg(long, env = "KRABKA_REPLICATOR_HEARTBEAT_INTERVAL", default_value = "1s", value_parser = krabka_units::parse::positive_time)]
89    pub heartbeat_interval: Time,
90    #[arg(long, env = "KRABKA_REPLICATOR_CHECKPOINT_INTERVAL", default_value = "5s", value_parser = krabka_units::parse::positive_time)]
91    pub checkpoint_interval: Time,
92    #[arg(long, env = "KRABKA_REPLICATOR_CLIENT_DNS_TIMEOUT", default_value = "10s", value_parser = krabka_units::parse::positive_time)]
93    pub client_dns_timeout: Time,
94    #[arg(long, env = "KRABKA_REPLICATOR_CLIENT_CONNECT_TIMEOUT", default_value = "5s", value_parser = krabka_units::parse::positive_time)]
95    pub client_connect_timeout: Time,
96    #[arg(long, env = "KRABKA_REPLICATOR_CLIENT_REQUEST_TIMEOUT", default_value = "30s", value_parser = krabka_units::parse::positive_time)]
97    pub client_request_timeout: Time,
98    #[arg(
99        long,
100        env = "KRABKA_REPLICATOR_DATA_TOPIC_REPLICATION_FACTOR",
101        default_value = "1"
102    )]
103    pub data_topic_replication_factor: ReplicationFactor,
104    #[arg(
105        long,
106        env = "KRABKA_REPLICATOR_INTERNAL_TOPIC_REPLICATION_FACTOR",
107        default_value = "1"
108    )]
109    pub internal_topic_replication_factor: ReplicationFactor,
110}
111
112impl ReplicatorRuntimePolicy {
113    /// Validate relationships between independently parsed policy values.
114    ///
115    /// # Errors
116    /// Returns an error when retry bounds are inconsistent or the DNS timeout
117    /// cannot be represented by the client boundary.
118    pub fn validate(&self) -> Result<(), String> {
119        if self.worker_build_initial_backoff > self.worker_build_max_backoff {
120            return Err("worker build initial backoff exceeds maximum".to_owned());
121        }
122        if self.worker_build_retry_budget < self.worker_build_initial_backoff {
123            return Err("worker build retry budget is below initial backoff".to_owned());
124        }
125        krabka_client_core::ClientDnsTimeout::new(self.client_dns_timeout)?;
126        Ok(())
127    }
128}
129
130impl Default for ReplicatorRuntimePolicy {
131    fn default() -> Self {
132        Self {
133            topic_create_timeout: secs(10),
134            source_poll_timeout: millis(500),
135            internal_drain_poll_timeout: millis(500),
136            internal_drain_empty_polls: NonZeroUsize::new(3).expect("default is positive"),
137            worker_build_retry_budget: secs(30),
138            worker_build_initial_backoff: millis(250),
139            worker_build_max_backoff: secs(8),
140            connect_commit_interval: millis(500),
141            connect_max_batch_records: NonZeroUsize::new(500).expect("default is positive"),
142            supervisor_interval: secs(3),
143            heartbeat_interval: secs(1),
144            checkpoint_interval: secs(5),
145            client_dns_timeout: secs(10),
146            client_connect_timeout: secs(5),
147            client_request_timeout: secs(30),
148            data_topic_replication_factor: ReplicationFactor(1),
149            internal_topic_replication_factor: ReplicationFactor(1),
150        }
151    }
152}
153
154/// Top-level replicator configuration.
155#[derive(Debug, Clone, Serialize, Deserialize)]
156pub struct ReplicatorConfig {
157    /// Named cluster definitions keyed by alias.
158    pub clusters: BTreeMap<String, ClusterConfig>,
159    /// Directional replication flows.
160    pub flows: Vec<FlowConfig>,
161    /// Residency / compliance policies.
162    #[serde(default)]
163    pub policies: Vec<PolicyConfig>,
164}
165
166/// Configuration for a single Kafka cluster.
167#[derive(Debug, Clone, Serialize, Deserialize)]
168pub struct ClusterConfig {
169    /// Bootstrap server address (host:port).
170    pub bootstrap: String,
171    /// Logical region name (e.g. `"us"`, `"eu"`).
172    pub region: String,
173    /// Availability / compliance zones this cluster belongs to.
174    #[serde(default)]
175    pub zones: Vec<String>,
176}
177
178/// A directional replication flow from one cluster to another.
179#[derive(Debug, Clone, Serialize, Deserialize)]
180pub struct FlowConfig {
181    /// Source cluster alias.
182    pub from: String,
183    /// Target cluster alias.
184    pub to: String,
185    /// Topic selector (include / exclude glob patterns).
186    #[serde(default)]
187    pub topics: Selectors,
188    /// Consumer-group selector (include / exclude glob patterns).
189    #[serde(default)]
190    pub groups: Selectors,
191    /// How replicated topics are named on the target cluster.
192    #[serde(default)]
193    pub naming: NamingPolicy,
194    /// Delivery guarantee.
195    #[serde(default)]
196    pub delivery: Delivery,
197}
198
199/// Include / exclude glob-pattern selector lists.
200#[derive(Debug, Clone, Default, Serialize, Deserialize)]
201pub struct Selectors {
202    /// Patterns that must match.
203    #[serde(default)]
204    pub include: Vec<String>,
205    /// Patterns that cause a match to be skipped.
206    #[serde(default)]
207    pub exclude: Vec<String>,
208}
209
210/// How replicated topic names are derived on the target cluster.
211#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
212#[serde(rename_all = "kebab-case")]
213pub enum NamingPolicy {
214    /// Prefix topic name with the source cluster alias (`MirrorMaker` 2 default).
215    #[default]
216    Default,
217    /// Keep the original topic name unchanged.
218    Identity,
219}
220
221/// Delivery guarantee for a flow.
222#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
223#[serde(rename_all = "kebab-case")]
224pub enum Delivery {
225    /// At-least-once delivery.
226    #[default]
227    AtLeastOnce,
228    /// Exactly-once target writes and source checkpoints.
229    ExactlyOnce,
230}
231
232/// A residency / compliance policy that constrains where topic data may land.
233#[derive(Debug, Clone, Serialize, Deserialize)]
234pub struct PolicyConfig {
235    /// Human-readable policy name.
236    pub name: String,
237    /// Topics this policy applies to.
238    #[serde(default)]
239    pub topics: Vec<String>,
240    /// Zone residency constraints.
241    #[serde(default)]
242    pub residency: Option<Residency>,
243}
244
245/// Zone-based data-residency constraints.
246#[derive(Debug, Clone, Default, Serialize, Deserialize)]
247pub struct Residency {
248    /// Zones that data may flow to.
249    #[serde(default)]
250    pub allow_zones: Vec<String>,
251    /// Zones to which data must not flow.
252    #[serde(default)]
253    pub deny_zones: Vec<String>,
254}
255
256impl ReplicatorConfig {
257    /// Parse a [`ReplicatorConfig`] from a YAML string.
258    ///
259    /// # Errors
260    ///
261    /// Returns [`ReplicatorError::Config`] if the YAML is malformed or fails
262    /// to deserialize.
263    pub fn from_yaml(s: &str) -> Result<Self, ReplicatorError> {
264        serde_yaml::from_str(s).map_err(|e| ReplicatorError::Config(e.to_string()))
265    }
266
267    /// Validate a parsed config.
268    ///
269    /// # Errors
270    ///
271    /// Returns [`ReplicatorError::Config`] if:
272    /// - A flow references an unknown cluster alias.
273    pub fn validate(&self) -> Result<(), ReplicatorError> {
274        for f in &self.flows {
275            if !self.clusters.contains_key(&f.from) {
276                return Err(ReplicatorError::Config(format!(
277                    "flow.from unknown cluster `{}`",
278                    f.from
279                )));
280            }
281            if !self.clusters.contains_key(&f.to) {
282                return Err(ReplicatorError::Config(format!(
283                    "flow.to unknown cluster `{}`",
284                    f.to
285                )));
286            }
287        }
288        Ok(())
289    }
290}
291
292#[cfg(test)]
293mod tests {
294
295    use super::*;
296
297    const YAML: &str = r#"
298clusters:
299  us-east: { bootstrap: "127.0.0.1:9092", region: us, zones: [us] }
300  eu-west: { bootstrap: "127.0.0.1:9093", region: eu, zones: [eu, gdpr] }
301flows:
302  - from: us-east
303    to: eu-west
304    topics: { include: ["orders", "telemetry.*"], exclude: ["*.internal"] }
305    groups: { include: ["analytics-*"] }
306    naming: default
307    delivery: at-least-once
308policies:
309  - name: keep-pii-in-eu
310    topics: ["customers"]
311    residency: { allow_zones: [gdpr] }
312"#;
313
314    #[test]
315    fn parses_and_validates() {
316        let cfg = ReplicatorConfig::from_yaml(YAML).unwrap();
317        assert2::assert!(cfg.clusters.len() == 2);
318        assert2::assert!(
319            cfg.clusters["eu-west"].zones.clone() == vec!["eu".to_string(), "gdpr".to_string()]
320        );
321        assert2::assert!(cfg.flows[0].naming == NamingPolicy::Default);
322        assert2::assert!(cfg.flows[0].delivery == Delivery::AtLeastOnce);
323        cfg.validate().unwrap();
324    }
325
326    #[test]
327    fn accepts_exactly_once_delivery() {
328        let y = YAML.replace("delivery: at-least-once", "delivery: exactly-once");
329        let cfg = ReplicatorConfig::from_yaml(&y).unwrap();
330        assert2::assert!(cfg.flows[0].delivery == Delivery::ExactlyOnce);
331        cfg.validate().unwrap();
332    }
333
334    #[test]
335    fn rejects_unknown_cluster_ref() {
336        let y = YAML.replace("to: eu-west", "to: nowhere");
337        let cfg = ReplicatorConfig::from_yaml(&y).unwrap();
338        assert2::assert!(cfg.validate().is_err());
339    }
340
341    #[test]
342    fn runtime_policy_defaults_preserve_existing_behavior() {
343        let policy = ReplicatorRuntimePolicy::default();
344        assert2::assert!(policy.topic_create_timeout == secs(10));
345        assert2::assert!(policy.source_poll_timeout == millis(500));
346        assert2::assert!(policy.internal_drain_empty_polls.get() == 3);
347        assert2::assert!(policy.worker_build_retry_budget == secs(30));
348        assert2::assert!(policy.worker_build_initial_backoff == millis(250));
349        assert2::assert!(policy.worker_build_max_backoff == secs(8));
350        assert2::assert!(policy.connect_max_batch_records.get() == 500);
351        assert2::assert!(policy.supervisor_interval == secs(3));
352        assert2::assert!(policy.heartbeat_interval == secs(1));
353        assert2::assert!(policy.checkpoint_interval == secs(5));
354        assert2::assert!(policy.client_dns_timeout == secs(10));
355        assert2::assert!(policy.data_topic_replication_factor.get() == 1);
356        policy.validate().unwrap();
357    }
358
359    #[test]
360    fn runtime_policy_rejects_invalid_relations_and_replication_factors() {
361        let policy = ReplicatorRuntimePolicy {
362            worker_build_initial_backoff: secs(9),
363            ..ReplicatorRuntimePolicy::default()
364        };
365        assert2::assert!(policy.validate().is_err());
366
367        let policy = ReplicatorRuntimePolicy {
368            worker_build_retry_budget: millis(100),
369            ..ReplicatorRuntimePolicy::default()
370        };
371        assert2::assert!(policy.validate().is_err());
372        assert2::assert!("0".parse::<ReplicationFactor>().is_err());
373        assert2::assert!("2".parse::<ReplicationFactor>().unwrap().get() == 2);
374    }
375}
krabka_replicator/config.rs

krabka_replicator/
config.rs
