microclaw_engine/internal/observability/
logs.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3use std::time::{Duration, SystemTime, UNIX_EPOCH};
4
5use opentelemetry::logs::{
6    AnyValue as OtelAnyValue, LogRecord, Logger, LoggerProvider, Severity as OtelSeverity,
7};
8use opentelemetry::KeyValue;
9use opentelemetry_otlp::{Protocol, WithExportConfig, WithHttpConfig};
10use opentelemetry_proto::tonic::logs::v1::SeverityNumber;
11use opentelemetry_sdk::logs::log_processor_with_async_runtime::BatchLogProcessor;
12use opentelemetry_sdk::logs::{BatchConfigBuilder, SdkLogger, SdkLoggerProvider};
13use opentelemetry_sdk::runtime;
14use serde_yaml::Value;
15use tracing::{debug, info, warn};
16
17use crate::internal::observability::sdk::{get_bool, get_trimmed, parse_headers, OTelSdkContext};
18
19#[derive(Debug, Clone)]
20pub struct OtlpLogRecord {
21    pub timestamp_unix_nano: u64,
22    pub severity_number: SeverityNumber,
23    pub severity_text: String,
24    pub body: String,
25    pub attributes: Vec<KeyValue>,
26}
27
28#[derive(Clone)]
29pub struct OtlpLogExporter {
30    logger: Arc<SdkLogger>,
31    provider: Arc<SdkLoggerProvider>,
32}
33
34impl OtlpLogExporter {
35    pub fn from_observability(observability: Option<&Value>) -> Option<Arc<Self>> {
36        let map = observability?.as_mapping()?;
37        let enabled = get_bool(map, "otlp_logs_enabled").unwrap_or(false);
38        if !enabled {
39            debug!("otlp log exporter disabled by config");
40            return None;
41        }
42        let endpoint = get_trimmed(map, "otlp_logs_endpoint")?.to_string();
43        let sdk = OTelSdkContext::from_observability(map);
44        let headers = parse_headers(map);
45
46        let mut exporter_builder = opentelemetry_otlp::LogExporter::builder()
47            .with_http()
48            .with_protocol(Protocol::HttpBinary)
49            .with_http_client(reqwest::Client::new())
50            .with_endpoint(endpoint.clone());
51        if !headers.is_empty() {
52            exporter_builder =
53                exporter_builder.with_headers(headers.into_iter().collect::<HashMap<_, _>>());
54        }
55        let exporter = match exporter_builder.build() {
56            Ok(exporter) => exporter,
57            Err(err) => {
58                warn!(
59                    endpoint = %endpoint,
60                    "failed to build otlp log exporter: {}",
61                    err
62                );
63                return None;
64            }
65        };
66        let batch_config = BatchConfigBuilder::default()
67            .with_max_queue_size(8192)
68            .with_max_export_batch_size(512)
69            .with_scheduled_delay(Duration::from_millis(1000))
70            .build();
71        let processor = BatchLogProcessor::builder(exporter, runtime::Tokio)
72            .with_batch_config(batch_config)
73            .build();
74        let provider = SdkLoggerProvider::builder()
75            .with_resource(sdk.resource)
76            .with_log_processor(processor)
77            .build();
78        let logger = provider.logger("microclaw.observability.logs");
79        info!(
80            endpoint = %endpoint,
81            service_name = %sdk.service_name,
82            "otlp log exporter initialized"
83        );
84
85        Some(Arc::new(Self {
86            logger: Arc::new(logger),
87            provider: Arc::new(provider),
88        }))
89    }
90
91    pub fn send_log(&self, record: OtlpLogRecord) {
92        let mut log_record = self.logger.create_log_record();
93        log_record.set_timestamp(unix_nano_to_time(record.timestamp_unix_nano));
94        log_record.set_observed_timestamp(SystemTime::now());
95        log_record.set_severity_number(to_otel_severity(record.severity_number));
96        log_record.set_body(OtelAnyValue::from(record.body));
97        if !record.severity_text.is_empty() {
98            log_record.add_attribute("severity.text", record.severity_text);
99        }
100        for kv in record.attributes {
101            log_record.add_attribute(kv.key, kv.value.to_string());
102        }
103        self.logger.emit(log_record);
104        debug!("log record submitted to otel sdk");
105    }
106
107    pub fn logger_provider(&self) -> Arc<SdkLoggerProvider> {
108        self.provider.clone()
109    }
110}
111
112impl Drop for OtlpLogExporter {
113    fn drop(&mut self) {
114        if let Err(err) = self.provider.force_flush() {
115            warn!("otlp log exporter force_flush failed: {}", err);
116        } else {
117            debug!("otlp log exporter force_flush succeeded");
118        }
119    }
120}
121
122fn to_otel_severity(severity: SeverityNumber) -> OtelSeverity {
123    match severity as i32 {
124        1..=4 => OtelSeverity::Trace,
125        5..=8 => OtelSeverity::Debug,
126        9..=12 => OtelSeverity::Info,
127        13..=16 => OtelSeverity::Warn,
128        17..=20 => OtelSeverity::Error,
129        21..=24 => OtelSeverity::Fatal,
130        _ => OtelSeverity::Info,
131    }
132}
133
134fn unix_nano_to_time(value: u64) -> SystemTime {
135    UNIX_EPOCH + Duration::from_nanos(value)
136}
microclaw_engine/internal/observability/logs.rs

microclaw_engine/internal/observability/
logs.rs
