rocketmq_observability/
status.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::sync::Arc;
16
17use chrono::DateTime;
18use chrono::Utc;
19use parking_lot::RwLock;
20use serde::Deserialize;
21use serde::Serialize;
22
23use crate::config::LogsExporter;
24use crate::config::MetricsExporter;
25use crate::config::ObservabilityConfig;
26use crate::config::SubscriberInstallStatus;
27use crate::config::TraceExporter;
28
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
30#[serde(rename_all = "snake_case")]
31pub enum ObservabilityStatusValueV1 {
32    Initializing,
33    Ready,
34    Disabled,
35    Unknown,
36    NotInstrumented,
37    InProgress,
38    Shutdown,
39    Failed,
40}
41
42#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
43pub struct ObservabilityCompiledFeaturesV1 {
44    pub metrics: bool,
45    pub traces: bool,
46    pub logs: bool,
47    pub otlp_grpc: bool,
48    pub prometheus: bool,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
52pub struct ObservabilitySignalStatusV1 {
53    pub enabled: bool,
54    pub exporter: String,
55    pub initialization: ObservabilityStatusValueV1,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
59pub struct ObservabilitySubscriberStatusV1 {
60    pub attempted: bool,
61    pub installed: bool,
62}
63
64/// Authenticated, sanitized status for an initialized observability runtime.
65///
66/// Export, queue, and drop health are deliberately reported as `unknown` or
67/// `not_instrumented` until the exporter SDK exposes authoritative state.
68#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
69pub struct ObservabilityStatusViewV1 {
70    pub schema_version: String,
71    pub observed_at: DateTime<Utc>,
72    pub enabled: bool,
73    pub compiled_features: ObservabilityCompiledFeaturesV1,
74    pub metrics: ObservabilitySignalStatusV1,
75    pub traces: ObservabilitySignalStatusV1,
76    pub logs: ObservabilitySignalStatusV1,
77    pub otlp_protocol: String,
78    pub otlp_endpoint: Option<String>,
79    pub subscriber: ObservabilitySubscriberStatusV1,
80    pub initialization: ObservabilityStatusValueV1,
81    pub export: ObservabilityStatusValueV1,
82    pub queue: ObservabilityStatusValueV1,
83    pub drops: ObservabilityStatusValueV1,
84    pub shutdown: ObservabilityStatusValueV1,
85}
86
87impl ObservabilityStatusViewV1 {
88    pub const SCHEMA_VERSION: &'static str = "rocketmq.observability-status.v1";
89}
90
91#[derive(Debug, Clone)]
92pub struct ObservabilityStatusHandle {
93    inner: Arc<RwLock<ObservabilityStatusState>>,
94}
95
96#[derive(Debug, Clone)]
97struct ObservabilityStatusState {
98    enabled: bool,
99    metrics: ObservabilitySignalStatusV1,
100    traces: ObservabilitySignalStatusV1,
101    logs: ObservabilitySignalStatusV1,
102    otlp_protocol: String,
103    otlp_endpoint: Option<String>,
104    subscriber: ObservabilitySubscriberStatusV1,
105    initialization: ObservabilityStatusValueV1,
106    export: ObservabilityStatusValueV1,
107    queue: ObservabilityStatusValueV1,
108    drops: ObservabilityStatusValueV1,
109    shutdown: ObservabilityStatusValueV1,
110}
111
112impl Default for ObservabilityStatusHandle {
113    fn default() -> Self {
114        Self::from_config(&ObservabilityConfig::default())
115    }
116}
117
118impl ObservabilityStatusHandle {
119    pub(crate) fn from_config(config: &ObservabilityConfig) -> Self {
120        let initialization = if config.enabled {
121            ObservabilityStatusValueV1::Initializing
122        } else {
123            ObservabilityStatusValueV1::Disabled
124        };
125        let export = if config.enabled {
126            ObservabilityStatusValueV1::Unknown
127        } else {
128            ObservabilityStatusValueV1::Disabled
129        };
130        Self {
131            inner: Arc::new(RwLock::new(ObservabilityStatusState {
132                enabled: config.enabled,
133                metrics: metric_status(config),
134                traces: trace_status(config),
135                logs: log_status(config),
136                otlp_protocol: otlp_protocol(config),
137                otlp_endpoint: uses_otlp(config).then(|| sanitize_endpoint(config.otlp.endpoint.as_str())),
138                subscriber: ObservabilitySubscriberStatusV1 {
139                    attempted: false,
140                    installed: false,
141                },
142                initialization,
143                export,
144                queue: status_when_enabled(config, ObservabilityStatusValueV1::NotInstrumented),
145                drops: status_when_enabled(config, ObservabilityStatusValueV1::NotInstrumented),
146                shutdown: status_when_enabled(config, ObservabilityStatusValueV1::Unknown),
147            })),
148        }
149    }
150
151    pub fn view(&self) -> ObservabilityStatusViewV1 {
152        let state = self.inner.read().clone();
153        ObservabilityStatusViewV1 {
154            schema_version: ObservabilityStatusViewV1::SCHEMA_VERSION.to_string(),
155            observed_at: Utc::now(),
156            enabled: state.enabled,
157            compiled_features: compiled_features(),
158            metrics: state.metrics,
159            traces: state.traces,
160            logs: state.logs,
161            otlp_protocol: state.otlp_protocol,
162            otlp_endpoint: state.otlp_endpoint,
163            subscriber: state.subscriber,
164            initialization: state.initialization,
165            export: state.export,
166            queue: state.queue,
167            drops: state.drops,
168            shutdown: state.shutdown,
169        }
170    }
171
172    pub(crate) fn mark_initialized(&self, subscriber: SubscriberInstallStatus) {
173        let mut state = self.inner.write();
174        state.subscriber = ObservabilitySubscriberStatusV1 {
175            attempted: subscriber.attempted,
176            installed: subscriber.installed,
177        };
178        state.initialization = if state.enabled {
179            ObservabilityStatusValueV1::Ready
180        } else {
181            ObservabilityStatusValueV1::Disabled
182        };
183        mark_signal_initialized(&mut state.metrics);
184        mark_signal_initialized(&mut state.traces);
185        mark_signal_initialized(&mut state.logs);
186        state.export = if !state.enabled || !any_signal_enabled(&state) {
187            ObservabilityStatusValueV1::Disabled
188        } else if uses_remote_exporter(&state) {
189            ObservabilityStatusValueV1::Unknown
190        } else {
191            ObservabilityStatusValueV1::Ready
192        };
193    }
194
195    pub(crate) fn mark_shutdown_started(&self) {
196        self.inner.write().shutdown = ObservabilityStatusValueV1::InProgress;
197    }
198
199    pub(crate) fn mark_shutdown_finished(&self, healthy: bool) {
200        self.inner.write().shutdown = if healthy {
201            ObservabilityStatusValueV1::Shutdown
202        } else {
203            ObservabilityStatusValueV1::Failed
204        };
205    }
206}
207
208fn mark_signal_initialized(signal: &mut ObservabilitySignalStatusV1) {
209    if signal.enabled && signal.initialization != ObservabilityStatusValueV1::NotInstrumented {
210        signal.initialization = ObservabilityStatusValueV1::Ready;
211    }
212}
213
214fn metric_status(config: &ObservabilityConfig) -> ObservabilitySignalStatusV1 {
215    signal_status(
216        config.enabled && config.metrics.enabled,
217        match config.metrics.exporter {
218            MetricsExporter::Disable => "disabled",
219            MetricsExporter::OtlpGrpc => "otlp_grpc",
220            MetricsExporter::Prometheus => "prometheus",
221            MetricsExporter::Log => "log",
222        },
223        cfg!(feature = "otel-metrics"),
224    )
225}
226
227fn trace_status(config: &ObservabilityConfig) -> ObservabilitySignalStatusV1 {
228    signal_status(
229        config.enabled && config.traces.enabled,
230        match config.traces.exporter {
231            TraceExporter::Disable => "disabled",
232            TraceExporter::OtlpGrpc => "otlp_grpc",
233            TraceExporter::Log => "log",
234        },
235        cfg!(feature = "otel-traces"),
236    )
237}
238
239fn log_status(config: &ObservabilityConfig) -> ObservabilitySignalStatusV1 {
240    signal_status(
241        config.enabled && config.logs.enabled,
242        match config.logs.exporter {
243            LogsExporter::Disable => "disabled",
244            LogsExporter::OtlpGrpc => "otlp_grpc",
245            LogsExporter::Log => "log",
246        },
247        cfg!(feature = "otel-logs"),
248    )
249}
250
251fn signal_status(enabled: bool, exporter: &str, compiled: bool) -> ObservabilitySignalStatusV1 {
252    ObservabilitySignalStatusV1 {
253        enabled,
254        exporter: exporter.to_string(),
255        initialization: if !enabled {
256            ObservabilityStatusValueV1::Disabled
257        } else if compiled {
258            ObservabilityStatusValueV1::Initializing
259        } else {
260            ObservabilityStatusValueV1::NotInstrumented
261        },
262    }
263}
264
265fn uses_remote_exporter(state: &ObservabilityStatusState) -> bool {
266    [&state.metrics, &state.traces, &state.logs]
267        .into_iter()
268        .any(|signal| signal.enabled && signal.exporter == "otlp_grpc")
269}
270
271fn any_signal_enabled(state: &ObservabilityStatusState) -> bool {
272    state.metrics.enabled || state.traces.enabled || state.logs.enabled
273}
274
275fn status_when_enabled(
276    config: &ObservabilityConfig,
277    enabled_status: ObservabilityStatusValueV1,
278) -> ObservabilityStatusValueV1 {
279    if config.enabled {
280        enabled_status
281    } else {
282        ObservabilityStatusValueV1::Disabled
283    }
284}
285
286pub(crate) fn uses_otlp(config: &ObservabilityConfig) -> bool {
287    config.enabled
288        && ((config.metrics.enabled && config.metrics.exporter == MetricsExporter::OtlpGrpc)
289            || (config.traces.enabled && config.traces.exporter == TraceExporter::OtlpGrpc)
290            || (config.logs.enabled && config.logs.exporter == LogsExporter::OtlpGrpc))
291}
292
293fn otlp_protocol(config: &ObservabilityConfig) -> String {
294    match config.otlp.protocol {
295        crate::config::OtlpProtocol::Grpc => "grpc",
296        crate::config::OtlpProtocol::HttpBinary => "http_binary",
297        crate::config::OtlpProtocol::HttpJson => "http_json",
298    }
299    .to_string()
300}
301
302fn sanitize_endpoint(endpoint: &str) -> String {
303    let Some((scheme, _rest)) = endpoint.trim().split_once("://") else {
304        return "<redacted>".to_string();
305    };
306    match scheme {
307        "http" | "https" | "grpc" => format!("{scheme}://<redacted>"),
308        _ => "<redacted>".to_string(),
309    }
310}
311
312fn compiled_features() -> ObservabilityCompiledFeaturesV1 {
313    ObservabilityCompiledFeaturesV1 {
314        metrics: cfg!(feature = "otel-metrics"),
315        traces: cfg!(feature = "otel-traces"),
316        logs: cfg!(feature = "otel-logs"),
317        otlp_grpc: cfg!(feature = "otlp-grpc"),
318        prometheus: cfg!(feature = "prometheus"),
319    }
320}
321
322#[cfg(test)]
323mod tests {
324    use super::*;
325
326    #[test]
327    fn endpoint_status_does_not_disclose_authority_or_credentials() {
328        let mut config = ObservabilityConfig {
329            enabled: true,
330            ..ObservabilityConfig::default()
331        };
332        config.metrics.enabled = true;
333        config.metrics.exporter = MetricsExporter::OtlpGrpc;
334        config.otlp.endpoint = "https://user:secret@collector.internal:4317/v1/metrics".to_string();
335
336        let view = ObservabilityStatusHandle::from_config(&config).view();
337        let json = serde_json::to_string(&view).expect("status should serialize");
338
339        assert_eq!(view.otlp_endpoint.as_deref(), Some("https://<redacted>"));
340        assert!(!json.contains("secret"));
341        assert!(!json.contains("collector.internal"));
342    }
343
344    #[test]
345    fn unavailable_exporter_health_is_reported_honestly() {
346        let mut config = ObservabilityConfig {
347            enabled: true,
348            ..ObservabilityConfig::default()
349        };
350        config.metrics.enabled = true;
351        config.metrics.exporter = MetricsExporter::OtlpGrpc;
352        let handle = ObservabilityStatusHandle::from_config(&config);
353        handle.mark_initialized(SubscriberInstallStatus::default());
354
355        let view = handle.view();
356        assert_eq!(view.initialization, ObservabilityStatusValueV1::Ready);
357        assert_eq!(view.export, ObservabilityStatusValueV1::Unknown);
358        assert_eq!(view.queue, ObservabilityStatusValueV1::NotInstrumented);
359        assert_eq!(view.drops, ObservabilityStatusValueV1::NotInstrumented);
360    }
361
362    #[test]
363    fn global_disable_does_not_report_signal_initialization_or_export_health() {
364        let mut config = ObservabilityConfig::default();
365        config.metrics.enabled = true;
366        config.metrics.exporter = MetricsExporter::OtlpGrpc;
367        let handle = ObservabilityStatusHandle::from_config(&config);
368        handle.mark_initialized(SubscriberInstallStatus::default());
369
370        let view = handle.view();
371
372        assert_eq!(view.initialization, ObservabilityStatusValueV1::Disabled);
373        assert_eq!(view.metrics.initialization, ObservabilityStatusValueV1::Disabled);
374        assert_eq!(view.export, ObservabilityStatusValueV1::Disabled);
375        assert_eq!(view.queue, ObservabilityStatusValueV1::Disabled);
376        assert_eq!(view.drops, ObservabilityStatusValueV1::Disabled);
377        assert_eq!(view.shutdown, ObservabilityStatusValueV1::Disabled);
378        assert_eq!(view.otlp_endpoint, None);
379    }
380}
rocketmq_observability/status.rs

rocketmq_observability/
status.rs
