nidus_observability/
lib.rs

1#![deny(missing_docs)]
2
3//! Production observability composition for Nidus applications.
4//!
5//! `Observability` composes the lower-level Nidus hooks that already exist for
6//! HTTP, events, jobs, tracing, and adapter-owned operations. It does not
7//! install process-global exporters or monkey-patch third-party crates.
8
9use std::{
10    borrow::Cow,
11    collections::{BTreeMap, BTreeSet},
12    future::Future,
13    sync::{Arc, LazyLock, Mutex},
14    time::{Duration, Instant},
15};
16
17use tracing::Instrument;
18
19#[cfg(feature = "events")]
20use nidus_events::{EventBus, EventObserver, ObservedEventBus, ObservedEventContext};
21#[cfg(feature = "http")]
22use nidus_http::middleware::{
23    ApiDefaults, HttpMetricsHook, MetricsLayer, PrometheusMetrics, metrics_layer,
24};
25#[cfg(feature = "jobs")]
26use nidus_jobs::{JobObserver, JobResultStatus, ObservedJobContext, ObservedJobRunner};
27
28/// Production observability configuration and in-process metrics state.
29#[derive(Clone, Debug)]
30pub struct Observability {
31    config: Arc<ObservabilityConfig>,
32    state: Arc<Mutex<ObservabilityState>>,
33    #[cfg(feature = "http")]
34    http_metrics: PrometheusMetrics,
35}
36
37impl Observability {
38    /// Creates production observability metadata for a service.
39    pub fn production(service_name: impl Into<String>) -> Self {
40        let service_name = service_name.into();
41        Self {
42            config: Arc::new(ObservabilityConfig {
43                service_name: service_name.clone(),
44                version: None,
45                environment: None,
46                prometheus: false,
47                http_metrics: true,
48                event_metrics: true,
49                job_metrics: true,
50                adapter_instrumentation: true,
51                tracing: false,
52                max_series: None,
53                excluded_routes: BTreeSet::new(),
54                #[cfg(feature = "otel")]
55                otel: None,
56            }),
57            state: Arc::new(Mutex::new(ObservabilityState::default())),
58            #[cfg(feature = "http")]
59            http_metrics: PrometheusMetrics::new(),
60        }
61    }
62
63    /// Sets the service version label used by tracing and OpenTelemetry config.
64    pub fn version(mut self, version: impl Into<String>) -> Self {
65        Arc::make_mut(&mut self.config).version = Some(version.into());
66        self
67    }
68
69    /// Sets the deployment environment label.
70    pub fn environment(mut self, environment: impl Into<String>) -> Self {
71        Arc::make_mut(&mut self.config).environment = Some(environment.into());
72        self
73    }
74
75    /// Enables Prometheus text exposition for configured metrics.
76    pub fn prometheus(mut self) -> Self {
77        Arc::make_mut(&mut self.config).prometheus = true;
78        self
79    }
80
81    /// Enables HTTP tracing when this observability object is applied by Nidus.
82    pub fn tracing(mut self) -> Self {
83        Arc::make_mut(&mut self.config).tracing = true;
84        self
85    }
86
87    /// Builds OpenTelemetry resource configuration from environment variables.
88    ///
89    /// This stores configuration only. Exporters and tracing subscribers remain
90    /// explicit application choices.
91    #[cfg(feature = "otel")]
92    pub fn otel_from_env(mut self) -> Self {
93        let mut config = nidus_http::otel::OtelConfig::new(self.service_name());
94        if let Some(version) = self.version_label() {
95            config = config.version(version);
96        }
97        if let Some(environment) = self.environment_label() {
98            config = config.environment(environment);
99        }
100        if let Ok(endpoint) = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") {
101            config = config.with_otlp_endpoint(endpoint);
102        }
103        Arc::make_mut(&mut self.config).otel = Some(config);
104        self
105    }
106
107    /// No-op OpenTelemetry setup when the `otel` feature is disabled.
108    #[cfg(not(feature = "otel"))]
109    pub fn otel_from_env(self) -> Self {
110        self
111    }
112
113    /// Disables HTTP request metrics.
114    pub fn without_http_metrics(mut self) -> Self {
115        Arc::make_mut(&mut self.config).http_metrics = false;
116        self
117    }
118
119    /// Disables observed event metrics.
120    pub fn without_event_metrics(mut self) -> Self {
121        Arc::make_mut(&mut self.config).event_metrics = false;
122        self
123    }
124
125    /// Disables observed job metrics.
126    pub fn without_job_metrics(mut self) -> Self {
127        Arc::make_mut(&mut self.config).job_metrics = false;
128        self
129    }
130
131    /// Disables adapter-owned operation instrumentation.
132    pub fn without_adapter_instrumentation(mut self) -> Self {
133        Arc::make_mut(&mut self.config).adapter_instrumentation = false;
134        self
135    }
136
137    /// Caps distinct low-cardinality labels per metric family.
138    pub fn max_series(mut self, max_series: usize) -> Self {
139        Arc::make_mut(&mut self.config).max_series = Some(max_series);
140        #[cfg(feature = "http")]
141        {
142            self.http_metrics = self.http_metrics.with_max_series(max_series);
143        }
144        self
145    }
146
147    /// Excludes an HTTP route label from HTTP metrics.
148    pub fn exclude_route(mut self, route: impl Into<String>) -> Self {
149        let route = route.into();
150        Arc::make_mut(&mut self.config)
151            .excluded_routes
152            .insert(route.clone());
153        #[cfg(feature = "http")]
154        {
155            self.http_metrics = self.http_metrics.exclude_route(route);
156        }
157        self
158    }
159
160    /// Returns the service name.
161    pub fn service_name(&self) -> &str {
162        &self.config.service_name
163    }
164
165    /// Returns the configured service version label.
166    pub fn version_label(&self) -> Option<&str> {
167        self.config.version.as_deref()
168    }
169
170    /// Returns the configured deployment environment label.
171    pub fn environment_label(&self) -> Option<&str> {
172        self.config.environment.as_deref()
173    }
174
175    /// Returns whether HTTP metrics are enabled.
176    pub fn http_metrics_enabled(&self) -> bool {
177        self.config.prometheus && self.config.http_metrics
178    }
179
180    /// Returns whether Prometheus exposition is enabled.
181    pub fn prometheus_enabled(&self) -> bool {
182        self.config.prometheus
183    }
184
185    /// Returns whether HTTP tracing should be applied by Nidus.
186    pub fn tracing_enabled(&self) -> bool {
187        self.config.tracing
188    }
189
190    /// Returns OpenTelemetry resource config when built with `otel`.
191    #[cfg(feature = "otel")]
192    pub fn otel_config(&self) -> Option<&nidus_http::otel::OtelConfig> {
193        self.config.otel.as_ref()
194    }
195
196    /// Returns the underlying HTTP Prometheus collector.
197    #[cfg(feature = "http")]
198    pub fn prometheus_metrics(&self) -> PrometheusMetrics {
199        self.http_metrics.clone()
200    }
201
202    /// Creates a Tower HTTP metrics layer.
203    #[cfg(feature = "http")]
204    pub fn http_layer(&self) -> MetricsLayer<ObservabilityHttpMetricsHook> {
205        metrics_layer(self.http_metrics_hook())
206    }
207
208    /// Creates a `/metrics` router for Prometheus text exposition.
209    #[cfg(feature = "http")]
210    pub fn routes(&self) -> nidus_http::Router {
211        if !self.config.prometheus {
212            return nidus_http::Router::new();
213        }
214        let observability = self.clone();
215        nidus_http::Router::new().route(
216            "/metrics",
217            axum::routing::get(move || {
218                let observability = observability.clone();
219                async move { observability.render_prometheus() }
220            }),
221        )
222    }
223
224    /// Creates an event observer for [`nidus_events::ObservedEventBus`].
225    #[cfg(feature = "events")]
226    pub fn event_observer(&self) -> ObservabilityEventObserver {
227        ObservabilityEventObserver {
228            observability: self.clone(),
229        }
230    }
231
232    /// Creates an observed in-process event bus wired to this observability object.
233    #[cfg(feature = "events")]
234    pub fn observed_event_bus<T>(&self) -> ObservedEventBus<T, ObservabilityEventObserver>
235    where
236        T: Clone + Send + Sync + 'static,
237    {
238        EventBus::new().observed(self.event_observer())
239    }
240
241    /// Creates a job observer for [`nidus_jobs::ObservedJobRunner`].
242    #[cfg(feature = "jobs")]
243    pub fn job_observer(&self) -> ObservabilityJobObserver {
244        ObservabilityJobObserver {
245            observability: self.clone(),
246        }
247    }
248
249    /// Creates an observed job runner wired to this observability object.
250    #[cfg(feature = "jobs")]
251    pub fn job_runner(&self) -> ObservedJobRunner<ObservabilityJobObserver> {
252        ObservedJobRunner::new(self.job_observer())
253    }
254
255    /// Creates an adapter observer for Nidus-owned adapter operations.
256    pub fn adapter_observer(&self) -> ObservabilityAdapterObserver {
257        ObservabilityAdapterObserver {
258            observability: self.clone(),
259        }
260    }
261
262    /// Runs a future inside a named tracing span.
263    pub async fn instrument<Fut, T>(
264        &self,
265        operation: impl Into<Cow<'static, str>>,
266        future: Fut,
267    ) -> T
268    where
269        Fut: Future<Output = T>,
270    {
271        let operation = operation.into();
272        future
273            .instrument(tracing::info_span!(
274                "operation",
275                otel.name = %operation,
276                service.name = %self.service_name(),
277                service.version = self.version_label(),
278                deployment.environment = self.environment_label()
279            ))
280            .await
281    }
282
283    /// Records module graph validation owned by Nidus application bootstrap.
284    pub fn record_module_graph_validation(&self, status: OperationStatus, duration: Duration) {
285        let span = tracing::info_span!(
286            "module.graph.validate",
287            status = status.as_str(),
288            duration_ms = duration.as_millis()
289        );
290        let _entered = span.enter();
291        self.record_lifecycle_operation("module.graph.validate", status, duration);
292    }
293
294    /// Records application lifecycle operation telemetry.
295    pub fn record_lifecycle_operation(
296        &self,
297        operation: &'static str,
298        status: OperationStatus,
299        duration: Duration,
300    ) {
301        let mut state = lock_state(&self.state);
302        let operation = state.intern_label("lifecycle", operation, self.config.max_series);
303        let status = status.as_str();
304        *state
305            .lifecycle_total
306            .entry((Arc::clone(&operation), status))
307            .or_default() += 1;
308        state
309            .lifecycle_duration
310            .entry((operation, status))
311            .or_default()
312            .observe(duration);
313    }
314
315    /// Renders all configured Prometheus metrics.
316    pub fn render_prometheus(&self) -> String {
317        if !self.config.prometheus {
318            return String::new();
319        }
320        let mut output = String::new();
321        #[cfg(feature = "http")]
322        {
323            output.push_str(&self.http_metrics.render());
324        }
325        let state = lock_state(&self.state).clone();
326        output.push_str(&render_observability_metrics(&state));
327        output
328    }
329
330    #[cfg(feature = "http")]
331    fn http_metrics_hook(&self) -> ObservabilityHttpMetricsHook {
332        ObservabilityHttpMetricsHook {
333            enabled: self.http_metrics_enabled(),
334            metrics: self.http_metrics.clone(),
335        }
336    }
337
338    #[cfg(feature = "events")]
339    fn record_event(&self, event_name: &str) {
340        if !(self.config.prometheus && self.config.event_metrics) {
341            return;
342        }
343        let mut state = lock_state(&self.state);
344        let event_name = state.intern_label("events", event_name, self.config.max_series);
345        *state.events_published.entry(event_name).or_default() += 1;
346    }
347
348    #[cfg(feature = "jobs")]
349    fn record_job_started(&self, job_name: &'static str) {
350        if !(self.config.prometheus && self.config.job_metrics) {
351            return;
352        }
353        let mut state = lock_state(&self.state);
354        let job_name = state.intern_label("jobs", job_name, self.config.max_series);
355        *state.jobs_started.entry(job_name).or_default() += 1;
356    }
357
358    #[cfg(feature = "jobs")]
359    fn record_job_finished(
360        &self,
361        job_name: &'static str,
362        status: JobStatusLabel,
363        duration: Option<Duration>,
364    ) {
365        if !(self.config.prometheus && self.config.job_metrics) {
366            return;
367        }
368        let mut state = lock_state(&self.state);
369        let job_name = state.intern_label("jobs", job_name, self.config.max_series);
370        let status = status.as_str();
371        *state
372            .jobs_finished
373            .entry((Arc::clone(&job_name), status))
374            .or_default() += 1;
375        if let Some(duration) = duration {
376            state
377                .job_duration
378                .entry((job_name, status))
379                .or_default()
380                .observe(duration);
381        }
382    }
383
384    fn record_adapter(
385        &self,
386        adapter: &'static str,
387        operation: &'static str,
388        status: OperationStatus,
389        duration: Duration,
390    ) {
391        if !(self.config.prometheus && self.config.adapter_instrumentation) {
392            return;
393        }
394        let span = tracing::info_span!(
395            "adapter.operation",
396            adapter.name = adapter,
397            operation.name = operation,
398            status = status.as_str(),
399            duration_ms = duration.as_millis()
400        );
401        let _entered = span.enter();
402        let mut state = lock_state(&self.state);
403        let series = state.adapter_series(adapter, operation, self.config.max_series);
404        let status = status.as_str();
405        *state
406            .adapter_operations
407            .entry((series, status))
408            .or_default() += 1;
409        state
410            .adapter_duration
411            .entry((series, status))
412            .or_default()
413            .observe(duration);
414    }
415}
416
417/// Extension methods that apply [`Observability`] to [`ApiDefaults`].
418#[cfg(feature = "http")]
419pub trait ApiDefaultsObservabilityExt {
420    /// Installs HTTP metrics from an observability object when enabled.
421    fn observability(self, observability: &Observability) -> Self;
422
423    /// Applies API defaults and merges observability routes such as `/metrics`.
424    fn apply_with_observability(
425        self,
426        router: nidus_http::Router,
427        observability: &Observability,
428    ) -> nidus_http::Router;
429}
430
431#[cfg(feature = "http")]
432impl ApiDefaultsObservabilityExt for ApiDefaults {
433    fn observability(self, observability: &Observability) -> Self {
434        if observability.http_metrics_enabled() {
435            self.metrics(observability.prometheus_metrics())
436        } else {
437            self
438        }
439    }
440
441    fn apply_with_observability(
442        self,
443        router: nidus_http::Router,
444        observability: &Observability,
445    ) -> nidus_http::Router {
446        self.observability(observability)
447            .apply(router)
448            .merge(observability.routes())
449    }
450}
451
452#[derive(Clone, Debug)]
453struct ObservabilityConfig {
454    service_name: String,
455    version: Option<String>,
456    environment: Option<String>,
457    prometheus: bool,
458    http_metrics: bool,
459    event_metrics: bool,
460    job_metrics: bool,
461    adapter_instrumentation: bool,
462    tracing: bool,
463    max_series: Option<usize>,
464    excluded_routes: BTreeSet<String>,
465    #[cfg(feature = "otel")]
466    otel: Option<nidus_http::otel::OtelConfig>,
467}
468
469#[derive(Clone, Debug, Default)]
470struct ObservabilityState {
471    known_labels: BTreeMap<&'static str, BTreeSet<Arc<str>>>,
472    known_adapter_series: BTreeSet<AdapterSeries>,
473    events_published: BTreeMap<Arc<str>, u64>,
474    jobs_started: BTreeMap<Arc<str>, u64>,
475    jobs_finished: BTreeMap<(Arc<str>, &'static str), u64>,
476    job_duration: BTreeMap<(Arc<str>, &'static str), DurationHistogram>,
477    lifecycle_total: BTreeMap<(Arc<str>, &'static str), u64>,
478    lifecycle_duration: BTreeMap<(Arc<str>, &'static str), DurationHistogram>,
479    adapter_operations: BTreeMap<(AdapterSeries, &'static str), u64>,
480    adapter_duration: BTreeMap<(AdapterSeries, &'static str), DurationHistogram>,
481}
482
483static OVERFLOW_LABEL: LazyLock<Arc<str>> = LazyLock::new(|| Arc::from("<overflow>"));
484
485impl ObservabilityState {
486    fn intern_label(
487        &mut self,
488        family: &'static str,
489        label: &str,
490        max_series: Option<usize>,
491    ) -> Arc<str> {
492        let labels = self.known_labels.entry(family).or_default();
493        if let Some(label) = labels.get(label) {
494            return Arc::clone(label);
495        }
496        if max_series.is_some_and(|max| labels.len() >= max) {
497            return Arc::clone(&OVERFLOW_LABEL);
498        }
499
500        let label: Arc<str> = Arc::from(label);
501        labels.insert(Arc::clone(&label));
502        label
503    }
504
505    fn adapter_series(
506        &mut self,
507        adapter: &'static str,
508        operation: &'static str,
509        max_series: Option<usize>,
510    ) -> AdapterSeries {
511        let series = AdapterSeries { adapter, operation };
512        if self.known_adapter_series.contains(&series) {
513            return series;
514        }
515        if max_series.is_some_and(|max| self.known_adapter_series.len() >= max) {
516            return AdapterSeries::OVERFLOW;
517        }
518
519        self.known_adapter_series.insert(series);
520        series
521    }
522}
523
524#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
525struct AdapterSeries {
526    adapter: &'static str,
527    operation: &'static str,
528}
529
530impl AdapterSeries {
531    const OVERFLOW: Self = Self {
532        adapter: "<overflow>",
533        operation: "<overflow>",
534    };
535}
536
537/// HTTP metrics hook returned by [`Observability::http_layer`].
538#[cfg(feature = "http")]
539#[derive(Clone, Debug)]
540pub struct ObservabilityHttpMetricsHook {
541    enabled: bool,
542    metrics: PrometheusMetrics,
543}
544
545#[cfg(feature = "http")]
546impl HttpMetricsHook for ObservabilityHttpMetricsHook {
547    fn on_request(&self, method: &http::Method, route: Option<&str>) {
548        if self.enabled {
549            self.metrics.on_request(method, route);
550        }
551    }
552
553    fn on_response(
554        &self,
555        method: &http::Method,
556        route: Option<&str>,
557        status: http::StatusCode,
558        latency: Duration,
559    ) {
560        if self.enabled {
561            self.metrics.on_response(method, route, status, latency);
562        }
563    }
564
565    fn on_error(&self, method: &http::Method, route: Option<&str>, latency: Duration) {
566        if self.enabled {
567            self.metrics.on_error(method, route, latency);
568        }
569    }
570}
571
572/// Event observer that records low-cardinality event publication metrics.
573#[cfg(feature = "events")]
574#[derive(Clone, Debug)]
575pub struct ObservabilityEventObserver {
576    observability: Observability,
577}
578
579#[cfg(feature = "events")]
580impl<T> EventObserver<T> for ObservabilityEventObserver
581where
582    T: Clone + Send + Sync + 'static,
583{
584    fn on_event_published(&self, context: &ObservedEventContext) {
585        let span = tracing::info_span!(
586            "event.publish",
587            event.name = context.event_name(),
588            event.operation_id = context.operation_id()
589        );
590        let _entered = span.enter();
591        self.observability.record_event(context.event_name());
592    }
593}
594
595/// Job observer that records starts, completions, and durations.
596#[cfg(feature = "jobs")]
597#[derive(Clone, Debug)]
598pub struct ObservabilityJobObserver {
599    observability: Observability,
600}
601
602#[cfg(feature = "jobs")]
603impl JobObserver for ObservabilityJobObserver {
604    fn on_job_started(&self, context: &ObservedJobContext) {
605        self.observability.record_job_started(context.job_name());
606    }
607
608    fn on_job_finished(&self, context: &ObservedJobContext, status: JobResultStatus) {
609        let status = match status {
610            JobResultStatus::Success => JobStatusLabel::Success,
611            JobResultStatus::Failure => JobStatusLabel::Failure,
612        };
613        self.observability
614            .record_job_finished(context.job_name(), status, context.duration());
615    }
616}
617
618/// Adapter observer for Nidus-owned adapter operations.
619#[derive(Clone, Debug)]
620pub struct ObservabilityAdapterObserver {
621    observability: Observability,
622}
623
624impl ObservabilityAdapterObserver {
625    /// Records an adapter-owned operation with a measured duration.
626    pub fn record(
627        &self,
628        adapter: &'static str,
629        operation: &'static str,
630        status: OperationStatus,
631        duration: Duration,
632    ) {
633        self.observability
634            .record_adapter(adapter, operation, status, duration);
635    }
636
637    /// Runs a synchronous adapter operation and records success or failure.
638    pub fn observe_result<T, E>(
639        &self,
640        adapter: &'static str,
641        operation: &'static str,
642        run: impl FnOnce() -> std::result::Result<T, E>,
643    ) -> std::result::Result<T, E> {
644        let started_at = Instant::now();
645        let result = run();
646        self.record(
647            adapter,
648            operation,
649            OperationStatus::from_success(result.is_ok()),
650            started_at.elapsed(),
651        );
652        result
653    }
654}
655
656/// Operation status label used by lifecycle and adapter metrics.
657#[derive(Clone, Copy, Debug, Eq, PartialEq)]
658pub enum OperationStatus {
659    /// The operation succeeded.
660    Success,
661    /// The operation failed.
662    Failure,
663}
664
665impl OperationStatus {
666    /// Returns the stable Prometheus label value.
667    pub const fn as_str(self) -> &'static str {
668        match self {
669            Self::Success => "success",
670            Self::Failure => "failure",
671        }
672    }
673
674    /// Creates a status from a boolean success flag.
675    pub const fn from_success(ok: bool) -> Self {
676        if ok { Self::Success } else { Self::Failure }
677    }
678}
679
680impl From<bool> for OperationStatus {
681    fn from(ok: bool) -> Self {
682        Self::from_success(ok)
683    }
684}
685
686#[cfg(feature = "jobs")]
687#[derive(Clone, Copy, Debug, Eq, PartialEq)]
688enum JobStatusLabel {
689    Success,
690    Failure,
691}
692
693#[cfg(feature = "jobs")]
694impl JobStatusLabel {
695    const fn as_str(self) -> &'static str {
696        match self {
697            Self::Success => "success",
698            Self::Failure => "failure",
699        }
700    }
701}
702
703#[derive(Clone, Debug, Default)]
704struct DurationHistogram {
705    count: u64,
706    sum: f64,
707    bucket_counts: [u64; DURATION_BUCKETS.len()],
708}
709
710impl DurationHistogram {
711    fn observe(&mut self, duration: Duration) {
712        let seconds = duration.as_secs_f64();
713        self.count += 1;
714        self.sum += seconds;
715        for (bucket, count) in DURATION_BUCKETS.iter().zip(self.bucket_counts.iter_mut()) {
716            if seconds <= *bucket {
717                *count += 1;
718            }
719        }
720    }
721}
722
723const DURATION_BUCKETS: [f64; 11] = [
724    0.005, 0.010, 0.025, 0.050, 0.100, 0.250, 0.500, 1.000, 2.500, 5.000, 10.000,
725];
726
727fn render_observability_metrics(state: &ObservabilityState) -> String {
728    let mut output = String::new();
729    output.push_str("# TYPE nidus_events_published_total counter\n");
730    for (event, count) in &state.events_published {
731        output.push_str(&format!(
732            "nidus_events_published_total{{event=\"{}\"}} {}\n",
733            escape_label(event),
734            count
735        ));
736    }
737    output.push_str("# TYPE nidus_jobs_started_total counter\n");
738    for (job, count) in &state.jobs_started {
739        output.push_str(&format!(
740            "nidus_jobs_started_total{{job=\"{}\"}} {}\n",
741            escape_label(job),
742            count
743        ));
744    }
745    output.push_str("# TYPE nidus_jobs_finished_total counter\n");
746    for ((job, status), count) in &state.jobs_finished {
747        output.push_str(&format!(
748            "nidus_jobs_finished_total{{job=\"{}\",status=\"{}\"}} {}\n",
749            escape_label(job),
750            escape_label(status),
751            count
752        ));
753    }
754    render_histogram(
755        &mut output,
756        "nidus_job_duration_seconds",
757        &["job", "status"],
758        state
759            .job_duration
760            .iter()
761            .map(|((job, status), histogram)| (vec![job.as_ref(), *status], histogram)),
762    );
763    output.push_str("# TYPE nidus_lifecycle_total counter\n");
764    for ((operation, status), count) in &state.lifecycle_total {
765        output.push_str(&format!(
766            "nidus_lifecycle_total{{operation=\"{}\",status=\"{}\"}} {}\n",
767            escape_label(operation),
768            escape_label(status),
769            count
770        ));
771    }
772    render_histogram(
773        &mut output,
774        "nidus_lifecycle_duration_seconds",
775        &["operation", "status"],
776        state
777            .lifecycle_duration
778            .iter()
779            .map(|((operation, status), histogram)| (vec![operation.as_ref(), *status], histogram)),
780    );
781    output.push_str("# TYPE nidus_adapter_operations_total counter\n");
782    for ((series, status), count) in &state.adapter_operations {
783        output.push_str(&format!(
784            "nidus_adapter_operations_total{{adapter=\"{}\",operation=\"{}\",status=\"{}\"}} {}\n",
785            escape_label(series.adapter),
786            escape_label(series.operation),
787            escape_label(status),
788            count
789        ));
790    }
791    render_histogram(
792        &mut output,
793        "nidus_adapter_operation_duration_seconds",
794        &["adapter", "operation", "status"],
795        state
796            .adapter_duration
797            .iter()
798            .map(|((series, status), histogram)| {
799                (vec![series.adapter, series.operation, *status], histogram)
800            }),
801    );
802    output
803}
804
805fn render_histogram<'a>(
806    output: &mut String,
807    name: &str,
808    label_names: &[&str],
809    histograms: impl Iterator<Item = (Vec<&'a str>, &'a DurationHistogram)>,
810) {
811    output.push_str(&format!("# TYPE {name} histogram\n"));
812    for (label_values, histogram) in histograms {
813        for (bucket, count) in DURATION_BUCKETS.iter().zip(histogram.bucket_counts.iter()) {
814            output.push_str(&format!(
815                "{name}_bucket{{{},le=\"{}\"}} {}\n",
816                render_labels(label_names, &label_values),
817                format_bucket(*bucket),
818                count
819            ));
820        }
821        output.push_str(&format!(
822            "{name}_bucket{{{},le=\"+Inf\"}} {}\n",
823            render_labels(label_names, &label_values),
824            histogram.count
825        ));
826        output.push_str(&format!(
827            "{name}_count{{{}}} {}\n",
828            render_labels(label_names, &label_values),
829            histogram.count
830        ));
831        output.push_str(&format!(
832            "{name}_sum{{{}}} {:.6}\n",
833            render_labels(label_names, &label_values),
834            histogram.sum
835        ));
836    }
837}
838
839fn render_labels(names: &[&str], values: &[&str]) -> String {
840    names
841        .iter()
842        .zip(values.iter())
843        .map(|(name, value)| format!("{name}=\"{}\"", escape_label(value)))
844        .collect::<Vec<_>>()
845        .join(",")
846}
847
848fn format_bucket(bucket: f64) -> String {
849    format!("{bucket:.3}")
850}
851
852fn escape_label(value: &str) -> String {
853    value
854        .replace('\\', r"\\")
855        .replace('\n', r"\n")
856        .replace('"', r#"\""#)
857}
858
859fn lock_state(state: &Mutex<ObservabilityState>) -> std::sync::MutexGuard<'_, ObservabilityState> {
860    state
861        .lock()
862        .unwrap_or_else(|poisoned| poisoned.into_inner())
863}
864
865#[cfg(test)]
866mod tests {
867    use std::sync::Arc;
868
869    use super::{AdapterSeries, ObservabilityState};
870
871    #[test]
872    fn state_reuses_interned_and_overflow_labels() {
873        let mut state = ObservabilityState::default();
874        let first = state.intern_label("events", "orders.created", None);
875        let repeated = state.intern_label("events", "orders.created", None);
876        assert!(Arc::ptr_eq(&first, &repeated));
877
878        let first_overflow = state.intern_label("events", "orders.updated", Some(1));
879        let second_overflow = state.intern_label("events", "orders.deleted", Some(1));
880        assert_eq!(&*first_overflow, "<overflow>");
881        assert!(Arc::ptr_eq(&first_overflow, &second_overflow));
882    }
883
884    #[test]
885    fn adapter_series_preserve_label_boundaries_and_share_overflow() {
886        let mut state = ObservabilityState::default();
887        let first = state.adapter_series("adapter:primary", "get", Some(2));
888        let second = state.adapter_series("adapter", "primary:get", Some(2));
889        assert_ne!(first, second);
890        assert_eq!(
891            state.adapter_series("adapter:primary", "get", Some(2)),
892            first
893        );
894
895        let overflow = state.adapter_series("adapter", "set", Some(2));
896        assert_eq!(overflow, AdapterSeries::OVERFLOW);
897    }
898}
nidus_observability/lib.rs

nidus_observability/
lib.rs
