klukai_types/
config.rs

1use std::net::{Ipv6Addr, SocketAddr, SocketAddrV6};
2
3use camino::Utf8PathBuf;
4use serde::{Deserialize, Serialize};
5use serde_with::{OneOrMany, formats::PreferOne, serde_as};
6
7pub const DEFAULT_GOSSIP_PORT: u16 = 4001;
8const DEFAULT_GOSSIP_IDLE_TIMEOUT: u32 = 30;
9
10#[cfg(test)]
11pub const DEFAULT_MAX_SYNC_BACKOFF: u32 = 2;
12#[cfg(not(test))]
13pub const DEFAULT_MAX_SYNC_BACKOFF: u32 = 15;
14
15const fn default_apply_queue() -> usize {
16    50
17}
18
19const fn default_wal_threshold() -> usize {
20    5
21}
22
23const fn default_processing_queue() -> usize {
24    20000
25}
26
27/// Used for the apply channel
28const fn default_huge_channel() -> usize {
29    2048
30}
31
32//
33const fn default_big_channel() -> usize {
34    1024
35}
36
37const fn default_mid_channel() -> usize {
38    512
39}
40
41const fn default_small_channel() -> usize {
42    256
43}
44
45const fn default_apply_timeout() -> usize {
46    10
47}
48
49fn default_sql_tx_timeout() -> usize {
50    60
51}
52
53fn default_min_sync_backoff() -> u32 {
54    1
55}
56
57fn default_max_sync_backoff() -> u32 {
58    DEFAULT_MAX_SYNC_BACKOFF
59}
60
61#[derive(Debug, Clone, Serialize, Deserialize)]
62pub struct Config {
63    pub db: DbConfig,
64    pub api: ApiConfig,
65    pub gossip: GossipConfig,
66
67    #[serde(default)]
68    pub perf: PerfConfig,
69
70    #[serde(default)]
71    pub admin: AdminConfig,
72
73    #[serde(default)]
74    pub telemetry: TelemetryConfig,
75
76    #[serde(default)]
77    pub log: LogConfig,
78
79    #[serde(default)]
80    pub consul: Option<ConsulConfig>,
81}
82
83#[derive(Debug, Default, Clone, Serialize, Deserialize)]
84#[serde(rename_all = "kebab-case")]
85pub struct TelemetryConfig {
86    pub prometheus: Option<PrometheusConfig>,
87    pub open_telemetry: Option<OtelConfig>,
88}
89
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct PrometheusConfig {
92    #[serde(alias = "addr")]
93    pub bind_addr: SocketAddr,
94}
95
96#[derive(Debug, Clone, Serialize, Deserialize)]
97#[serde(rename_all = "kebab-case")]
98pub enum OtelConfig {
99    FromEnv,
100    Exporter { endpoint: String },
101}
102
103#[derive(Debug, Clone, Serialize, Deserialize)]
104pub struct AdminConfig {
105    #[serde(alias = "path")]
106    pub uds_path: Utf8PathBuf,
107}
108
109impl Default for AdminConfig {
110    fn default() -> Self {
111        Self {
112            uds_path: default_admin_path(),
113        }
114    }
115}
116
117#[derive(Debug, Clone, Serialize, Deserialize)]
118pub struct DbConfig {
119    pub path: Utf8PathBuf,
120    #[serde(default)]
121    pub schema_paths: Vec<Utf8PathBuf>,
122    #[serde(default)]
123    pub subscriptions_path: Option<Utf8PathBuf>,
124}
125
126impl DbConfig {
127    pub fn subscriptions_path(&self) -> Utf8PathBuf {
128        self.subscriptions_path
129            .as_ref()
130            .cloned()
131            .unwrap_or_else(|| {
132                self.path
133                    .parent()
134                    .map(|parent| parent.join("subscriptions"))
135                    .unwrap_or_else(|| "/subscriptions".into())
136            })
137    }
138}
139
140#[serde_as]
141#[derive(Debug, Clone, Serialize, Deserialize)]
142pub struct ApiConfig {
143    #[serde(alias = "addr")]
144    #[serde_as(deserialize_as = "OneOrMany<_, PreferOne>")]
145    pub bind_addr: Vec<SocketAddr>,
146    #[serde(alias = "authz", default)]
147    pub authorization: Option<AuthzConfig>,
148}
149
150#[derive(Debug, Clone, Serialize, Deserialize)]
151#[serde(rename_all = "kebab-case")]
152pub enum AuthzConfig {
153    #[serde(alias = "bearer")]
154    BearerToken(String),
155}
156
157#[derive(Debug, Clone, Serialize, Deserialize)]
158pub struct GossipConfig {
159    #[serde(alias = "addr")]
160    pub bind_addr: SocketAddr,
161    pub external_addr: Option<SocketAddr>,
162    #[serde(default = "default_gossip_client_addr")]
163    pub client_addr: SocketAddr,
164    #[serde(default)]
165    pub bootstrap: Vec<String>,
166    #[serde(default)]
167    pub tls: Option<TlsConfig>,
168    #[serde(default)]
169    pub plaintext: bool,
170    #[serde(default)]
171    pub max_mtu: Option<u16>,
172    #[serde(default = "default_gossip_idle_timeout")]
173    pub idle_timeout_secs: u32,
174    #[serde(default)]
175    pub disable_gso: bool,
176}
177
178#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct PerfConfig {
180    #[serde(default = "default_huge_channel")]
181    pub apply_channel_len: usize,
182    #[serde(default = "default_big_channel")]
183    pub changes_channel_len: usize,
184    #[serde(default = "default_big_channel")]
185    pub empties_channel_len: usize,
186    #[serde(default = "default_mid_channel")]
187    pub to_send_channel_len: usize,
188    #[serde(default = "default_mid_channel")]
189    pub notifications_channel_len: usize,
190    #[serde(default = "default_mid_channel")]
191    pub schedule_channel_len: usize,
192    #[serde(default = "default_mid_channel")]
193    pub clearbuf_channel_len: usize,
194    #[serde(default = "default_mid_channel")]
195    pub bcast_channel_len: usize,
196    #[serde(default = "default_small_channel")]
197    pub foca_channel_len: usize,
198    #[serde(default = "default_apply_timeout")]
199    pub apply_queue_timeout: usize,
200    #[serde(default = "default_apply_queue")]
201    pub apply_queue_len: usize,
202    #[serde(default = "default_wal_threshold")]
203    pub wal_threshold_gb: usize,
204    #[serde(default = "default_processing_queue")]
205    pub processing_queue_len: usize,
206    #[serde(default = "default_sql_tx_timeout")]
207    pub sql_tx_timeout: usize,
208    #[serde(default = "default_min_sync_backoff")]
209    pub min_sync_backoff: u32,
210    #[serde(default = "default_max_sync_backoff")]
211    pub max_sync_backoff: u32,
212}
213
214impl Default for PerfConfig {
215    fn default() -> Self {
216        Self {
217            apply_channel_len: default_huge_channel(),
218            changes_channel_len: default_big_channel(),
219            empties_channel_len: default_big_channel(),
220            to_send_channel_len: default_mid_channel(),
221            notifications_channel_len: default_mid_channel(),
222            schedule_channel_len: default_mid_channel(),
223            clearbuf_channel_len: default_mid_channel(),
224            bcast_channel_len: default_mid_channel(),
225            foca_channel_len: default_small_channel(),
226            apply_queue_timeout: default_apply_timeout(),
227            apply_queue_len: default_apply_queue(),
228            wal_threshold_gb: default_wal_threshold(),
229            processing_queue_len: default_processing_queue(),
230            sql_tx_timeout: default_sql_tx_timeout(),
231            min_sync_backoff: default_min_sync_backoff(),
232            max_sync_backoff: default_max_sync_backoff(),
233        }
234    }
235}
236
237fn default_gossip_idle_timeout() -> u32 {
238    DEFAULT_GOSSIP_IDLE_TIMEOUT
239}
240
241pub const DEFAULT_GOSSIP_CLIENT_ADDR: SocketAddr =
242    SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::UNSPECIFIED, 0u16, 0, 0));
243
244fn default_gossip_client_addr() -> SocketAddr {
245    DEFAULT_GOSSIP_CLIENT_ADDR
246}
247
248#[derive(Debug, Clone, Serialize, Deserialize)]
249pub struct TlsConfig {
250    /// Certificate file
251    pub cert_file: Utf8PathBuf,
252    /// Private key file
253    pub key_file: Utf8PathBuf,
254
255    /// CA (Certificate Authority) file
256    #[serde(default)]
257    pub ca_file: Option<Utf8PathBuf>,
258
259    #[serde(default)]
260    pub insecure: bool,
261
262    /// Mutual TLS configuration
263    #[serde(default)]
264    pub client: Option<TlsClientConfig>,
265}
266
267#[derive(Debug, Clone, Serialize, Deserialize)]
268pub struct TlsClientConfig {
269    /// Certificate file
270    pub cert_file: Utf8PathBuf,
271    /// Private key file
272    pub key_file: Utf8PathBuf,
273}
274
275pub fn default_admin_path() -> Utf8PathBuf {
276    "/var/run/corrosion/admin.sock".into()
277}
278
279#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
280pub struct LogConfig {
281    #[serde(default)]
282    pub format: LogFormat,
283    #[serde(default = "default_as_true")]
284    pub colors: bool,
285}
286
287fn default_as_true() -> bool {
288    true
289}
290
291#[derive(Debug, thiserror::Error)]
292pub enum ConfigError {
293    #[error(transparent)]
294    Config(#[from] config::ConfigError),
295}
296
297impl Config {
298    pub fn builder() -> ConfigBuilder {
299        ConfigBuilder::default()
300    }
301
302    /// Reads configuration from a TOML file, given its path. Environment
303    /// variables can override whatever is set in the config file.
304    pub fn load(config_path: &str) -> Result<Self, ConfigError> {
305        let config = config::Config::builder()
306            .add_source(config::File::new(config_path, config::FileFormat::Toml))
307            .add_source(config::Environment::default().separator("__"))
308            .build()?;
309        Ok(config.try_deserialize()?)
310    }
311}
312
313#[derive(Debug, Default)]
314pub struct ConfigBuilder {
315    pub db_path: Option<Utf8PathBuf>,
316    gossip_addr: Option<SocketAddr>,
317    api_addr: Vec<SocketAddr>,
318    external_addr: Option<SocketAddr>,
319    admin_path: Option<Utf8PathBuf>,
320    prometheus_addr: Option<SocketAddr>,
321    bootstrap: Option<Vec<String>>,
322    log: Option<LogConfig>,
323    schema_paths: Vec<Utf8PathBuf>,
324    max_change_size: Option<i64>,
325    tls: Option<TlsConfig>,
326    perf: Option<PerfConfig>,
327    consul: Option<ConsulConfig>,
328}
329
330impl ConfigBuilder {
331    pub fn db_path<S: Into<Utf8PathBuf>>(mut self, db_path: S) -> Self {
332        self.db_path = Some(db_path.into());
333        self
334    }
335
336    pub fn gossip_addr(mut self, addr: SocketAddr) -> Self {
337        self.gossip_addr = Some(addr);
338        self
339    }
340
341    pub fn api_addr(mut self, addr: SocketAddr) -> Self {
342        self.api_addr.push(addr);
343        self
344    }
345
346    pub fn prometheus_addr(mut self, addr: SocketAddr) -> Self {
347        self.prometheus_addr = Some(addr);
348        self
349    }
350
351    pub fn bootstrap<V: Into<Vec<String>>>(mut self, bootstrap: V) -> Self {
352        self.bootstrap = Some(bootstrap.into());
353        self
354    }
355
356    pub fn log(mut self, log: LogConfig) -> Self {
357        self.log = Some(log);
358        self
359    }
360
361    pub fn add_schema_path<S: Into<Utf8PathBuf>>(mut self, path: S) -> Self {
362        self.schema_paths.push(path.into());
363        self
364    }
365
366    pub fn admin_path<S: Into<Utf8PathBuf>>(mut self, path: S) -> Self {
367        self.admin_path = Some(path.into());
368        self
369    }
370
371    pub fn max_change_size(mut self, size: i64) -> Self {
372        self.max_change_size = Some(size);
373        self
374    }
375
376    pub fn tls_config(mut self, config: TlsConfig) -> Self {
377        self.tls = Some(config);
378        self
379    }
380
381    pub fn consul(mut self, config: ConsulConfig) -> Self {
382        self.consul = Some(config);
383        self
384    }
385
386    pub fn build(self) -> Result<Config, ConfigBuilderError> {
387        let db_path = self.db_path.ok_or(ConfigBuilderError::DbPathRequired)?;
388
389        let telemetry = TelemetryConfig {
390            prometheus: self
391                .prometheus_addr
392                .map(|bind_addr| PrometheusConfig { bind_addr }),
393            open_telemetry: None,
394        };
395
396        if self.api_addr.is_empty() {
397            return Err(ConfigBuilderError::ApiAddrRequired);
398        }
399
400        Ok(Config {
401            db: DbConfig {
402                path: db_path,
403                schema_paths: self.schema_paths,
404                subscriptions_path: None,
405            },
406            api: ApiConfig {
407                bind_addr: self.api_addr,
408                authorization: None,
409            },
410            gossip: GossipConfig {
411                bind_addr: self
412                    .gossip_addr
413                    .ok_or(ConfigBuilderError::GossipAddrRequired)?,
414                external_addr: self.external_addr,
415                client_addr: default_gossip_client_addr(),
416                bootstrap: self.bootstrap.unwrap_or_default(),
417                plaintext: self.tls.is_none(),
418                tls: self.tls,
419                idle_timeout_secs: default_gossip_idle_timeout(),
420                max_mtu: None, // TODO: add a builder function for it
421                disable_gso: false,
422            },
423            perf: self.perf.unwrap_or_default(),
424            admin: AdminConfig {
425                uds_path: self.admin_path.unwrap_or_else(default_admin_path),
426            },
427            telemetry,
428            log: self.log.unwrap_or_default(),
429            consul: self.consul,
430        })
431    }
432}
433
434#[derive(Debug, thiserror::Error)]
435pub enum ConfigBuilderError {
436    #[error("db.path required")]
437    DbPathRequired,
438    #[error("gossip.addr required")]
439    GossipAddrRequired,
440    #[error("api.addr required")]
441    ApiAddrRequired,
442}
443
444/// Log format (JSON only)
445#[derive(Debug, Default, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
446#[serde(rename_all = "lowercase")]
447#[allow(missing_docs)]
448pub enum LogFormat {
449    #[default]
450    Plaintext,
451    Json,
452}
453
454#[derive(Debug, Clone, Deserialize, Serialize)]
455#[serde(rename_all = "kebab-case")]
456pub struct ConsulConfig {
457    pub client: crate::consul::config::Config,
458}
klukai_types/config.rs

klukai_types/
config.rs
