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

klukai_types/
config.rs
