rocketmq_observability/
handle.rs

1// Copyright 2026 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
15//! Cloneable, lifecycle-safe telemetry access for business components.
16//!
17//! A [`TelemetryHandle`] never owns or exposes an OpenTelemetry SDK provider. Providers and
18//! exporter workers stay under `TelemetryRuntimeGuard`; the handle contains only immutable
19//! policy, pre-created instruments, and a shared lifecycle gate.
20
21use std::fmt;
22use std::sync::atomic::AtomicU8;
23use std::sync::atomic::Ordering;
24use std::sync::Arc;
25
26use crate::config::ObservabilityConfig;
27use crate::metrics::labels::MetricLabelPolicy;
28#[cfg(feature = "otel-metrics")]
29use crate::metrics::release_identity::ReleaseIdentityRegistration;
30#[cfg(feature = "otel-metrics")]
31use crate::metrics::release_identity::ReleaseIdentityRegistrationStatus;
32#[cfg(feature = "otel-metrics")]
33use crate::metrics::release_identity::ValidatedReleaseIdentity;
34
35/// Fixed instrumentation scope for broker metrics.
36pub const BROKER_METER_SCOPE: &str = "rocketmq-broker";
37/// Fixed instrumentation scope for store metrics.
38pub const STORE_METER_SCOPE: &str = "rocketmq-store";
39/// Fixed instrumentation scope for client metrics.
40pub const CLIENT_METER_SCOPE: &str = "rocketmq-client";
41/// Fixed instrumentation scope for transport metrics.
42pub const TRANSPORT_METER_SCOPE: &str = "rocketmq-transport";
43/// Fixed instrumentation scope for controller metrics.
44pub const CONTROLLER_METER_SCOPE: &str = "rocketmq-controller";
45/// Fixed instrumentation scope for NameServer metrics.
46pub const NAMESRV_METER_SCOPE: &str = "rocketmq-namesrv";
47/// Fixed instrumentation scope for proxy metrics.
48pub const PROXY_METER_SCOPE: &str = "rocketmq-proxy";
49/// Fixed instrumentation scope for MCP metrics.
50pub const MCP_METER_SCOPE: &str = "rocketmq-mcp";
51/// Fixed instrumentation scope for runtime lifecycle and scheduler metrics.
52pub const RUNTIME_METER_SCOPE: &str = "rocketmq-runtime";
53/// Fixed instrumentation scope for AI SRE Control Plane metrics.
54pub const SRE_CONTROL_PLANE_METER_SCOPE: &str = "rocketmq-sre-control-plane";
55/// Fixed instrumentation scope for AI SRE Connector metrics.
56pub const SRE_CONNECTOR_METER_SCOPE: &str = "rocketmq-sre-connector";
57/// Fixed instrumentation scope for tiered-store metrics.
58pub const TIERED_STORE_METER_SCOPE: &str = "rocketmq-tieredstore";
59/// Fixed instrumentation scope for dashboard operations metrics.
60pub const DASHBOARD_METER_SCOPE: &str = "rocketmq-dashboard";
61
62/// Current state of a telemetry handle's runtime-owned lifecycle.
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64#[repr(u8)]
65pub enum TelemetryState {
66    /// The runtime accepts telemetry records.
67    Active = 0,
68    /// Shutdown has closed new telemetry admission while owned providers drain.
69    Closing = 1,
70    /// Provider shutdown is complete; all handle operations are no-ops.
71    Closed = 2,
72}
73
74impl TelemetryState {
75    fn from_raw(value: u8) -> Self {
76        match value {
77            value if value == Self::Active as u8 => Self::Active,
78            value if value == Self::Closing as u8 => Self::Closing,
79            _ => Self::Closed,
80        }
81    }
82}
83
84/// Immutable per-handle tracing and message-correlation policy.
85#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
86pub struct TracePolicy {
87    /// Whether this handle may create exportable tracing spans.
88    pub enabled: bool,
89    /// Whether message trace context should be injected and extracted.
90    pub propagate_context: bool,
91    /// Whether spans may include an opaque message identifier.
92    pub record_message_id: bool,
93    /// Whether spans may include message keys.
94    pub record_message_keys: bool,
95    /// Whether spans may include message body size.
96    pub record_body_size: bool,
97}
98
99impl TracePolicy {
100    fn from_config(config: &ObservabilityConfig) -> Self {
101        Self {
102            enabled: config.enabled && config.traces.enabled,
103            propagate_context: config.traces.propagate_context,
104            record_message_id: config.traces.record_message_id,
105            record_message_keys: config.traces.record_message_keys,
106            record_body_size: config.traces.record_body_size,
107        }
108    }
109
110    const fn disabled() -> Self {
111        Self {
112            enabled: false,
113            propagate_context: false,
114            record_message_id: false,
115            record_message_keys: false,
116            record_body_size: false,
117        }
118    }
119}
120
121/// Immutable, non-sensitive metrics settings captured from the final resolved configuration.
122#[derive(Debug, Clone, Copy, PartialEq)]
123pub struct MetricsRuntimePolicy {
124    /// Whether this handle may record metrics.
125    pub enabled: bool,
126    /// Fraction of sampled Broker metric operations.
127    pub sample_ratio: f64,
128    /// Export and dependent snapshot refresh cadence in milliseconds.
129    pub export_interval_millis: u64,
130    /// Maximum cardinality and bounded Broker metric work.
131    pub cardinality_limit: usize,
132}
133
134impl MetricsRuntimePolicy {
135    fn from_config(config: &ObservabilityConfig) -> Self {
136        Self {
137            enabled: config.enabled && config.metrics.enabled,
138            sample_ratio: config.metrics.sample_ratio,
139            export_interval_millis: config.metrics.export_interval_millis,
140            cardinality_limit: config.metrics.cardinality_limit,
141        }
142    }
143
144    const fn disabled() -> Self {
145        Self {
146            enabled: false,
147            sample_ratio: 0.0,
148            export_interval_millis: 0,
149            cardinality_limit: 0,
150        }
151    }
152}
153
154impl Default for MetricsRuntimePolicy {
155    fn default() -> Self {
156        Self::disabled()
157    }
158}
159
160#[derive(Clone)]
161enum HandleBackend {
162    Noop,
163    Active(Arc<ActiveHandle>),
164}
165
166struct ActiveHandle {
167    state: AtomicU8,
168    metrics_runtime_policy: MetricsRuntimePolicy,
169    trace_policy: TracePolicy,
170    metric_label_policy: MetricLabelPolicy,
171    #[cfg(feature = "otel-metrics")]
172    meters: Option<FixedMeters>,
173    #[cfg(feature = "otel-metrics")]
174    release_identity: parking_lot::Mutex<Option<ReleaseIdentityRegistration>>,
175}
176
177/// Failure to register a release identity on an explicit telemetry handle.
178#[cfg(feature = "otel-metrics")]
179#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
180pub enum ReleaseIdentityRegistrationError {
181    /// The handle is no-op, closing, or closed.
182    #[error("telemetry handle is not active")]
183    Inactive,
184    /// Metrics were not initialized for this runtime.
185    #[error("telemetry handle has no meter provider")]
186    MeterUnavailable,
187    /// The release service does not match a fixed RocketMQ instrumentation scope.
188    #[error("release service does not match a fixed RocketMQ instrumentation scope")]
189    UnsupportedService,
190    /// This handle already registered a different release identity.
191    #[error("telemetry handle already registered a different release identity")]
192    ConflictingIdentity,
193}
194
195/// Cloneable telemetry capability injected into business components.
196///
197/// This type deliberately has no shutdown method and contains no SDK provider. Clone it freely;
198/// the corresponding non-cloneable `TelemetryRuntimeGuard` remains the sole shutdown owner.
199#[derive(Clone)]
200pub struct TelemetryHandle {
201    backend: HandleBackend,
202}
203
204impl fmt::Debug for TelemetryHandle {
205    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
206        formatter
207            .debug_struct("TelemetryHandle")
208            .field("state", &self.state())
209            .field("trace_policy", &self.trace_policy())
210            .finish_non_exhaustive()
211    }
212}
213
214impl Default for TelemetryHandle {
215    fn default() -> Self {
216        Self::noop()
217    }
218}
219
220impl TelemetryHandle {
221    /// Creates a handle that never reads or installs OpenTelemetry global state.
222    #[must_use]
223    pub const fn noop() -> Self {
224        Self {
225            backend: HandleBackend::Noop,
226        }
227    }
228
229    #[cfg(feature = "otel-metrics")]
230    pub(crate) fn active(
231        config: &ObservabilityConfig,
232        meter_provider: Option<&opentelemetry_sdk::metrics::SdkMeterProvider>,
233    ) -> Self {
234        Self {
235            backend: HandleBackend::Active(Arc::new(ActiveHandle {
236                state: AtomicU8::new(TelemetryState::Active as u8),
237                metrics_runtime_policy: MetricsRuntimePolicy::from_config(config),
238                trace_policy: TracePolicy::from_config(config),
239                metric_label_policy: MetricLabelPolicy::new(
240                    config.metrics.cardinality_limit,
241                    config.metrics.topic_label_enabled,
242                    config.metrics.consumer_group_label_enabled,
243                ),
244                meters: meter_provider.map(FixedMeters::new),
245                release_identity: parking_lot::Mutex::new(None),
246            })),
247        }
248    }
249
250    #[cfg(not(feature = "otel-metrics"))]
251    pub(crate) fn active(config: &ObservabilityConfig) -> Self {
252        Self {
253            backend: HandleBackend::Active(Arc::new(ActiveHandle {
254                state: AtomicU8::new(TelemetryState::Active as u8),
255                metrics_runtime_policy: MetricsRuntimePolicy::from_config(config),
256                trace_policy: TracePolicy::from_config(config),
257                metric_label_policy: MetricLabelPolicy::new(
258                    config.metrics.cardinality_limit,
259                    config.metrics.topic_label_enabled,
260                    config.metrics.consumer_group_label_enabled,
261                ),
262            })),
263        }
264    }
265
266    /// Returns the current shared lifecycle state.
267    #[must_use]
268    pub fn state(&self) -> TelemetryState {
269        match &self.backend {
270            HandleBackend::Noop => TelemetryState::Closed,
271            HandleBackend::Active(inner) => TelemetryState::from_raw(inner.state.load(Ordering::Acquire)),
272        }
273    }
274
275    /// Returns whether this handle currently admits telemetry operations.
276    #[must_use]
277    pub fn is_active(&self) -> bool {
278        self.state() == TelemetryState::Active
279    }
280
281    /// Returns whether this active runtime currently enables metrics.
282    #[must_use]
283    pub fn metrics_enabled(&self) -> bool {
284        self.metrics_runtime_policy().enabled
285    }
286
287    /// Returns final metrics runtime policy while the telemetry lifecycle is active.
288    ///
289    /// No-op, closing, and closed handles return a disabled policy so surviving clones cannot
290    /// continue metrics work after their runtime owner begins shutdown.
291    #[must_use]
292    pub fn metrics_runtime_policy(&self) -> MetricsRuntimePolicy {
293        match &self.backend {
294            HandleBackend::Active(inner) if self.is_active() => inner.metrics_runtime_policy,
295            HandleBackend::Noop | HandleBackend::Active(_) => MetricsRuntimePolicy::disabled(),
296        }
297    }
298
299    /// Returns this handle's immutable trace policy while the runtime is active.
300    ///
301    /// A closing or closed handle returns a disabled policy so clones that outlive their runtime
302    /// cannot continue producing telemetry.
303    #[must_use]
304    pub fn trace_policy(&self) -> TracePolicy {
305        match &self.backend {
306            HandleBackend::Active(inner) if self.is_active() => inner.trace_policy,
307            HandleBackend::Noop | HandleBackend::Active(_) => TracePolicy::disabled(),
308        }
309    }
310
311    /// Returns this telemetry instance's shared metric-label policy.
312    ///
313    /// Clones of this handle share one bounded-cardinality budget. Independently initialized
314    /// handles receive independent policies and budgets.
315    #[must_use]
316    pub fn metric_label_policy(&self) -> MetricLabelPolicy {
317        match &self.backend {
318            HandleBackend::Active(inner) if self.is_active() => inner.metric_label_policy.clone(),
319            HandleBackend::Noop | HandleBackend::Active(_) => MetricLabelPolicy::disabled(),
320        }
321    }
322
323    /// Returns a recorder bound to a fixed instrumentation scope.
324    #[must_use]
325    pub fn child(&self, scope: &'static str) -> TelemetryRecorder {
326        TelemetryRecorder {
327            handle: self.clone(),
328            scope,
329        }
330    }
331
332    /// Returns a pre-created meter to crate-owned typed metric recorders.
333    ///
334    /// Raw meters never cross the observability crate boundary. Unsupported scopes and closing,
335    /// closed, or no-op handles return `None`.
336    #[cfg(feature = "otel-metrics")]
337    #[must_use]
338    pub(crate) fn meter(&self, scope: &str) -> Option<opentelemetry::metrics::Meter> {
339        match &self.backend {
340            HandleBackend::Active(inner) if self.is_active() => {
341                inner.meters.as_ref().and_then(|meters| meters.get(scope))
342            }
343            HandleBackend::Noop | HandleBackend::Active(_) => None,
344        }
345    }
346
347    /// Registers one instance-scoped release identity before service readiness.
348    ///
349    /// Registration is idempotent for the same identity. Registering a different identity on the
350    /// same runtime fails closed, and no process-global meter provider is read or modified.
351    ///
352    /// # Errors
353    ///
354    /// Returns [`ReleaseIdentityRegistrationError`] when the runtime is inactive, metrics were not
355    /// initialized, the service does not name a fixed scope, or a conflicting identity was
356    /// registered previously.
357    #[cfg(feature = "otel-metrics")]
358    pub fn register_release_identity(
359        &self,
360        identity: ValidatedReleaseIdentity,
361    ) -> Result<ReleaseIdentityRegistrationStatus, ReleaseIdentityRegistrationError> {
362        let HandleBackend::Active(inner) = &self.backend else {
363            return Err(ReleaseIdentityRegistrationError::Inactive);
364        };
365
366        let mut registration = inner.release_identity.lock();
367        if TelemetryState::from_raw(inner.state.load(Ordering::Acquire)) != TelemetryState::Active {
368            return Err(ReleaseIdentityRegistrationError::Inactive);
369        }
370
371        if let Some(existing) = registration.as_ref() {
372            return if existing.identity() == &identity {
373                Ok(ReleaseIdentityRegistrationStatus::AlreadyRegistered)
374            } else {
375                Err(ReleaseIdentityRegistrationError::ConflictingIdentity)
376            };
377        }
378
379        let meters = inner
380            .meters
381            .as_ref()
382            .ok_or(ReleaseIdentityRegistrationError::MeterUnavailable)?;
383        let meter = meters
384            .get(identity.service())
385            .ok_or(ReleaseIdentityRegistrationError::UnsupportedService)?;
386        let release_identity = ReleaseIdentityRegistration::new(identity);
387        let status = release_identity.register(&meter);
388        *registration = Some(release_identity);
389        Ok(status)
390    }
391
392    /// Returns whether this active runtime registered a release identity.
393    #[must_use]
394    pub fn release_identity_registered(&self) -> bool {
395        #[cfg(feature = "otel-metrics")]
396        {
397            let HandleBackend::Active(inner) = &self.backend else {
398                return false;
399            };
400            self.is_active()
401                && inner
402                    .release_identity
403                    .lock()
404                    .as_ref()
405                    .is_some_and(ReleaseIdentityRegistration::is_registered)
406        }
407
408        #[cfg(not(feature = "otel-metrics"))]
409        {
410            false
411        }
412    }
413
414    pub(crate) fn begin_closing(&self) {
415        if let HandleBackend::Active(inner) = &self.backend {
416            #[cfg(feature = "otel-metrics")]
417            let _registration = inner.release_identity.lock();
418            let _ = inner.state.compare_exchange(
419                TelemetryState::Active as u8,
420                TelemetryState::Closing as u8,
421                Ordering::AcqRel,
422                Ordering::Acquire,
423            );
424        }
425    }
426
427    pub(crate) fn mark_closed(&self) {
428        if let HandleBackend::Active(inner) = &self.backend {
429            inner.state.store(TelemetryState::Closed as u8, Ordering::Release);
430        }
431    }
432}
433
434/// Lightweight recorder scoped to one business component.
435#[derive(Debug, Clone)]
436pub struct TelemetryRecorder {
437    handle: TelemetryHandle,
438    scope: &'static str,
439}
440
441impl TelemetryRecorder {
442    /// Returns the recorder's fixed instrumentation scope.
443    #[must_use]
444    pub const fn scope(&self) -> &'static str {
445        self.scope
446    }
447
448    /// Returns whether the parent runtime currently admits telemetry records.
449    #[must_use]
450    pub fn is_active(&self) -> bool {
451        self.handle.is_active()
452    }
453
454    /// Returns the parent handle's current trace policy.
455    #[must_use]
456    pub fn trace_policy(&self) -> TracePolicy {
457        self.handle.trace_policy()
458    }
459
460    /// Returns the parent handle's instance-scoped metric-label policy.
461    #[must_use]
462    pub fn metric_label_policy(&self) -> MetricLabelPolicy {
463        self.handle.metric_label_policy()
464    }
465
466    /// Returns the recorder's pre-created meter to crate-owned typed metric recorders.
467    #[cfg(feature = "otel-metrics")]
468    #[must_use]
469    pub(crate) fn meter(&self) -> Option<opentelemetry::metrics::Meter> {
470        self.handle.meter(self.scope)
471    }
472}
473
474#[cfg(feature = "otel-metrics")]
475#[derive(Clone)]
476struct FixedMeters {
477    broker: opentelemetry::metrics::Meter,
478    store: opentelemetry::metrics::Meter,
479    client: opentelemetry::metrics::Meter,
480    transport: opentelemetry::metrics::Meter,
481    controller: opentelemetry::metrics::Meter,
482    namesrv: opentelemetry::metrics::Meter,
483    proxy: opentelemetry::metrics::Meter,
484    mcp: opentelemetry::metrics::Meter,
485    runtime: opentelemetry::metrics::Meter,
486    sre_control_plane: opentelemetry::metrics::Meter,
487    sre_connector: opentelemetry::metrics::Meter,
488    tiered_store: opentelemetry::metrics::Meter,
489    dashboard: opentelemetry::metrics::Meter,
490}
491
492#[cfg(feature = "otel-metrics")]
493impl FixedMeters {
494    fn new(provider: &opentelemetry_sdk::metrics::SdkMeterProvider) -> Self {
495        use opentelemetry::metrics::MeterProvider;
496
497        Self {
498            broker: provider.meter(BROKER_METER_SCOPE),
499            store: provider.meter(STORE_METER_SCOPE),
500            client: provider.meter(CLIENT_METER_SCOPE),
501            transport: provider.meter(TRANSPORT_METER_SCOPE),
502            controller: provider.meter(CONTROLLER_METER_SCOPE),
503            namesrv: provider.meter(NAMESRV_METER_SCOPE),
504            proxy: provider.meter(PROXY_METER_SCOPE),
505            mcp: provider.meter(MCP_METER_SCOPE),
506            runtime: provider.meter(RUNTIME_METER_SCOPE),
507            sre_control_plane: provider.meter(SRE_CONTROL_PLANE_METER_SCOPE),
508            sre_connector: provider.meter(SRE_CONNECTOR_METER_SCOPE),
509            tiered_store: provider.meter(TIERED_STORE_METER_SCOPE),
510            dashboard: provider.meter(DASHBOARD_METER_SCOPE),
511        }
512    }
513
514    fn get(&self, scope: &str) -> Option<opentelemetry::metrics::Meter> {
515        match scope {
516            BROKER_METER_SCOPE => Some(self.broker.clone()),
517            STORE_METER_SCOPE => Some(self.store.clone()),
518            CLIENT_METER_SCOPE => Some(self.client.clone()),
519            TRANSPORT_METER_SCOPE => Some(self.transport.clone()),
520            CONTROLLER_METER_SCOPE => Some(self.controller.clone()),
521            NAMESRV_METER_SCOPE => Some(self.namesrv.clone()),
522            PROXY_METER_SCOPE => Some(self.proxy.clone()),
523            MCP_METER_SCOPE => Some(self.mcp.clone()),
524            RUNTIME_METER_SCOPE => Some(self.runtime.clone()),
525            SRE_CONTROL_PLANE_METER_SCOPE => Some(self.sre_control_plane.clone()),
526            SRE_CONNECTOR_METER_SCOPE => Some(self.sre_connector.clone()),
527            TIERED_STORE_METER_SCOPE => Some(self.tiered_store.clone()),
528            DASHBOARD_METER_SCOPE => Some(self.dashboard.clone()),
529            _ => None,
530        }
531    }
532}
rocketmq_observability/handle.rs

rocketmq_observability/
handle.rs
