laminar_connectors/otel/config/
mod.rs

1//! OTel source connector configuration.
2//!
3//! Parses and validates configuration for the OTLP/gRPC receiver.
4
5use crate::config::{ConfigKeySpec, ConnectorConfig};
6use crate::error::ConnectorError;
7
8/// Which OTel signal type to receive.
9///
10/// Each source handles exactly one signal type. Create separate sources
11/// for different signals (each on its own port).
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum OtelSignal {
14    /// Trace spans.
15    Traces,
16    /// Metric data points.
17    Metrics,
18    /// Log records.
19    Logs,
20}
21
22impl OtelSignal {
23    /// Parse from a string value (case-insensitive).
24    ///
25    /// # Errors
26    ///
27    /// Returns `ConnectorError::ConfigurationError` for unknown signal types.
28    pub fn parse(s: &str) -> Result<Self, ConnectorError> {
29        match s.to_lowercase().as_str() {
30            "traces" | "trace" => Ok(Self::Traces),
31            "metrics" | "metric" => Ok(Self::Metrics),
32            "logs" | "log" => Ok(Self::Logs),
33            other => Err(ConnectorError::ConfigurationError(format!(
34                "unknown OTel signal type '{other}': expected traces, metrics, or logs \
35                 (create separate sources for each signal type)"
36            ))),
37        }
38    }
39}
40
41/// OTel source connector configuration.
42#[derive(Debug, Clone)]
43pub struct OtelSourceConfig {
44    /// gRPC bind address.
45    pub bind_address: String,
46    /// gRPC listen port.
47    pub port: u16,
48    /// Which signal type to receive (one per source).
49    pub signals: OtelSignal,
50    /// Max rows per batch.
51    pub batch_size: usize,
52    /// Bounded channel capacity.
53    pub channel_capacity: usize,
54}
55
56impl Default for OtelSourceConfig {
57    fn default() -> Self {
58        Self {
59            bind_address: "0.0.0.0".to_string(),
60            port: 4317,
61            signals: OtelSignal::Traces,
62            batch_size: 1024,
63            channel_capacity: 64,
64        }
65    }
66}
67
68impl OtelSourceConfig {
69    /// Parse configuration from a `ConnectorConfig`.
70    ///
71    /// # Errors
72    ///
73    /// Returns `ConnectorError::ConfigurationError` if any value is invalid.
74    pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
75        let mut cfg = Self::default();
76
77        if let Some(addr) = config.get("bind.address").or(config.get("bind_address")) {
78            cfg.bind_address = addr.to_string();
79        }
80
81        if let Some(port_str) = config.get("port") {
82            cfg.port = port_str.parse::<u16>().map_err(|e| {
83                ConnectorError::ConfigurationError(format!("invalid port '{port_str}': {e}"))
84            })?;
85        }
86
87        if let Some(sig) = config.get("signals").or(config.get("signal")) {
88            cfg.signals = OtelSignal::parse(sig)?;
89        }
90
91        if let Some(bs) = config.get("batch_size").or(config.get("batch.size")) {
92            cfg.batch_size = bs.parse::<usize>().map_err(|e| {
93                ConnectorError::ConfigurationError(format!("invalid batch_size '{bs}': {e}"))
94            })?;
95            if cfg.batch_size == 0 {
96                return Err(ConnectorError::ConfigurationError(
97                    "batch_size must be > 0".into(),
98                ));
99            }
100        }
101
102        if let Some(cc) = config
103            .get("channel_capacity")
104            .or(config.get("channel.capacity"))
105        {
106            cfg.channel_capacity = cc.parse::<usize>().map_err(|e| {
107                ConnectorError::ConfigurationError(format!("invalid channel_capacity '{cc}': {e}"))
108            })?;
109            if cfg.channel_capacity == 0 {
110                return Err(ConnectorError::ConfigurationError(
111                    "channel_capacity must be > 0".into(),
112                ));
113            }
114        }
115
116        Ok(cfg)
117    }
118
119    /// Full socket address for binding.
120    #[must_use]
121    pub fn socket_addr(&self) -> String {
122        format!("{}:{}", self.bind_address, self.port)
123    }
124}
125
126/// Configuration keys accepted by the OTel source connector.
127#[must_use]
128pub fn otel_source_config_keys() -> Vec<ConfigKeySpec> {
129    vec![
130        ConfigKeySpec::optional("port", "gRPC listen port", "4317"),
131        ConfigKeySpec::optional("bind.address", "Listen address", "0.0.0.0"),
132        ConfigKeySpec::optional("signals", "Signal type: traces, metrics, or logs", "traces"),
133        ConfigKeySpec::optional("batch_size", "Max rows per RecordBatch", "1024"),
134        ConfigKeySpec::optional("channel_capacity", "Batch channel capacity", "64"),
135    ]
136}
137
138#[cfg(test)]
139mod tests {
140    use super::*;
141
142    #[test]
143    fn test_zero_batch_size() {
144        let config = ConnectorConfig::with_properties(
145            "otel",
146            [("batch_size".to_string(), "0".to_string())]
147                .into_iter()
148                .collect(),
149        );
150        assert!(OtelSourceConfig::from_config(&config).is_err());
151    }
152
153    #[test]
154    fn test_invalid_signal() {
155        let config = ConnectorConfig::with_properties(
156            "otel",
157            [("signals".to_string(), "invalid".to_string())]
158                .into_iter()
159                .collect(),
160        );
161        assert!(OtelSourceConfig::from_config(&config).is_err());
162    }
163
164    #[test]
165    fn test_signal_parse() {
166        assert_eq!(OtelSignal::parse("traces").unwrap(), OtelSignal::Traces);
167        assert_eq!(OtelSignal::parse("TRACE").unwrap(), OtelSignal::Traces);
168        assert_eq!(OtelSignal::parse("metrics").unwrap(), OtelSignal::Metrics);
169        assert_eq!(OtelSignal::parse("logs").unwrap(), OtelSignal::Logs);
170        assert!(OtelSignal::parse("bad").is_err());
171    }
172
173    #[test]
174    fn test_all_signal_rejected() {
175        let err = OtelSignal::parse("all").unwrap_err();
176        assert!(err.to_string().contains("separate sources"));
177    }
178
179    #[test]
180    fn test_socket_addr() {
181        let cfg = OtelSourceConfig::default();
182        assert_eq!(cfg.socket_addr(), "0.0.0.0:4317");
183    }
184}
laminar_connectors/otel/config/mod.rs

laminar_connectors/otel/config/
mod.rs
