rocketmq_observability/
resolver.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::ffi::OsStr;
16use std::ffi::OsString;
17use std::fmt;
18use std::net::IpAddr;
19use std::net::SocketAddr;
20
21use crate::metrics::release_identity::ProcessTelemetryConfig;
22use crate::metrics::release_identity::METRICS_BIND_ADDR_ENV;
23use crate::metrics::release_identity::METRICS_ENABLED_ENV;
24use crate::metrics::release_identity::METRICS_EXPORTER_ENV;
25use crate::metrics::release_identity::METRICS_PATH_ENV;
26use crate::metrics::release_identity::RELEASE_COMMIT_ENV;
27use crate::metrics::release_identity::RELEASE_NONCE_ENV;
28use crate::ObservabilityError;
29use crate::ObservabilityOverrides;
30use crate::StandardOtlpEnvironmentStatus;
31use crate::TelemetryBootstrapConfig;
32use crate::OTEL_EXPORTER_OTLP_ENDPOINT;
33use crate::OTEL_EXPORTER_OTLP_PROTOCOL;
34
35/// Raw process environment values used by the deterministic telemetry resolver.
36///
37/// Missing variables remain `None`; validation and UTF-8 conversion happen only
38/// when a present value participates in resolution.
39#[derive(Clone, Default)]
40pub struct TelemetryEnvironmentValues {
41    /// `ROCKETMQ_RELEASE_COMMIT`, when present.
42    pub release_commit: Option<OsString>,
43    /// `ROCKETMQ_RELEASE_NONCE`, when present.
44    pub release_nonce: Option<OsString>,
45    /// `ROCKETMQ_METRICS_ENABLED`, when present.
46    pub metrics_enabled: Option<OsString>,
47    /// `ROCKETMQ_METRICS_EXPORTER`, when present.
48    pub metrics_exporter: Option<OsString>,
49    /// `ROCKETMQ_METRICS_BIND_ADDR`, when present.
50    pub metrics_bind_addr: Option<OsString>,
51    /// `ROCKETMQ_METRICS_PATH`, when present.
52    pub metrics_path: Option<OsString>,
53    /// `OTEL_EXPORTER_OTLP_ENDPOINT`, when present.
54    pub otlp_endpoint: Option<OsString>,
55    /// `OTEL_EXPORTER_OTLP_PROTOCOL`, when present.
56    pub otlp_protocol: Option<OsString>,
57    /// The service-specific trace sample ratio, when configured and present.
58    pub trace_sample_ratio: Option<OsString>,
59}
60
61impl fmt::Debug for TelemetryEnvironmentValues {
62    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
63        formatter
64            .debug_struct("TelemetryEnvironmentValues")
65            .field("release_commit_present", &self.release_commit.is_some())
66            .field("release_nonce_present", &self.release_nonce.is_some())
67            .field("metrics_enabled_present", &self.metrics_enabled.is_some())
68            .field("metrics_exporter_present", &self.metrics_exporter.is_some())
69            .field("metrics_bind_addr_present", &self.metrics_bind_addr.is_some())
70            .field("metrics_path_present", &self.metrics_path.is_some())
71            .field("otlp_endpoint_present", &self.otlp_endpoint.is_some())
72            .field("otlp_protocol_present", &self.otlp_protocol.is_some())
73            .field("trace_sample_ratio_present", &self.trace_sample_ratio.is_some())
74            .finish()
75    }
76}
77
78impl TelemetryEnvironmentValues {
79    /// Reads the supported telemetry variables from the current process.
80    pub fn read(spec: TelemetryEnvironmentSpec) -> Self {
81        Self::read_with(spec, read_telemetry_environment_value)
82    }
83
84    fn read_with(spec: TelemetryEnvironmentSpec, mut read: impl FnMut(&'static str) -> Option<OsString>) -> Self {
85        Self {
86            release_commit: read(RELEASE_COMMIT_ENV),
87            release_nonce: read(RELEASE_NONCE_ENV),
88            metrics_enabled: read(METRICS_ENABLED_ENV),
89            metrics_exporter: read(METRICS_EXPORTER_ENV),
90            metrics_bind_addr: read(METRICS_BIND_ADDR_ENV),
91            metrics_path: read(METRICS_PATH_ENV),
92            otlp_endpoint: read(OTEL_EXPORTER_OTLP_ENDPOINT),
93            otlp_protocol: read(OTEL_EXPORTER_OTLP_PROTOCOL),
94            trace_sample_ratio: spec.trace_sample_ratio_env.and_then(read),
95        }
96    }
97}
98
99pub(crate) fn read_telemetry_environment_value(name: &'static str) -> Option<OsString> {
100    std::env::var_os(name)
101}
102
103/// Service-specific environment variable names understood by telemetry resolution.
104#[derive(Debug, Clone, Copy, Default)]
105pub struct TelemetryEnvironmentSpec {
106    /// Optional service-specific trace sample ratio variable name.
107    pub trace_sample_ratio_env: Option<&'static str>,
108}
109
110/// Fully resolved telemetry configuration and its validated process metadata.
111pub struct TelemetryResolution {
112    /// Final merged and validated bootstrap configuration.
113    pub bootstrap: TelemetryBootstrapConfig,
114    /// Validated release identity and process metrics selection.
115    pub process: ProcessTelemetryConfig,
116    /// Final Prometheus listener, present only for an active Prometheus exporter.
117    pub prometheus_listener_addr: Option<SocketAddr>,
118    /// Whether at least one present environment field participated in resolution.
119    pub environment_applied: bool,
120}
121
122impl fmt::Debug for TelemetryResolution {
123    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
124        formatter
125            .debug_struct("TelemetryResolution")
126            .field("observability_enabled", &self.bootstrap.observability.enabled)
127            .field("metrics_exporter", &self.bootstrap.observability.metrics.exporter)
128            .field("trace_exporter", &self.bootstrap.observability.traces.exporter)
129            .field("log_exporter", &self.bootstrap.observability.logs.exporter)
130            .field("logging_enabled", &self.bootstrap.logging.enabled)
131            .field("process_metrics_enabled", &self.process.metrics_enabled())
132            .field("process_metrics_exporter", &self.process.metrics_exporter())
133            .field("prometheus_listener_addr", &self.prometheus_listener_addr)
134            .field("environment_applied", &self.environment_applied)
135            .finish()
136    }
137}
138
139/// Resolves telemetry using values read from the current process environment.
140///
141/// # Errors
142///
143/// Returns [`ObservabilityError`] when a present environment value or the final
144/// merged configuration is invalid.
145pub fn resolve_telemetry_from_env(
146    service: &'static str,
147    bootstrap: TelemetryBootstrapConfig,
148    file: &ObservabilityOverrides,
149    spec: TelemetryEnvironmentSpec,
150) -> Result<TelemetryResolution, ObservabilityError> {
151    let environment = TelemetryEnvironmentValues::read(spec);
152    resolve_telemetry_values(service, bootstrap, file, &environment, spec)
153}
154
155/// Resolves defaults, file overrides, and caller-supplied environment values.
156///
157/// The merge order is deterministic: bootstrap defaults, then file overrides,
158/// then only environment fields that are present.
159///
160/// # Errors
161///
162/// Returns [`ObservabilityError`] when a present environment value or the final
163/// merged configuration is invalid.
164pub fn resolve_telemetry_values(
165    service: &'static str,
166    mut bootstrap: TelemetryBootstrapConfig,
167    file: &ObservabilityOverrides,
168    environment: &TelemetryEnvironmentValues,
169    spec: TelemetryEnvironmentSpec,
170) -> Result<TelemetryResolution, ObservabilityError> {
171    file.apply_to(&mut bootstrap.observability);
172    let process_overrides = ProcessTelemetryConfig::try_from_observability_and_values(
173        service,
174        &bootstrap.observability,
175        environment.release_commit.as_deref(),
176        environment.release_nonce.as_deref(),
177        environment.metrics_enabled.as_deref(),
178        environment.metrics_exporter.as_deref(),
179        environment.metrics_bind_addr.as_deref(),
180        environment.metrics_path.as_deref(),
181    )
182    .map_err(|error| ObservabilityError::invalid_config(error.to_string()))?;
183    process_overrides.apply_to(&mut bootstrap.observability);
184    let otlp_status = apply_standard_otlp_environment_values(
185        &mut bootstrap,
186        environment.otlp_endpoint.as_deref(),
187        environment.otlp_protocol.as_deref(),
188    )?;
189    let trace_sample_ratio_applied = apply_optional_trace_sample_ratio(
190        &mut bootstrap.observability,
191        spec.trace_sample_ratio_env,
192        environment.trace_sample_ratio.as_deref(),
193    )?;
194    normalize_and_validate(&bootstrap.observability)?;
195    let process = ProcessTelemetryConfig::try_from_observability_and_values(
196        service,
197        &bootstrap.observability,
198        environment.release_commit.as_deref(),
199        environment.release_nonce.as_deref(),
200        None,
201        None,
202        None,
203        None,
204    )
205    .map_err(|error| ObservabilityError::invalid_config(error.to_string()))?;
206    let prometheus_listener_addr = if bootstrap.observability.metrics.enabled
207        && bootstrap.observability.metrics.exporter == crate::MetricsExporter::Prometheus
208    {
209        process.prometheus_listener_addr()
210    } else {
211        None
212    };
213
214    Ok(TelemetryResolution {
215        bootstrap,
216        process,
217        prometheus_listener_addr,
218        environment_applied: environment.release_commit.is_some()
219            || environment.release_nonce.is_some()
220            || environment.metrics_enabled.is_some()
221            || environment.metrics_exporter.is_some()
222            || environment.metrics_bind_addr.is_some()
223            || environment.metrics_path.is_some()
224            || otlp_status == StandardOtlpEnvironmentStatus::Applied
225            || trace_sample_ratio_applied,
226    })
227}
228
229/// Validates the final cross-signal observability configuration.
230///
231/// Error messages name only the invalid field and constraint; endpoint,
232/// header, and resource-attribute values are never included.
233///
234/// # Errors
235///
236/// Returns [`ObservabilityError`] when a ratio, duration,
237/// exporter, endpoint, or Prometheus listener setting violates its invariant.
238pub fn normalize_and_validate(config: &crate::ObservabilityConfig) -> Result<(), ObservabilityError> {
239    if !is_valid_sample_ratio(config.metrics.sample_ratio) {
240        return Err(ObservabilityError::invalid_config(
241            "metrics.sample_ratio must be finite and between 0.0 and 1.0",
242        ));
243    }
244    if !is_valid_sample_ratio(config.traces.sample_ratio) {
245        return Err(ObservabilityError::invalid_config(
246            "traces.sample_ratio must be finite and between 0.0 and 1.0",
247        ));
248    }
249    if config.metrics.export_interval_millis == 0 {
250        return Err(ObservabilityError::invalid_config(
251            "metrics.export_interval_millis must be greater than 0",
252        ));
253    }
254    if config.metrics.export_timeout_millis == 0 {
255        return Err(ObservabilityError::invalid_config(
256            "metrics.export_timeout_millis must be greater than 0",
257        ));
258    }
259    if config.metrics.cardinality_limit == 0 {
260        return Err(ObservabilityError::invalid_config(
261            "metrics.cardinality_limit must be greater than 0",
262        ));
263    }
264    if config.otlp.timeout_millis == 0 {
265        return Err(ObservabilityError::invalid_config(
266            "otlp.timeout_millis must be greater than 0",
267        ));
268    }
269
270    let uses_otlp = (config.metrics.enabled && config.metrics.exporter == crate::MetricsExporter::OtlpGrpc)
271        || (config.traces.enabled && config.traces.exporter == crate::TraceExporter::OtlpGrpc)
272        || (config.logs.enabled && config.logs.exporter == crate::LogsExporter::OtlpGrpc);
273    if uses_otlp && config.otlp.endpoint.trim().is_empty() {
274        return Err(ObservabilityError::invalid_config(
275            "otlp.endpoint must not be blank when an OTLP exporter is enabled",
276        ));
277    }
278    if uses_otlp && config.otlp.protocol != crate::OtlpProtocol::Grpc {
279        return Err(ObservabilityError::invalid_config(
280            "only OTLP gRPC is implemented; set observability.otlp.protocol to grpc",
281        ));
282    }
283
284    if parse_prometheus_host(&config.prometheus.host).is_none() {
285        return Err(ObservabilityError::invalid_config(
286            "prometheus.host must be an IP address",
287        ));
288    }
289    if config.prometheus.port == 0 {
290        return Err(ObservabilityError::invalid_config(
291            "prometheus.port must be greater than 0",
292        ));
293    }
294    if !is_canonical_prometheus_path(&config.prometheus.path) {
295        return Err(ObservabilityError::invalid_config(
296            "prometheus.path must be a canonical absolute path of at most 128 ASCII characters",
297        ));
298    }
299
300    Ok(())
301}
302
303fn is_valid_sample_ratio(value: f64) -> bool {
304    value.is_finite() && (0.0..=1.0).contains(&value)
305}
306
307pub(crate) fn parse_prometheus_host(value: &str) -> Option<IpAddr> {
308    if value.trim() != value {
309        return None;
310    }
311    value
312        .strip_prefix('[')
313        .and_then(|value| value.strip_suffix(']'))
314        .unwrap_or(value)
315        .parse()
316        .ok()
317}
318
319pub(crate) fn is_canonical_prometheus_path(value: &str) -> bool {
320    !value.is_empty()
321        && value.len() <= 128
322        && value.starts_with('/')
323        && (value == "/"
324            || value
325                .split('/')
326                .skip(1)
327                .all(|segment| !segment.is_empty() && segment != "." && segment != ".."))
328        && value
329            .as_bytes()
330            .iter()
331            .all(|byte| byte.is_ascii_alphanumeric() || matches!(*byte, b'/' | b'-' | b'_' | b'.'))
332}
333
334fn apply_optional_trace_sample_ratio(
335    config: &mut crate::ObservabilityConfig,
336    environment_name: Option<&'static str>,
337    value: Option<&OsStr>,
338) -> Result<bool, ObservabilityError> {
339    let (Some(environment_name), Some(value)) = (environment_name, value) else {
340        return Ok(false);
341    };
342    let value = value
343        .to_str()
344        .ok_or_else(|| ObservabilityError::invalid_config(format!("{environment_name} must contain valid UTF-8")))?;
345    let sample_ratio = value.trim().parse::<f64>().map_err(|_| {
346        ObservabilityError::invalid_config(format!("{environment_name} must be a floating-point number"))
347    })?;
348    if !is_valid_sample_ratio(sample_ratio) {
349        return Err(ObservabilityError::invalid_config(format!(
350            "{environment_name} must be finite and between 0.0 and 1.0"
351        )));
352    }
353    config.traces.sample_ratio = sample_ratio;
354    Ok(true)
355}
356
357pub(crate) fn apply_standard_otlp_environment_values(
358    config: &mut TelemetryBootstrapConfig,
359    endpoint: Option<&OsStr>,
360    protocol: Option<&OsStr>,
361) -> Result<StandardOtlpEnvironmentStatus, ObservabilityError> {
362    let Some(endpoint) = endpoint else {
363        return Ok(StandardOtlpEnvironmentStatus::Unchanged);
364    };
365    let endpoint = endpoint.to_str().ok_or_else(|| {
366        ObservabilityError::invalid_config(format!("{OTEL_EXPORTER_OTLP_ENDPOINT} must contain valid UTF-8"))
367    })?;
368    let endpoint = endpoint.trim();
369    if endpoint.is_empty() {
370        return Ok(StandardOtlpEnvironmentStatus::Unchanged);
371    }
372
373    let protocol = protocol.ok_or_else(|| {
374        ObservabilityError::invalid_config(format!(
375            "{OTEL_EXPORTER_OTLP_PROTOCOL} must be set to grpc when {OTEL_EXPORTER_OTLP_ENDPOINT} is configured"
376        ))
377    })?;
378    let protocol = protocol.to_str().ok_or_else(|| {
379        ObservabilityError::invalid_config(format!("{OTEL_EXPORTER_OTLP_PROTOCOL} must contain valid UTF-8"))
380    })?;
381    if protocol != "grpc" {
382        return Err(ObservabilityError::invalid_config(format!(
383            "{OTEL_EXPORTER_OTLP_PROTOCOL} must be exactly grpc"
384        )));
385    }
386
387    config.observability.enabled = true;
388    config.observability.metrics.enabled = true;
389    config.observability.metrics.exporter = crate::MetricsExporter::OtlpGrpc;
390    config.observability.traces.enabled = true;
391    config.observability.traces.exporter = crate::TraceExporter::OtlpGrpc;
392    config.observability.logs.enabled = true;
393    config.observability.logs.exporter = crate::LogsExporter::OtlpGrpc;
394    config.observability.otlp.endpoint = endpoint.to_owned();
395    config.observability.otlp.protocol = crate::OtlpProtocol::Grpc;
396
397    Ok(StandardOtlpEnvironmentStatus::Applied)
398}
399
400#[cfg(test)]
401mod tests {
402    use super::*;
403
404    #[test]
405    fn environment_reader_uses_supported_names_without_defaulting_absent_values() {
406        let mut names = Vec::new();
407
408        let values = TelemetryEnvironmentValues::read_with(
409            TelemetryEnvironmentSpec {
410                trace_sample_ratio_env: Some("ROCKETMQ_BROKER_TRACE_SAMPLE_RATIO"),
411            },
412            |name| {
413                names.push(name);
414                match name {
415                    "ROCKETMQ_RELEASE_COMMIT" => Some("commit-value".into()),
416                    "ROCKETMQ_BROKER_TRACE_SAMPLE_RATIO" => Some("0.5".into()),
417                    _ => None,
418                }
419            },
420        );
421
422        assert_eq!(
423            names,
424            [
425                "ROCKETMQ_RELEASE_COMMIT",
426                "ROCKETMQ_RELEASE_NONCE",
427                "ROCKETMQ_METRICS_ENABLED",
428                "ROCKETMQ_METRICS_EXPORTER",
429                "ROCKETMQ_METRICS_BIND_ADDR",
430                "ROCKETMQ_METRICS_PATH",
431                "OTEL_EXPORTER_OTLP_ENDPOINT",
432                "OTEL_EXPORTER_OTLP_PROTOCOL",
433                "ROCKETMQ_BROKER_TRACE_SAMPLE_RATIO",
434            ]
435        );
436        assert_eq!(
437            values.release_commit.as_deref(),
438            Some(std::ffi::OsStr::new("commit-value"))
439        );
440        assert_eq!(values.release_nonce, None);
441        assert_eq!(values.metrics_enabled, None);
442        assert_eq!(values.metrics_exporter, None);
443        assert_eq!(values.metrics_bind_addr, None);
444        assert_eq!(values.metrics_path, None);
445        assert_eq!(values.otlp_endpoint, None);
446        assert_eq!(values.otlp_protocol, None);
447        assert_eq!(values.trace_sample_ratio.as_deref(), Some(std::ffi::OsStr::new("0.5")));
448    }
449}
rocketmq_observability/resolver.rs

rocketmq_observability/
resolver.rs
