nemo_relay/observability/
otel.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! OpenTelemetry subscriber support for NeMo Relay.
5//!
6//! This crate adapts NeMo Relay lifecycle events into OpenTelemetry trace spans:
7//!
8//! - scope/tool/LLM `Start` events open spans
9//! - matching `End` events close spans
10//! - `Mark` events become span events on the active parent span when possible
11//! - orphan marks fall back to zero-duration spans so they still reach OTLP
12//!
13//! The public API is intentionally small:
14//!
15//! - [`OpenTelemetryConfig`] configures the OTLP exporter and resource metadata
16//! - [`OpenTelemetrySubscriber`] exposes a NeMo Relay [`EventSubscriberFn`] and
17//!   convenience `register` / `deregister` / `force_flush` / `shutdown` methods
18
19use std::collections::HashMap;
20use std::sync::{Arc, Mutex};
21use std::time::{Duration, SystemTime, UNIX_EPOCH};
22
23use crate::api::event::Event;
24use crate::api::event::ScopeCategory;
25use crate::api::runtime::EventSubscriberFn;
26use crate::api::scope::ScopeType;
27use crate::api::subscriber::{deregister_subscriber, flush_subscribers, register_subscriber};
28use crate::error::FlowError;
29use chrono::{DateTime, Utc};
30use opentelemetry::trace::{
31    Span as _, SpanContext, SpanKind, TraceContextExt, Tracer, TracerProvider as _,
32};
33use opentelemetry::{Context, KeyValue};
34use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig};
35use opentelemetry_sdk::Resource;
36use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider, Span};
37use serde::Serialize;
38use uuid::Uuid;
39
40#[cfg(target_arch = "wasm32")]
41use async_trait::async_trait;
42#[cfg(target_arch = "wasm32")]
43use opentelemetry_http::{
44    Bytes, HttpClient, HttpError, Request as HttpRequest, Response as HttpResponse,
45};
46#[cfg(not(target_arch = "wasm32"))]
47use opentelemetry_otlp::WithTonicConfig;
48#[cfg(not(target_arch = "wasm32"))]
49use tokio::runtime::Handle;
50#[cfg(not(target_arch = "wasm32"))]
51use tonic::metadata::{MetadataKey, MetadataMap, MetadataValue};
52#[cfg(target_arch = "wasm32")]
53use wasm_bindgen::{JsCast, JsValue};
54#[cfg(target_arch = "wasm32")]
55use wasm_bindgen_futures::{JsFuture, spawn_local};
56#[cfg(target_arch = "wasm32")]
57use web_sys::{Request as WebRequest, RequestInit};
58
59/// Result type for the OpenTelemetry subscriber crate.
60pub type Result<T> = std::result::Result<T, OpenTelemetryError>;
61
62/// Errors produced while configuring or operating the OpenTelemetry subscriber.
63#[derive(Debug, thiserror::Error)]
64pub enum OpenTelemetryError {
65    /// The tonic gRPC exporter requires an active Tokio runtime.
66    #[error("the OTLP gRPC exporter requires an active Tokio runtime")]
67    MissingTokioRuntime,
68    /// The requested transport is not available on this target.
69    #[error("the OTLP {transport} transport is not supported on this target")]
70    UnsupportedTransport {
71        /// Human-readable transport label used in the error message.
72        transport: &'static str,
73    },
74    /// Failed to parse a configured gRPC metadata header.
75    #[error("invalid OTLP gRPC header {key:?}: {message}")]
76    InvalidGrpcHeader {
77        /// Header name that failed to parse.
78        key: String,
79        /// Parser failure message.
80        message: String,
81    },
82    /// Failed to build the OTLP exporter.
83    #[error("failed to build the OTLP exporter: {0}")]
84    ExporterBuild(String),
85    /// The underlying tracer provider returned an error.
86    #[error("OpenTelemetry tracer provider error: {0}")]
87    Provider(String),
88    /// Registration errors from the core runtime.
89    #[error(transparent)]
90    Core(#[from] FlowError),
91}
92
93/// Supported OTLP trace transports.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
95pub enum OtlpTransport {
96    /// OTLP/HTTP protobuf, typically `http://host:4318/v1/traces`.
97    #[default]
98    HttpBinary,
99    /// OTLP/gRPC, typically `http://host:4317`.
100    Grpc,
101}
102
103/// Configuration for the OpenTelemetry subscriber.
104#[derive(Debug, Clone)]
105pub struct OpenTelemetryConfig {
106    endpoint: Option<String>,
107    headers: HashMap<String, String>,
108    resource_attributes: HashMap<String, String>,
109    service_name: String,
110    service_namespace: Option<String>,
111    service_version: Option<String>,
112    instrumentation_scope: String,
113    timeout: Duration,
114    transport: OtlpTransport,
115}
116
117impl Default for OpenTelemetryConfig {
118    fn default() -> Self {
119        Self {
120            endpoint: None,
121            headers: HashMap::new(),
122            resource_attributes: HashMap::new(),
123            service_name: "nemo-relay".to_string(),
124            service_namespace: None,
125            service_version: None,
126            instrumentation_scope: "nemo-relay-otel".to_string(),
127            timeout: Duration::from_secs(3),
128            transport: OtlpTransport::HttpBinary,
129        }
130    }
131}
132
133impl OpenTelemetryConfig {
134    /// Creates an HTTP OTLP config for the given service name.
135    pub fn http_binary(service_name: impl Into<String>) -> Self {
136        Self {
137            service_name: service_name.into(),
138            transport: OtlpTransport::HttpBinary,
139            ..Self::default()
140        }
141    }
142
143    /// Creates a gRPC OTLP config for the given service name.
144    pub fn grpc(service_name: impl Into<String>) -> Self {
145        Self {
146            service_name: service_name.into(),
147            transport: OtlpTransport::Grpc,
148            ..Self::default()
149        }
150    }
151
152    /// Overrides the OTLP endpoint. If unset, exporter defaults and OTEL_* env vars apply.
153    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
154        self.endpoint = Some(endpoint.into());
155        self
156    }
157
158    /// Adds a header/metadata entry for the exporter.
159    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
160        self.headers.insert(key.into(), value.into());
161        self
162    }
163
164    /// Adds a resource attribute as a string key/value pair.
165    pub fn with_resource_attribute(
166        mut self,
167        key: impl Into<String>,
168        value: impl Into<String>,
169    ) -> Self {
170        self.resource_attributes.insert(key.into(), value.into());
171        self
172    }
173
174    /// Sets the OTLP request timeout.
175    pub fn with_timeout(mut self, timeout: Duration) -> Self {
176        self.timeout = timeout;
177        self
178    }
179
180    /// Sets the service namespace resource attribute.
181    pub fn with_service_namespace(mut self, namespace: impl Into<String>) -> Self {
182        self.service_namespace = Some(namespace.into());
183        self
184    }
185
186    /// Sets the service version resource attribute.
187    pub fn with_service_version(mut self, version: impl Into<String>) -> Self {
188        self.service_version = Some(version.into());
189        self
190    }
191
192    /// Sets the instrumentation scope name used for emitted spans.
193    pub fn with_instrumentation_scope(mut self, scope: impl Into<String>) -> Self {
194        self.instrumentation_scope = scope.into();
195        self
196    }
197}
198
199/// OpenTelemetry-backed NeMo Relay subscriber.
200#[derive(Clone)]
201pub struct OpenTelemetrySubscriber {
202    inner: Arc<Inner>,
203}
204
205struct Inner {
206    processor: Arc<Mutex<OtelEventProcessor>>,
207    subscriber: EventSubscriberFn,
208}
209
210impl OpenTelemetrySubscriber {
211    /// Builds a subscriber backed by a new OTLP tracer provider.
212    pub fn new(config: OpenTelemetryConfig) -> Result<Self> {
213        #[cfg(not(target_arch = "wasm32"))]
214        if config.transport == OtlpTransport::Grpc && tokio::runtime::Handle::try_current().is_err()
215        {
216            return Err(OpenTelemetryError::MissingTokioRuntime);
217        }
218        #[cfg(target_arch = "wasm32")]
219        if config.transport == OtlpTransport::Grpc {
220            return Err(OpenTelemetryError::UnsupportedTransport { transport: "gRPC" });
221        }
222
223        let provider = build_tracer_provider(&config)?;
224        Ok(Self::from_tracer_provider_with_scope(
225            provider,
226            config.instrumentation_scope,
227        ))
228    }
229
230    /// Builds a subscriber from an already-configured tracer provider.
231    pub fn from_tracer_provider(
232        provider: SdkTracerProvider,
233        instrumentation_scope: impl Into<String>,
234    ) -> Self {
235        Self::from_tracer_provider_with_scope(provider, instrumentation_scope.into())
236    }
237
238    fn from_tracer_provider_with_scope(
239        provider: SdkTracerProvider,
240        instrumentation_scope: String,
241    ) -> Self {
242        let processor = Arc::new(Mutex::new(OtelEventProcessor::new(
243            provider,
244            instrumentation_scope,
245        )));
246        let processor_for_callback = Arc::clone(&processor);
247        let subscriber: EventSubscriberFn = Arc::new(move |event: &Event| {
248            let Ok(mut guard) = processor_for_callback.lock() else {
249                // Observability should not take down the host process if the
250                // subscriber state was previously poisoned.
251                return;
252            };
253            guard.process(event);
254        });
255
256        Self {
257            inner: Arc::new(Inner {
258                processor,
259                subscriber,
260            }),
261        }
262    }
263
264    /// Returns the raw NeMo Relay subscriber callback for custom registration flows.
265    pub fn subscriber(&self) -> EventSubscriberFn {
266        Arc::clone(&self.inner.subscriber)
267    }
268
269    /// Registers this subscriber globally with the NeMo Relay runtime.
270    pub fn register(&self, name: &str) -> Result<()> {
271        register_subscriber(name, self.subscriber()).map_err(Into::into)
272    }
273
274    /// Deregisters a previously-registered global subscriber by name.
275    pub fn deregister(&self, name: &str) -> Result<bool> {
276        deregister_subscriber(name).map_err(Into::into)
277    }
278
279    /// Flushes finished spans through the underlying tracer provider.
280    pub fn force_flush(&self) -> Result<()> {
281        flush_subscribers()?;
282        let guard = self.inner.processor.lock().map_err(|_| {
283            OpenTelemetryError::Provider("the subscriber state lock was poisoned".to_string())
284        })?;
285        guard.force_flush()
286    }
287
288    /// Shuts down the underlying tracer provider.
289    ///
290    /// Call `deregister(...)` first if the subscriber is still registered with NeMo Relay.
291    pub fn shutdown(&self) -> Result<()> {
292        flush_subscribers()?;
293        let guard = self.inner.processor.lock().map_err(|_| {
294            OpenTelemetryError::Provider("the subscriber state lock was poisoned".to_string())
295        })?;
296        guard.shutdown()
297    }
298}
299
300#[cfg(target_arch = "wasm32")]
301#[derive(Debug, Clone, Copy, Default)]
302struct WasmHttpClient;
303
304#[cfg(target_arch = "wasm32")]
305#[async_trait]
306impl HttpClient for WasmHttpClient {
307    async fn send_bytes(
308        &self,
309        request: HttpRequest<Bytes>,
310    ) -> std::result::Result<HttpResponse<Bytes>, HttpError> {
311        let (parts, body) = request.into_parts();
312
313        let request = {
314            let request_url = parts.uri.to_string();
315            let init = RequestInit::new();
316            init.set_method(parts.method.as_str());
317            if !body.is_empty() {
318                let body_bytes = js_sys::Uint8Array::from(body.as_ref());
319                init.set_body_opt_u8_array(Some(&body_bytes));
320            }
321
322            let request =
323                WebRequest::new_with_str_and_init(&request_url, &init).map_err(js_error)?;
324            let request_headers = request.headers();
325            for (name, value) in &parts.headers {
326                let value = value
327                    .to_str()
328                    .map_err(|e| http_error(format!("invalid OTLP HTTP header {name}: {e}")))?;
329                request_headers
330                    .set(name.as_str(), value)
331                    .map_err(js_error)?;
332            }
333            request
334        };
335
336        let fetch_promise = if let Some(window) = web_sys::window() {
337            window.fetch_with_request(&request)
338        } else {
339            let global = js_sys::global();
340            let fetch = js_sys::Reflect::get(&global, &JsValue::from_str("fetch"))
341                .map_err(js_error)?
342                .dyn_into::<js_sys::Function>()
343                .map_err(js_error)?;
344            fetch.call1(&global, &request).map_err(js_error)?.into()
345        };
346        // Waiting on the fetch promise from a synchronous wasm call stack can deadlock
347        // Node/browser event processing, so dispatch the request asynchronously.
348        spawn_local(async move {
349            if let Err(error) = JsFuture::from(fetch_promise).await {
350                web_sys::console::warn_1(&JsValue::from_str(&format!(
351                    "OpenTelemetry OTLP/HTTP export failed: {error:?}"
352                )));
353            }
354        });
355
356        HttpResponse::builder()
357            .status(202)
358            .body(Bytes::new())
359            .map_err(|e| http_error(e.to_string()))
360    }
361}
362
363#[cfg(target_arch = "wasm32")]
364fn js_error(value: JsValue) -> HttpError {
365    http_error(
366        value
367            .as_string()
368            .unwrap_or_else(|| format!("JavaScript error: {value:?}")),
369    )
370}
371
372#[cfg(target_arch = "wasm32")]
373fn http_error(message: impl Into<String>) -> HttpError {
374    Box::new(std::io::Error::other(message.into()))
375}
376
377fn build_tracer_provider(config: &OpenTelemetryConfig) -> Result<SdkTracerProvider> {
378    let exporter = match config.transport {
379        OtlpTransport::HttpBinary => {
380            #[cfg(not(target_arch = "wasm32"))]
381            install_rustls_crypto_provider();
382            let mut builder = SpanExporter::builder()
383                .with_http()
384                .with_protocol(Protocol::HttpBinary)
385                .with_timeout(config.timeout);
386            if let Some(endpoint) = &config.endpoint {
387                builder = builder.with_endpoint(endpoint.clone());
388            }
389            if !config.headers.is_empty() {
390                builder = builder.with_headers(config.headers.clone());
391            }
392            #[cfg(target_arch = "wasm32")]
393            {
394                builder = builder.with_http_client(WasmHttpClient);
395            }
396            builder
397                .build()
398                .map_err(|e| OpenTelemetryError::ExporterBuild(e.to_string()))?
399        }
400        #[cfg(not(target_arch = "wasm32"))]
401        OtlpTransport::Grpc => {
402            let mut builder = SpanExporter::builder()
403                .with_tonic()
404                .with_protocol(Protocol::Grpc)
405                .with_timeout(config.timeout);
406            if let Some(endpoint) = &config.endpoint {
407                builder = builder.with_endpoint(endpoint.clone());
408            }
409            if !config.headers.is_empty() {
410                builder = builder.with_metadata(build_grpc_metadata(&config.headers)?);
411            }
412            builder
413                .build()
414                .map_err(|e| OpenTelemetryError::ExporterBuild(e.to_string()))?
415        }
416        #[cfg(target_arch = "wasm32")]
417        OtlpTransport::Grpc => {
418            return Err(OpenTelemetryError::UnsupportedTransport { transport: "gRPC" });
419        }
420    };
421
422    let mut resource_attributes = vec![KeyValue::new("service.name", config.service_name.clone())];
423    if let Some(service_namespace) = &config.service_namespace {
424        resource_attributes.push(KeyValue::new(
425            "service.namespace",
426            service_namespace.clone(),
427        ));
428    }
429    if let Some(service_version) = &config.service_version {
430        resource_attributes.push(KeyValue::new("service.version", service_version.clone()));
431    }
432    for (key, value) in &config.resource_attributes {
433        resource_attributes.push(KeyValue::new(key.clone(), value.clone()));
434    }
435
436    // Disable per-span attribute caps. Consumers may emit large attribute
437    // sets on long-running spans; the OTel SDK default (128) silently drops
438    // attributes added last in the span's lifecycle.
439    let builder = SdkTracerProvider::builder()
440        .with_resource(
441            Resource::builder_empty()
442                .with_attributes(resource_attributes)
443                .build(),
444        )
445        .with_max_attributes_per_span(u32::MAX)
446        .with_max_attributes_per_event(u32::MAX);
447
448    #[cfg(not(target_arch = "wasm32"))]
449    {
450        if Handle::try_current().is_ok() {
451            Ok(builder.with_batch_exporter(exporter).build())
452        } else {
453            Ok(builder.with_simple_exporter(exporter).build())
454        }
455    }
456    #[cfg(target_arch = "wasm32")]
457    {
458        Ok(builder.with_simple_exporter(exporter).build())
459    }
460}
461
462#[cfg(not(target_arch = "wasm32"))]
463fn install_rustls_crypto_provider() {
464    let _ = rustls::crypto::ring::default_provider().install_default();
465}
466
467#[cfg(not(target_arch = "wasm32"))]
468fn build_grpc_metadata(headers: &HashMap<String, String>) -> Result<MetadataMap> {
469    let mut metadata = MetadataMap::new();
470    for (key, value) in headers {
471        let metadata_key = MetadataKey::from_bytes(key.as_bytes()).map_err(|e| {
472            OpenTelemetryError::InvalidGrpcHeader {
473                key: key.clone(),
474                message: e.to_string(),
475            }
476        })?;
477        let metadata_value = MetadataValue::try_from(value.as_str()).map_err(|e| {
478            OpenTelemetryError::InvalidGrpcHeader {
479                key: key.clone(),
480                message: e.to_string(),
481            }
482        })?;
483        metadata.insert(metadata_key, metadata_value);
484    }
485    Ok(metadata)
486}
487
488struct ActiveSpan {
489    span: Span,
490    span_context: SpanContext,
491}
492
493struct OtelEventProcessor {
494    active_spans: HashMap<Uuid, ActiveSpan>,
495    provider: SdkTracerProvider,
496    tracer: SdkTracer,
497}
498
499impl OtelEventProcessor {
500    fn new(provider: SdkTracerProvider, instrumentation_scope: String) -> Self {
501        let tracer = provider.tracer(instrumentation_scope);
502        Self {
503            active_spans: HashMap::new(),
504            provider,
505            tracer,
506        }
507    }
508
509    fn process(&mut self, event: &Event) {
510        match event.scope_category() {
511            Some(ScopeCategory::Start) => self.process_start(event),
512            Some(ScopeCategory::End) => self.process_end(event),
513            None => self.process_mark(event),
514        }
515    }
516
517    fn force_flush(&self) -> Result<()> {
518        self.provider
519            .force_flush()
520            .map_err(|e| OpenTelemetryError::Provider(e.to_string()))
521    }
522
523    fn shutdown(&self) -> Result<()> {
524        self.provider
525            .shutdown()
526            .map_err(|e| OpenTelemetryError::Provider(e.to_string()))
527    }
528
529    fn process_start(&mut self, event: &Event) {
530        let mut span = self
531            .tracer
532            .span_builder(span_name(event))
533            .with_kind(span_kind(event))
534            .with_start_time(to_system_time(*event.timestamp()))
535            .start_with_context(&self.tracer, &self.parent_context(event));
536        span.set_attributes(start_attributes(event));
537        let span_context = local_parent_span_context(span.span_context());
538        self.active_spans
539            .insert(event.uuid(), ActiveSpan { span, span_context });
540    }
541
542    fn process_end(&mut self, event: &Event) {
543        let Some(mut active_span) = self.active_spans.remove(&event.uuid()) else {
544            return;
545        };
546        active_span.span.set_attributes(end_attributes(event));
547        active_span
548            .span
549            .end_with_timestamp(to_system_time(*event.timestamp()));
550    }
551
552    fn process_mark(&mut self, event: &Event) {
553        let mark_name = event.name().to_string();
554        let timestamp = to_system_time(*event.timestamp());
555        let attributes = mark_attributes(event);
556
557        if let Some(parent_span) = self.find_parent_span_mut(event) {
558            parent_span
559                .span
560                .add_event_with_timestamp(mark_name, timestamp, attributes);
561            return;
562        }
563
564        let mut span = self
565            .tracer
566            .span_builder(format!("mark:{mark_name}"))
567            .with_kind(SpanKind::Internal)
568            .with_start_time(timestamp)
569            .start_with_context(&self.tracer, &self.parent_context(event));
570        let mut span_attributes = attributes;
571        span_attributes.push(KeyValue::new("nemo_relay.mark.orphan", true));
572        span.set_attributes(span_attributes);
573        span.end_with_timestamp(timestamp);
574    }
575
576    fn parent_context(&self, event: &Event) -> Context {
577        self.find_parent_span(event)
578            .map(|active_span| {
579                Context::new().with_remote_span_context(active_span.span_context.clone())
580            })
581            .unwrap_or_default()
582    }
583
584    fn parent_span_uuid(&self, event: &Event) -> Option<Uuid> {
585        event
586            .parent_uuid()
587            .filter(|uuid| self.active_spans.contains_key(uuid))
588    }
589
590    fn find_parent_span(&self, event: &Event) -> Option<&ActiveSpan> {
591        self.parent_span_uuid(event)
592            .and_then(|uuid| self.active_spans.get(&uuid))
593    }
594
595    fn find_parent_span_mut(&mut self, event: &Event) -> Option<&mut ActiveSpan> {
596        self.parent_span_uuid(event)
597            .and_then(|uuid| self.active_spans.get_mut(&uuid))
598    }
599}
600
601fn span_kind(event: &Event) -> SpanKind {
602    match semantic_scope_type(event) {
603        Some(ScopeType::Llm) => SpanKind::Client,
604        Some(
605            ScopeType::Tool | ScopeType::Retriever | ScopeType::Embedder | ScopeType::Reranker,
606        ) => SpanKind::Client,
607        _ => SpanKind::Internal,
608    }
609}
610
611fn span_name(event: &Event) -> String {
612    event.name().to_string()
613}
614
615fn semantic_scope_type(event: &Event) -> Option<ScopeType> {
616    event.scope_type()
617}
618
619fn scope_type_name(scope_type: Option<ScopeType>) -> &'static str {
620    match scope_type {
621        Some(ScopeType::Agent) => "agent",
622        Some(ScopeType::Function) => "function",
623        Some(ScopeType::Tool) => "tool",
624        Some(ScopeType::Llm) => "llm",
625        Some(ScopeType::Retriever) => "retriever",
626        Some(ScopeType::Embedder) => "embedder",
627        Some(ScopeType::Reranker) => "reranker",
628        Some(ScopeType::Guardrail) => "guardrail",
629        Some(ScopeType::Evaluator) => "evaluator",
630        Some(ScopeType::Custom) => "custom",
631        Some(ScopeType::Unknown) | None => "unknown",
632    }
633}
634
635fn start_attributes(event: &Event) -> Vec<KeyValue> {
636    let mut attributes = common_attributes(event);
637    let handle_attributes = event.attributes();
638    push_serialized(
639        &mut attributes,
640        "nemo_relay.handle_attributes_json",
641        handle_attributes,
642    );
643    push_serialized(&mut attributes, "nemo_relay.start.data_json", event.data());
644    push_serialized(
645        &mut attributes,
646        "nemo_relay.start.metadata_json",
647        event.metadata(),
648    );
649    push_serialized(
650        &mut attributes,
651        "nemo_relay.start.input_json",
652        event.input(),
653    );
654    attributes
655}
656
657fn end_attributes(event: &Event) -> Vec<KeyValue> {
658    let mut attributes = Vec::new();
659    push_serialized(&mut attributes, "nemo_relay.end.data_json", event.data());
660    push_serialized(
661        &mut attributes,
662        "nemo_relay.end.metadata_json",
663        event.metadata(),
664    );
665    push_serialized(
666        &mut attributes,
667        "nemo_relay.end.output_json",
668        event.output(),
669    );
670    attributes
671}
672
673fn mark_attributes(event: &Event) -> Vec<KeyValue> {
674    let handle_attributes = event.attributes();
675    let mut attributes = vec![
676        KeyValue::new("nemo_relay.mark.uuid", event.uuid().to_string()),
677        KeyValue::new(
678            "nemo_relay.mark.parent_uuid",
679            event
680                .parent_uuid()
681                .map(|uuid| uuid.to_string())
682                .unwrap_or_default(),
683        ),
684    ];
685    push_serialized(
686        &mut attributes,
687        "nemo_relay.mark.attributes_json",
688        handle_attributes,
689    );
690    push_serialized(&mut attributes, "nemo_relay.mark.data_json", event.data());
691    push_serialized(
692        &mut attributes,
693        "nemo_relay.mark.metadata_json",
694        event.metadata(),
695    );
696    attributes
697}
698
699fn common_attributes(event: &Event) -> Vec<KeyValue> {
700    let mut attributes = vec![
701        KeyValue::new("nemo_relay.uuid", event.uuid().to_string()),
702        KeyValue::new(
703            "nemo_relay.parent_uuid",
704            event
705                .parent_uuid()
706                .map(|uuid| uuid.to_string())
707                .unwrap_or_default(),
708        ),
709        KeyValue::new(
710            "nemo_relay.scope_type",
711            scope_type_name(semantic_scope_type(event)),
712        ),
713    ];
714
715    if let Some(model_name) = event.model_name() {
716        attributes.push(KeyValue::new(
717            "nemo_relay.model_name",
718            model_name.to_string(),
719        ));
720    }
721    if let Some(tool_call_id) = event.tool_call_id() {
722        attributes.push(KeyValue::new(
723            "nemo_relay.tool_call_id",
724            tool_call_id.to_string(),
725        ));
726    }
727
728    attributes
729}
730
731fn push_serialized<T: Serialize + ?Sized>(
732    attributes: &mut Vec<KeyValue>,
733    key: &'static str,
734    value: Option<&T>,
735) {
736    if let Some(value) = value
737        && let Ok(json) = serde_json::to_string(value)
738    {
739        attributes.push(KeyValue::new(key, json));
740    }
741}
742
743fn local_parent_span_context(span_context: &SpanContext) -> SpanContext {
744    SpanContext::new(
745        span_context.trace_id(),
746        span_context.span_id(),
747        span_context.trace_flags(),
748        false,
749        span_context.trace_state().clone(),
750    )
751}
752
753fn to_system_time(timestamp: DateTime<Utc>) -> SystemTime {
754    let seconds = timestamp.timestamp();
755    let nanos = timestamp.timestamp_subsec_nanos();
756    if seconds >= 0 {
757        UNIX_EPOCH + Duration::new(seconds as u64, nanos)
758    } else if nanos == 0 {
759        UNIX_EPOCH - Duration::new(seconds.unsigned_abs(), 0)
760    } else {
761        UNIX_EPOCH - Duration::new(seconds.unsigned_abs() - 1, 1_000_000_000 - nanos)
762    }
763}
764
765#[cfg(test)]
766#[path = "../../tests/unit/observability/otel_tests.rs"]
767mod tests;
nemo_relay/observability/otel.rs

nemo_relay/observability/
otel.rs
