diff --git a/Cargo.lock b/Cargo.lock index 1964b5e421..53784e995c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4865,6 +4865,7 @@ dependencies = [ "http-body-util", "hyper-util", "metrics", + "metrics-util", "openshell-ocsf", "openshell-otel-test-support", "opentelemetry", @@ -5109,6 +5110,7 @@ dependencies = [ "libc", "metrics", "metrics-exporter-prometheus", + "metrics-util", "miette", "nix 0.29.0", "notify", diff --git a/Cargo.toml b/Cargo.toml index 5254c11284..541ab7f5ff 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -80,6 +80,7 @@ tracing-opentelemetry = { version = "0.34", default-features = false, features = # Metrics metrics = "0.24" metrics-exporter-prometheus = { version = "0.18", default-features = false, features = ["http-listener"] } +metrics-util = { version = "0.20", default-features = false, features = ["registry"] } # Unix/Process nix = { version = "0.29", features = ["signal", "process", "user", "fs", "term"] } diff --git a/crates/openshell-gateway/src/vm.rs b/crates/openshell-gateway/src/vm.rs index fcfc8ffe83..80c4034b3a 100644 --- a/crates/openshell-gateway/src/vm.rs +++ b/crates/openshell-gateway/src/vm.rs @@ -890,6 +890,7 @@ mod tests { }, }), ocsf: None, + metrics: None, agent: None, }), "production-us-west", @@ -922,6 +923,7 @@ mod tests { }, }), ocsf: None, + metrics: None, agent: None, }), "production-us-west", @@ -957,6 +959,7 @@ mod tests { }, schema_version: Some("2.0".into()), }), + metrics: None, agent: None, }), "production-us-west", @@ -989,6 +992,7 @@ mod tests { service_name: None, traces: None, ocsf: None, + metrics: None, agent: Some(AgentOtlpConfig { collector: OtlpCollectorConfig { endpoint: "http://agents.internal:4318".to_string(), diff --git a/crates/openshell-otel-test-support/Cargo.toml b/crates/openshell-otel-test-support/Cargo.toml index 05b67b233d..ed16cab262 100644 --- a/crates/openshell-otel-test-support/Cargo.toml +++ b/crates/openshell-otel-test-support/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "openshell-otel-test-support" -description = "Shared OTLP collector fixture for OpenShell tracing tests" +description = "Shared OTLP collector fixture for OpenShell telemetry tests" version.workspace = true edition.workspace = true rust-version.workspace = true @@ -17,7 +17,7 @@ http-body-util = { workspace = true } hyper = { workspace = true } hyper-util = { workspace = true, features = ["tokio"] } openshell-otel = { path = "../openshell-otel" } -opentelemetry-proto = { version = "0.33", default-features = false, features = ["gen-tonic", "logs", "trace"] } +opentelemetry-proto = { version = "0.33", default-features = false, features = ["gen-tonic", "logs", "metrics", "trace"] } prost = { workspace = true } tokio = { workspace = true } tokio-stream = { workspace = true } diff --git a/crates/openshell-otel-test-support/src/lib.rs b/crates/openshell-otel-test-support/src/lib.rs index ea34843649..5a970265c4 100644 --- a/crates/openshell-otel-test-support/src/lib.rs +++ b/crates/openshell-otel-test-support/src/lib.rs @@ -72,6 +72,10 @@ use opentelemetry_proto::tonic::collector::logs::v1::{ ExportLogsServiceRequest, ExportLogsServiceResponse, logs_service_server::{LogsService, LogsServiceServer}, }; +use opentelemetry_proto::tonic::collector::metrics::v1::{ + ExportMetricsServiceRequest, ExportMetricsServiceResponse, + metrics_service_server::{MetricsService, MetricsServiceServer}, +}; use opentelemetry_proto::tonic::collector::trace::v1::{ ExportTraceServiceRequest, ExportTraceServiceResponse, trace_service_server::{TraceService, TraceServiceServer}, @@ -144,6 +148,8 @@ pub struct ReceivedTraces { pub log_bodies: Vec, /// `authorization` metadata of each log export request. pub log_authorizations: Vec, + /// Names of exported metrics. + pub metric_names: Vec, pub service_names: Vec, pub gateway_names: Vec, pub compute_drivers: Vec, @@ -268,7 +274,32 @@ impl LogsService for Collector { } } -/// Loopback OTLP/gRPC server that captures exported spans, logs, and resources. +#[tonic::async_trait] +impl MetricsService for Collector { + async fn export( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + { + let mut received = self + .received + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for resource_metrics in request.into_inner().resource_metrics { + received.record_resource(resource_metrics.resource); + for scope_metrics in resource_metrics.scope_metrics { + received + .metric_names + .extend(scope_metrics.metrics.into_iter().map(|metric| metric.name)); + } + } + } + self.exported.notify_one(); + Ok(tonic::Response::new(ExportMetricsServiceResponse::default())) + } +} + +/// Loopback OTLP/gRPC server that captures exported spans, logs, metrics, and resources. pub struct OtlpTestServer { endpoint: String, received: Arc>, @@ -329,6 +360,8 @@ impl OtlpTestServer { } let trace_service = TraceServiceServer::new(collector.clone()) .accept_compressed(tonic::codec::CompressionEncoding::Gzip); + let metrics_service = MetricsServiceServer::new(collector.clone()) + .accept_compressed(tonic::codec::CompressionEncoding::Gzip); let logs_service = LogsServiceServer::new(collector) .accept_compressed(tonic::codec::CompressionEncoding::Gzip); let (shutdown, shutdown_rx) = tokio::sync::oneshot::channel(); @@ -336,6 +369,7 @@ impl OtlpTestServer { server .add_service(trace_service) .add_service(logs_service) + .add_service(metrics_service) .serve_with_incoming_shutdown( tokio_stream::wrappers::TcpListenerStream::new(listener), async { diff --git a/crates/openshell-otel/Cargo.toml b/crates/openshell-otel/Cargo.toml index b174f65871..9e976d9372 100644 --- a/crates/openshell-otel/Cargo.toml +++ b/crates/openshell-otel/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "openshell-otel" -description = "Shared OpenTelemetry trace and log export support for OpenShell services" +description = "Shared OpenTelemetry trace, log, and metric export support for OpenShell services" version.workspace = true edition.workspace = true rust-version.workspace = true @@ -18,11 +18,12 @@ http = { workspace = true } http-body-util = { workspace = true } hyper-util = { workspace = true, features = ["client-legacy", "http1", "tokio"] } metrics = { workspace = true } +metrics-util = { workspace = true } openshell-ocsf = { path = "../openshell-ocsf" } opentelemetry = { workspace = true } -opentelemetry_sdk = { workspace = true, features = ["logs"] } +opentelemetry_sdk = { workspace = true, features = ["logs", "metrics"] } opentelemetry-http = { workspace = true } -opentelemetry-otlp = { workspace = true, features = ["gzip-http", "http-proto", "logs"] } +opentelemetry-otlp = { workspace = true, features = ["gzip-http", "http-proto", "logs", "metrics"] } serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } diff --git a/crates/openshell-otel/src/lib.rs b/crates/openshell-otel/src/lib.rs index 47332df7f5..eeb30f92bb 100644 --- a/crates/openshell-otel/src/lib.rs +++ b/crates/openshell-otel/src/lib.rs @@ -1,11 +1,12 @@ // SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! Shared OpenTelemetry trace and log export support for `OpenShell` services. +//! Shared OpenTelemetry trace, log, and metric export support for `OpenShell` services. mod driver; mod grpc; mod logs; +mod meter; mod ocsf; mod propagation; mod transport; @@ -20,6 +21,7 @@ pub use grpc::{ RecordGrpcStatus, TracedGrpcStream, compute_driver_rpc_layer, compute_driver_rpc_operation, grpc_status_code_name, record_grpc_status, rpc, }; +pub use meter::MetricsBridge; pub use ocsf::{OcsfOtlp, OcsfOtlpLayer}; pub use propagation::{ HeaderMapExtractor, MetadataMapInjector, TRACEPARENT_ENV, TRACESTATE_ENV, @@ -37,6 +39,7 @@ pub use opentelemetry_otlp::LogExporter; use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig; use opentelemetry_otlp::{SpanExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; use opentelemetry_sdk::Resource; +pub use opentelemetry_sdk::metrics::SdkMeterProvider; use opentelemetry_sdk::trace::SdkTracer; pub use opentelemetry_sdk::trace::SdkTracerProvider; use tracing::Subscriber; diff --git a/crates/openshell-otel/src/meter.rs b/crates/openshell-otel/src/meter.rs new file mode 100644 index 0000000000..babb5938cc --- /dev/null +++ b/crates/openshell-otel/src/meter.rs @@ -0,0 +1,432 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OTLP metric export, over gRPC or HTTP/protobuf, and a `metrics` recorder +//! that feeds it. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, PoisonError}; + +use metrics::{ + Counter, CounterFn, Gauge, GaugeFn, Histogram, HistogramFn, Key, KeyName, Metadata, Recorder, + SharedString, Unit, +}; +use metrics_util::registry::{Registry, Storage}; +use opentelemetry::KeyValue; +use opentelemetry::metrics::Meter; +use opentelemetry_otlp::{MetricExporter, WithExportConfig, WithHttpConfig}; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; + +use crate::{OtlpHttpClient, OtlpTarget, Protocol, SetupError}; + +impl OtlpTarget { + /// Build a meter provider that exports metrics to this target on the + /// SDK's periodic interval (`OTEL_METRIC_EXPORT_INTERVAL`). + pub fn meter_provider(&self) -> Result { + let exporter = match self.protocol { + Protocol::Grpc => self + .configure(MetricExporter::builder().with_tonic()) + .build()?, + Protocol::HttpProtobuf => MetricExporter::builder() + .with_http() + .with_protocol(opentelemetry_otlp::Protocol::HttpBinary) + .with_endpoint(self.http_signal_url("v1/metrics")?) + .with_http_client(OtlpHttpClient::new(crate::transport::export_timeout( + "OTEL_EXPORTER_OTLP_METRICS_TIMEOUT", + ))?) + .build()?, + }; + Ok(SdkMeterProvider::builder() + .with_reader(PeriodicReader::builder(exporter).build()) + .with_resource(self.resource().clone()) + .build()) + } +} + +type Descriptions = Mutex, SharedString)>>; + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +/// A `metrics` recorder that records into OpenTelemetry instruments. +/// +/// Install it next to another recorder with `metrics_util::layers::FanoutBuilder`. +/// A counter only grows, so `absolute` below the current total is ignored. +/// A gauge exports its last value. Every histogram uses `histogram_boundaries`. +/// Descriptions apply to series registered after they are described. +pub struct MetricsBridge { + registry: Registry, + descriptions: Arc, +} + +impl MetricsBridge { + #[must_use] + pub fn new(meter: Meter, histogram_boundaries: Vec) -> Self { + let descriptions = Arc::new(Descriptions::default()); + let storage = Instruments { + meter, + histogram_boundaries, + descriptions: descriptions.clone(), + }; + Self { + registry: Registry::new(storage), + descriptions, + } + } + + fn describe(&self, name: KeyName, unit: Option, description: SharedString) { + lock(&self.descriptions).insert(name, (unit, description)); + } +} + +impl Recorder for MetricsBridge { + fn describe_counter(&self, name: KeyName, unit: Option, description: SharedString) { + self.describe(name, unit, description); + } + + fn describe_gauge(&self, name: KeyName, unit: Option, description: SharedString) { + self.describe(name, unit, description); + } + + fn describe_histogram(&self, name: KeyName, unit: Option, description: SharedString) { + self.describe(name, unit, description); + } + + fn register_counter(&self, key: &Key, _: &Metadata<'_>) -> Counter { + self.registry + .get_or_create_counter(key, |counter| Counter::from_arc(counter.clone())) + } + + fn register_gauge(&self, key: &Key, _: &Metadata<'_>) -> Gauge { + self.registry + .get_or_create_gauge(key, |gauge| Gauge::from_arc(gauge.clone())) + } + + fn register_histogram(&self, key: &Key, _: &Metadata<'_>) -> Histogram { + self.registry + .get_or_create_histogram(key, |histogram| Histogram::from_arc(histogram.clone())) + } +} + +/// Builds the instrument behind each new series. +struct Instruments { + meter: Meter, + histogram_boundaries: Vec, + descriptions: Arc, +} + +impl Instruments { + /// The OpenTelemetry description and UCUM unit, empty when never described. + fn metadata(&self, key: &Key) -> (SharedString, &'static str) { + lock(&self.descriptions) + .get(key.name()) + .map_or_else(Default::default, |(unit, description)| { + (description.clone(), unit.map_or("", ucum)) + }) + } +} + +fn ucum(unit: Unit) -> &'static str { + match unit { + Unit::Count => "1", + Unit::Seconds => "s", + Unit::Bytes => "By", + other => other.as_canonical_label(), + } +} + +fn attributes(key: &Key) -> Vec { + key.labels() + .map(|label| KeyValue::new(label.key().to_owned(), label.value().to_owned())) + .collect() +} + +impl Storage for Instruments { + type Counter = Arc; + type Gauge = Arc; + type Histogram = Arc; + + fn counter(&self, key: &Key) -> Self::Counter { + let (description, unit) = self.metadata(key); + Arc::new(OtelCounter { + counter: self + .meter + .u64_counter(key.name_shared().into_inner()) + .with_description(description) + .with_unit(unit) + .build(), + attributes: attributes(key), + total: AtomicU64::new(0), + }) + } + + fn gauge(&self, key: &Key) -> Self::Gauge { + let (description, unit) = self.metadata(key); + Arc::new(OtelGauge { + gauge: self + .meter + .f64_gauge(key.name_shared().into_inner()) + .with_description(description) + .with_unit(unit) + .build(), + attributes: attributes(key), + value: Mutex::new(0.0), + }) + } + + fn histogram(&self, key: &Key) -> Self::Histogram { + let (description, unit) = self.metadata(key); + Arc::new(OtelHistogram { + histogram: self + .meter + .f64_histogram(key.name_shared().into_inner()) + .with_description(description) + .with_unit(unit) + .with_boundaries(self.histogram_boundaries.clone()) + .build(), + attributes: attributes(key), + }) + } +} + +struct OtelCounter { + counter: opentelemetry::metrics::Counter, + attributes: Vec, + total: AtomicU64, +} + +impl CounterFn for OtelCounter { + fn increment(&self, value: u64) { + self.total.fetch_add(value, Ordering::Relaxed); + self.counter.add(value, &self.attributes); + } + + fn absolute(&self, value: u64) { + let previous = self.total.fetch_max(value, Ordering::Relaxed); + self.counter + .add(value.saturating_sub(previous), &self.attributes); + } +} + +/// Keeps the value here because the metrics API adjusts gauges by a delta and +/// OpenTelemetry records the new value. The lock orders concurrent updates so +/// the last recorded value is the current one. +struct OtelGauge { + gauge: opentelemetry::metrics::Gauge, + attributes: Vec, + value: Mutex, +} + +impl OtelGauge { + fn update(&self, change: impl FnOnce(f64) -> f64) { + let mut value = lock(&self.value); + *value = change(*value); + self.gauge.record(*value, &self.attributes); + } +} + +impl GaugeFn for OtelGauge { + fn increment(&self, delta: f64) { + self.update(|value| value + delta); + } + + fn decrement(&self, delta: f64) { + self.update(|value| value - delta); + } + + fn set(&self, value: f64) { + self.update(|_| value); + } +} + +struct OtelHistogram { + histogram: opentelemetry::metrics::Histogram, + attributes: Vec, +} + +impl HistogramFn for OtelHistogram { + fn record(&self, value: f64) { + self.histogram.record(value, &self.attributes); + } +} + +#[cfg(test)] +mod tests { + use metrics::{counter, describe_counter, describe_histogram, gauge, histogram}; + use opentelemetry::metrics::MeterProvider as _; + use opentelemetry_sdk::metrics::data::{ + AggregatedMetrics, GaugeDataPoint, MetricData, ResourceMetrics, ScopeMetrics, SumDataPoint, + }; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader}; + + use super::*; + use crate::{OtlpExportConfig, ServiceName}; + + fn bridge() -> (MetricsBridge, SdkMeterProvider, InMemoryMetricExporter) { + let exporter = InMemoryMetricExporter::default(); + let provider = SdkMeterProvider::builder() + .with_reader(PeriodicReader::builder(exporter.clone()).build()) + .build(); + let bridge = MetricsBridge::new(provider.meter("test"), vec![0.1, 1.0]); + (bridge, provider, exporter) + } + + fn flushed(provider: &SdkMeterProvider, exporter: &InMemoryMetricExporter) -> ResourceMetrics { + provider.force_flush().expect("flush"); + exporter + .get_finished_metrics() + .expect("metrics") + .pop() + .expect("an export") + } + + fn metric<'a>( + metrics: &'a ResourceMetrics, + name: &str, + ) -> &'a opentelemetry_sdk::metrics::data::Metric { + metrics + .scope_metrics() + .flat_map(ScopeMetrics::metrics) + .find(|metric| metric.name() == name) + .unwrap_or_else(|| panic!("{name} was not exported")) + } + + #[test] + fn counters_sum_per_label_set_and_never_decrease() { + let (bridge, provider, exporter) = bridge(); + metrics::with_local_recorder(&bridge, || { + counter!("hits_total", "route" => "a").increment(2); + counter!("hits_total", "route" => "a").increment(3); + counter!("hits_total", "route" => "b").absolute(7); + counter!("hits_total", "route" => "b").absolute(4); + counter!("zero_total").increment(0); + }); + let data = flushed(&provider, &exporter); + + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric(&data, "hits_total").data() + else { + panic!("hits_total is not a u64 sum"); + }; + let mut points: Vec<_> = sum + .data_points() + .map(|point| { + let route = point.attributes().next().expect("route").value.to_string(); + (route, point.value()) + }) + .collect(); + points.sort(); + assert_eq!(points, [("a".to_string(), 5), ("b".to_string(), 7)]); + + let AggregatedMetrics::U64(MetricData::Sum(zero)) = metric(&data, "zero_total").data() + else { + panic!("zero_total is not a u64 sum"); + }; + assert_eq!( + zero.data_points() + .map(SumDataPoint::value) + .collect::>(), + [0] + ); + } + + #[test] + fn gauges_export_their_last_value_across_handles() { + let (bridge, provider, exporter) = bridge(); + metrics::with_local_recorder(&bridge, || { + let first = gauge!("sessions"); + first.increment(3.0); + gauge!("sessions").decrement(1.0); + first.increment(0.5); + gauge!("capacity").set(8.0); + gauge!("capacity").set(9.0); + }); + let data = flushed(&provider, &exporter); + + for (name, expected) in [("sessions", 2.5), ("capacity", 9.0)] { + let AggregatedMetrics::F64(MetricData::Gauge(gauge)) = metric(&data, name).data() + else { + panic!("{name} is not an f64 gauge"); + }; + let values: Vec<_> = gauge.data_points().map(GaugeDataPoint::value).collect(); + assert_eq!(values, [expected], "{name}"); + } + } + + #[test] + fn histograms_use_the_configured_boundaries() { + let (bridge, provider, exporter) = bridge(); + metrics::with_local_recorder(&bridge, || { + let latency = histogram!("latency_seconds", "op" => "x"); + latency.record(0.05); + latency.record(0.5); + latency.record(30.0); + }); + let data = flushed(&provider, &exporter); + + let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = + metric(&data, "latency_seconds").data() + else { + panic!("latency_seconds is not an f64 histogram"); + }; + let point = histogram.data_points().next().expect("a data point"); + assert_eq!(point.count(), 3); + assert_eq!(point.bounds().collect::>(), [0.1, 1.0]); + assert_eq!(point.bucket_counts().collect::>(), [1, 1, 1]); + } + + #[test] + fn descriptions_and_units_reach_the_instrument() { + let (bridge, provider, exporter) = bridge(); + metrics::with_local_recorder(&bridge, || { + describe_counter!("described_total", Unit::Count, "Things that happened."); + describe_histogram!("described_seconds", Unit::Seconds, "How long it took."); + counter!("described_total").increment(1); + histogram!("described_seconds").record(0.2); + counter!("undescribed_total").increment(1); + }); + let data = flushed(&provider, &exporter); + + let counter = metric(&data, "described_total"); + assert_eq!(counter.description(), "Things that happened."); + assert_eq!(counter.unit(), "1"); + let histogram = metric(&data, "described_seconds"); + assert_eq!(histogram.description(), "How long it took."); + assert_eq!(histogram.unit(), "s"); + assert_eq!(metric(&data, "undescribed_total").unit(), ""); + } + + #[tokio::test] + async fn target_builds_a_meter_provider_with_its_resource() { + let target = OtlpTarget::new(&OtlpExportConfig { + endpoint: "http://127.0.0.1:4317", + service_name: ServiceName::Fixed("openshell-gateway"), + service_version: None, + resource_attributes: Vec::new(), + }) + .unwrap(); + target.meter_provider().expect("meter provider"); + } + + #[tokio::test] + async fn target_builds_an_http_meter_provider_but_not_over_tls() { + let target = |endpoint| { + OtlpTarget::new(&OtlpExportConfig { + endpoint, + service_name: ServiceName::Fixed("openshell-gateway"), + service_version: None, + resource_attributes: Vec::new(), + }) + .unwrap() + .with_protocol(Protocol::HttpProtobuf) + }; + target("http://127.0.0.1:4318") + .meter_provider() + .expect("meter provider"); + assert!(matches!( + target("https://collector.example:4318").meter_provider(), + Err(SetupError::HttpsUnsupported) + )); + } +} diff --git a/crates/openshell-server/Cargo.toml b/crates/openshell-server/Cargo.toml index 9b9462cd67..89ffbe998f 100644 --- a/crates/openshell-server/Cargo.toml +++ b/crates/openshell-server/Cargo.toml @@ -82,6 +82,7 @@ tracing-opentelemetry = { workspace = true } # Metrics metrics = { workspace = true } metrics-exporter-prometheus = { workspace = true } +metrics-util = { workspace = true } # Utilities base64 = { workspace = true } diff --git a/crates/openshell-server/src/cli.rs b/crates/openshell-server/src/cli.rs index 6a672c5aa5..8ed67b3859 100644 --- a/crates/openshell-server/src/cli.rs +++ b/crates/openshell-server/src/cli.rs @@ -856,6 +856,8 @@ async fn run_from_args( .map(|_| tracing::field::display(otlp.traces_protocol())), ocsf_endpoint = otlp.ocsf_endpoint().map(tracing::field::display), ocsf_protocol = otlp.ocsf_protocol().map(tracing::field::display), + metrics_endpoint = otlp.metrics_endpoint().map(tracing::field::display), + metrics_protocol = otlp.metrics_protocol().map(tracing::field::display), agent_endpoint = otlp.agent_endpoint().map(tracing::field::display), agent_protocol = otlp.agent_protocol().map(tracing::field::display), "OTLP exporting enabled" @@ -869,7 +871,17 @@ async fn run_from_args( info!(bind = %prepared.config.bind_address, "Starting OpenShell server"); - let result = Box::pin(run_server(prepared, tracing_log_bus, compute_drivers)).await; + let meter_provider = otlp_export + .as_ref() + .and_then(crate::otel_tracing::OtlpExport::meter_provider) + .cloned(); + let result = Box::pin(run_server( + prepared, + tracing_log_bus, + compute_drivers, + meter_provider, + )) + .await; tracing_handle.shutdown(); diff --git a/crates/openshell-server/src/config_file.rs b/crates/openshell-server/src/config_file.rs index 2c74c19360..ef03084cd8 100644 --- a/crates/openshell-server/src/config_file.rs +++ b/crates/openshell-server/src/config_file.rs @@ -322,11 +322,12 @@ impl TryFrom for OcsfLogConfig { /// /// `endpoint` is the collector for every enabled signal, and on its own /// enables trace export. `[otlp.traces]` overrides the trace endpoint, -/// `[otlp.ocsf]` with `enabled = true` exports OCSF events as log records, and -/// `[otlp.agent]` with `enabled = true` relays workload traces, each -/// optionally to a different endpoint. SDK tuning knobs are deliberately absent — see -/// [`crate::otel_tracing`] for what this table owns and what the `OTEL_*` -/// environment variables own. +/// `[otlp.ocsf]` with `enabled = true` exports OCSF events as log records, +/// `[otlp.agent]` with `enabled = true` relays workload traces, and +/// `[otlp.metrics]` with `enabled = true`, or `metrics = true`, pushes gateway +/// metrics, each optionally to a different endpoint. SDK tuning knobs are +/// deliberately absent. See [`crate::otel_tracing`] for what this table owns +/// and what the `OTEL_*` environment variables own. /// /// Endpoints are resolved when parsing, so each enabled signal carries its own. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -348,6 +349,14 @@ pub struct OtlpConfig { )] pub ocsf: Option, + /// Gateway metrics pushed as OTLP metrics, when enabled. + #[serde( + default, + skip_serializing_if = "Option::is_none", + serialize_with = "serialize_enabled" + )] + pub metrics: Option, + /// Relay of workload traces from sandboxes, when enabled. #[serde( default, @@ -460,6 +469,14 @@ pub struct OtlpTracesConfig { pub collector: OtlpCollectorConfig, } +/// `[openshell.gateway.otlp.metrics]`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OtlpMetricsConfig { + /// The collector that gateway metrics are pushed to. + #[serde(flatten)] + pub collector: OtlpCollectorConfig, +} + /// `[openshell.gateway.otlp.ocsf]`. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct OcsfOtlpConfig { @@ -493,9 +510,30 @@ struct RawOtlpConfig { #[serde(default)] ocsf: Option, #[serde(default)] + metrics: Option, + #[serde(default)] agent: Option, } +/// `metrics = true`, or a table with an optional endpoint. +#[derive(Deserialize)] +#[serde(untagged)] +enum RawOtlpMetrics { + Enabled(bool), + Table(RawOtlpMetricsTable), +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct RawOtlpMetricsTable { + #[serde(default)] + enabled: bool, + #[serde(default)] + endpoint: Option, + #[serde(default)] + protocol: Option, +} + #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct RawAgentOtlp { @@ -583,6 +621,23 @@ impl TryFrom for OtlpConfig { }) }) .transpose()?; + let metrics = raw + .metrics + .map(|metrics| match metrics { + RawOtlpMetrics::Enabled(enabled) => RawOtlpMetricsTable { + enabled, + endpoint: None, + protocol: None, + }, + RawOtlpMetrics::Table(table) => table, + }) + .filter(|metrics| metrics.enabled) + .map(|metrics| { + Ok::<_, String>(OtlpMetricsConfig { + collector: collector_for("metrics", metrics.endpoint, metrics.protocol)?, + }) + }) + .transpose()?; let agent = raw .agent .filter(|agent| agent.enabled) @@ -592,9 +647,9 @@ impl TryFrom for OtlpConfig { }) }) .transpose()?; - if traces.is_none() && ocsf.is_none() && agent.is_none() { + if traces.is_none() && ocsf.is_none() && metrics.is_none() && agent.is_none() { return Err( - "otlp needs an endpoint, or otlp.ocsf or otlp.agent with enabled = true" + "otlp needs an endpoint, or otlp.ocsf, otlp.metrics or otlp.agent with enabled = true" .to_string(), ); } @@ -602,6 +657,7 @@ impl TryFrom for OtlpConfig { service_name: raw.service_name, traces, ocsf, + metrics, agent, }) } @@ -1649,6 +1705,75 @@ service_name = "openshell-gateway-dev" ); } + #[test] + fn otlp_metrics_default_to_the_shared_endpoint() { + for metrics in [ + "metrics = true\n", + "\n[openshell.gateway.otlp.metrics]\nenabled = true\n", + ] { + let otlp = otlp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://127.0.0.1:4317'\n{metrics}" + )); + assert_eq!(otlp.traces_endpoint(), Some("http://127.0.0.1:4317")); + assert_eq!( + otlp.metrics, + Some(OtlpMetricsConfig { + collector: OtlpCollectorConfig { + endpoint: "http://127.0.0.1:4317".into(), + protocol: OtlpProtocol::Grpc, + }, + }), + "{metrics:?}" + ); + } + } + + #[test] + fn otlp_metrics_can_use_their_own_endpoint_without_traces() { + let otlp = otlp( + "[openshell.gateway.otlp.metrics]\nenabled = true\nendpoint = 'http://metrics:4317'\n", + ); + assert_eq!(otlp.traces_endpoint(), None); + assert_eq!( + otlp.metrics + .map(|metrics| metrics.collector.endpoint) + .as_deref(), + Some("http://metrics:4317") + ); + } + + #[test] + fn otlp_metrics_can_use_http_protobuf() { + let otlp = otlp( + "[openshell.gateway.otlp.metrics]\nenabled = true\nendpoint = 'http://metrics:4318'\n\ + protocol = 'http/protobuf'\n", + ); + assert_eq!( + otlp.metrics.map(|metrics| metrics.collector), + Some(OtlpCollectorConfig { + endpoint: "http://metrics:4318".into(), + protocol: OtlpProtocol::HttpProtobuf, + }) + ); + } + + #[test] + fn otlp_metrics_are_off_unless_enabled() { + for metrics in [ + "", + "metrics = false\n", + "\n[openshell.gateway.otlp.metrics]\nendpoint = 'http://metrics:4317'\n", + ] { + let otlp = otlp(&format!( + "[openshell.gateway.otlp]\nendpoint = 'http://127.0.0.1:4317'\n{metrics}" + )); + assert_eq!( + otlp.metrics, None, + "enabled metrics export with {metrics:?}" + ); + } + } + #[test] fn otlp_ocsf_is_off_unless_enabled() { for ocsf in ["", "enabled = false\n", "endpoint = 'http://siem:4317'\n"] { @@ -1813,8 +1938,10 @@ service_name = "openshell-gateway-dev" let tmp = write_tmp( "[openshell.gateway.otlp]\nendpoint = 'http://shared:4317'\nservice_name = 'gw'\n\n\ [openshell.gateway.otlp.ocsf]\nenabled = true\n\n\ - [openshell.gateway.otlp.agent]\nenabled = true\nendpoint = 'http://agents:4318'\n\ - protocol = 'http/protobuf'\n", + [openshell.gateway.otlp.metrics]\nenabled = true\nendpoint = 'http://metrics:4318'\n\ + protocol = 'http/protobuf'\n\n\ + [openshell.gateway.otlp.agent]\nenabled = true\nendpoint = 'http://agents:4318'\n\ + protocol = 'http/protobuf'\n", ); let file = load(tmp.path()).unwrap(); let encoded = toml::to_string(&file).unwrap(); @@ -1831,6 +1958,9 @@ service_name = "openshell-gateway-dev" "[openshell.gateway.otlp.ocsf]\nenabled = true\n", "[openshell.gateway.otlp.ocsf]\nenabled = 'yes'\nendpoint = 'http://siem:4317'\n", "[openshell.gateway.otlp.metrics]\nendpoint = 'http://metrics:4317'\n", + "[openshell.gateway.otlp.metrics]\nenabled = true\n", + "[openshell.gateway.otlp]\nmetrics = true\n", + "[openshell.gateway.otlp]\nendpoint = 'http://127.0.0.1:4317'\nmetrics = 'yes'\n", "[openshell.gateway.otlp]\nendpoint = 'http://127.0.0.1:4317'\nocsf_logs = true\n", ] { let tmp = write_tmp(table); diff --git a/crates/openshell-server/src/gateway_metrics.rs b/crates/openshell-server/src/gateway_metrics.rs index 0f55f42bbd..02a92a2055 100644 --- a/crates/openshell-server/src/gateway_metrics.rs +++ b/crates/openshell-server/src/gateway_metrics.rs @@ -16,9 +16,15 @@ use std::time::{Duration, Instant}; use metrics::{ - Gauge, Unit, counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram, + Gauge, Recorder, Unit, counter, describe_counter, describe_gauge, describe_histogram, gauge, + histogram, }; -use metrics_exporter_prometheus::{BuildError, Matcher, PrometheusBuilder, PrometheusHandle}; +use metrics_exporter_prometheus::{ + BuildError, Matcher, PrometheusBuilder, PrometheusHandle, PrometheusRecorder, +}; +use metrics_util::layers::FanoutBuilder; +use openshell_otel::{MetricsBridge, SdkMeterProvider}; +use opentelemetry::metrics::MeterProvider as _; use tonic::{Code, Status}; // Gauges @@ -241,18 +247,52 @@ pub fn configure_exporter(builder: PrometheusBuilder) -> Result MetricsBridge { + MetricsBridge::new( + provider.meter(crate::otel_tracing::INSTRUMENTATION_SCOPE), + LATENCY_BUCKETS_SECONDS.to_vec(), + ) +} + /// Install the process-wide recorder, then describe and zero-initialize the catalog. Call -/// once, from `run_server`. `mutation_lock_connections` is the `PostgreSQL` lock pool size, -/// `None` on a backend without a lock pool. +/// once, from `run_server`. Metrics go to the returned Prometheus handle when `prometheus` is +/// set, and to `otlp` when it is given. With neither, nothing is installed. +/// `mutation_lock_connections` is the `PostgreSQL` lock pool size, `None` on a backend without a +/// lock pool. pub fn install_global_recorder( relay: RelayCapacity, mutation_lock_connections: Option, -) -> Result { - let handle = configure_exporter(PrometheusBuilder::new())?.install_recorder()?; + prometheus: bool, + otlp: Option<&SdkMeterProvider>, +) -> Result, String> { + let prometheus = prometheus + .then(|| { + configure_exporter(PrometheusBuilder::new()).map(PrometheusBuilder::build_recorder) + }) + .transpose() + .map_err(|error: BuildError| error.to_string())?; + let handle = prometheus.as_ref().map(PrometheusRecorder::handle); + match (prometheus, otlp.map(otlp_recorder)) { + (None, None) => return Ok(None), + (Some(prometheus), None) => set_global_recorder(prometheus)?, + (None, Some(otlp)) => set_global_recorder(otlp)?, + (Some(prometheus), Some(otlp)) => set_global_recorder( + FanoutBuilder::default() + .add_recorder(prometheus) + .add_recorder(otlp) + .build(), + )?, + } describe_and_initialize(relay, mutation_lock_connections); Ok(handle) } +fn set_global_recorder(recorder: impl Recorder + Sync + 'static) -> Result<(), String> { + metrics::set_global_recorder(recorder).map_err(|error| error.to_string()) +} + /// Emit HELP metadata, create every fixed-label series at 0, and publish the relay caps and, /// with a lock pool, its size. An idle replica then exports 0 instead of "no data", which HPA /// and `rate()` need. @@ -587,12 +627,11 @@ pub struct MetricsCapture { impl MetricsCapture { pub fn install() -> Self { // Leaked (test only, one small allocation per test) so the guard can borrow it for 'static. - let recorder: &'static metrics_exporter_prometheus::PrometheusRecorder = - Box::leak(Box::new( - configure_exporter(PrometheusBuilder::new()) - .expect("valid exporter config") - .build_recorder(), - )); + let recorder: &'static PrometheusRecorder = Box::leak(Box::new( + configure_exporter(PrometheusBuilder::new()) + .expect("valid exporter config") + .build_recorder(), + )); let handle = recorder.handle(); let guard = metrics::set_default_local_recorder(recorder); Self { diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 12ca9223ea..82d258e287 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -597,6 +597,7 @@ pub(crate) async fn run_server( startup: ServerStartupConfig, tracing_log_bus: TracingLogBus, compute_drivers: ComputeDriverRegistry, + meter_provider: Option, ) -> Result<()> { let ServerStartupConfig { config, @@ -946,20 +947,16 @@ pub(crate) async fn run_server( let state = Arc::new(state); - // Install the recorder before the endpoint-status startup reconciliation so + // Install the recorders before the endpoint-status startup reconciliation so // the lock waits and timeouts recorded there are exported. Its listener // binds with the others. - let metrics_endpoint = config - .metrics_bind_address - .map(|metrics_bind_address| { - gateway_metrics::install_global_recorder( - supervisor_session::RELAY_CAPACITY, - store.mutation_lock_connections(), - ) - .map(|prometheus_handle| (metrics_bind_address, prometheus_handle)) - }) - .transpose() - .map_err(|e| Error::config(format!("failed to install metrics recorder: {e}")))?; + let prometheus_handle = gateway_metrics::install_global_recorder( + supervisor_session::RELAY_CAPACITY, + store.mutation_lock_connections(), + config.metrics_bind_address.is_some(), + meter_provider.as_ref(), + ) + .map_err(|e| Error::config(format!("failed to install metrics recorder: {e}")))?; grpc::policy::backfill_legacy_policy_history(&state) .await @@ -1018,7 +1015,9 @@ pub(crate) async fn run_server( } // Bind the Prometheus metrics endpoint on a dedicated port when configured. - if let Some((metrics_bind_address, prometheus_handle)) = metrics_endpoint { + if let (Some(metrics_bind_address), Some(prometheus_handle)) = + (config.metrics_bind_address, prometheus_handle) + { let metrics_listener = TcpListener::bind(metrics_bind_address).await.map_err(|e| { Error::transport(format!( "failed to bind metrics port {metrics_bind_address}: {e}", diff --git a/crates/openshell-server/src/otel_tracing.rs b/crates/openshell-server/src/otel_tracing.rs index 426c283f83..637045b7a1 100644 --- a/crates/openshell-server/src/otel_tracing.rs +++ b/crates/openshell-server/src/otel_tracing.rs @@ -3,29 +3,32 @@ //! OpenTelemetry export for the gateway. //! -//! Converts selected Rust `tracing` spans into OpenTelemetry traces and -//! gateway OCSF events into log records, and exports each over OTLP, gRPC or -//! HTTP/protobuf per collector, when configured. +//! Converts selected Rust `tracing` spans into OpenTelemetry traces, gateway +//! OCSF events into log records, and gateway metrics into OTLP metrics, and +//! exports each over OTLP, gRPC or HTTP/protobuf per collector, when +//! configured. //! //! # Configuration split //! //! `[openshell.gateway.otlp]` decides **whether and where** to export each //! signal: a shared `endpoint` enables traces, `[otlp.traces]` overrides //! their endpoint, and `[otlp.ocsf]` with `enabled = true` exports OCSF -//! records, optionally to their own endpoint. `OTEL_EXPORTER_OTLP_*ENDPOINT` is deliberately not -//! read, so enablement has one source. +//! records, optionally to their own endpoint. `[otlp.metrics]` with `enabled = true` +//! pushes gateway metrics, optionally to their own endpoint. +//! `OTEL_EXPORTER_OTLP_*ENDPOINT` is deliberately not read, so enablement has one source. //! //! **How** to export — sampling, batching, span limits, transport headers — //! is the SDK's `OTEL_*` environment surface, read as the provider is built //! and mirrored nowhere here. `docs/how-it-works/gateways/configuration.mdx` documents //! the variables operators are likely to want. //! -//! Diagnostic logs and metrics are not exported over OTLP. They have their own -//! surfaces (the gateway log stream and the Prometheus `/metrics` endpoint). +//! Diagnostic logs are not exported over OTLP. They have their own surface, the +//! gateway log stream. Metrics are always available on the Prometheus `/metrics` +//! endpoint, and are pushed over OTLP in addition when `[otlp.metrics]` is enabled. use openshell_core::jwt::SecretOtlpHeaders; use openshell_core::sandbox_env::OtlpProtocol; -use openshell_otel::{OcsfOtlp, OtlpExportConfig, OtlpTarget, ServiceName}; +use openshell_otel::{OcsfOtlp, OtlpExportConfig, OtlpTarget, SdkMeterProvider, ServiceName}; pub use openshell_otel::{SetupError, TraceContextInterceptor, mark_error}; #[cfg(test)] use opentelemetry_sdk::Resource; @@ -38,8 +41,8 @@ use crate::config_file::{OtlpCollectorConfig, OtlpConfig}; /// `service.name` reported when the config file does not override it. const DEFAULT_SERVICE_NAME: &str = "openshell-gateway"; -/// Instrumentation scope recorded on spans this gateway emits. -const INSTRUMENTATION_SCOPE: &str = "openshell-gateway"; +/// Instrumentation scope recorded on spans and metrics this gateway emits. +pub const INSTRUMENTATION_SCOPE: &str = "openshell-gateway"; /// Gateway identity recorded on every exported span. #[derive(Debug, Clone, Copy, Default)] @@ -114,11 +117,12 @@ fn build_provider( openshell_otel::build_provider(&export_config(cfg, endpoint, gateway)) } -/// OTLP export configured by `[openshell.gateway.otlp]`: gateway spans and -/// OCSF log records, each sent to its own configured collector. +/// OTLP export configured by `[openshell.gateway.otlp]`: gateway spans, OCSF +/// log records, and metrics, each sent to its own configured collector. pub struct OtlpExport { traces: Option>, ocsf: Option>, + metrics: Option>, /// Validated collector that supervisors relay workload traces to. The /// gateway itself exports nothing for this signal. agent: Option>, @@ -182,6 +186,18 @@ impl OtlpExport { }) }) .transpose()?; + let metrics = cfg + .metrics + .as_ref() + .map(|metrics| { + let target = target(&metrics.collector)?; + Ok::<_, SetupError>(Signal { + endpoint: target.endpoint().to_string(), + protocol: metrics.collector.protocol, + exporter: target.meter_provider()?, + }) + }) + .transpose()?; let agent = cfg .agent .as_ref() @@ -196,6 +212,7 @@ impl OtlpExport { Ok(Self { traces, ocsf, + metrics, agent, }) } @@ -244,14 +261,36 @@ impl OtlpExport { self.ocsf.as_ref().map(|ocsf| &ocsf.exporter) } - /// Flush buffered spans, then export queued OCSF records within their - /// shutdown budget. + /// The collector receiving metrics, when metrics export is enabled. + pub fn metrics_endpoint(&self) -> Option<&str> { + self.metrics + .as_ref() + .map(|metrics| metrics.endpoint.as_str()) + } + + /// The transport to the metrics collector, when metrics export is enabled. + pub fn metrics_protocol(&self) -> Option { + self.metrics.as_ref().map(|metrics| metrics.protocol) + } + + /// The meter provider, when metrics export is enabled. + pub fn meter_provider(&self) -> Option<&SdkMeterProvider> { + self.metrics.as_ref().map(|metrics| &metrics.exporter) + } + + /// Flush buffered spans and metrics, then export queued OCSF records + /// within their shutdown budget. pub async fn shutdown(self) { if let Some(traces) = self.traces && let Err(err) = traces.exporter.shutdown() { tracing::warn!(error = %err, "OTLP tracer provider shutdown failed"); } + if let Some(metrics) = self.metrics + && let Err(err) = metrics.exporter.shutdown() + { + tracing::warn!(error = %err, "OTLP meter provider shutdown failed"); + } if let Some(ocsf) = self.ocsf { ocsf.exporter.shutdown().await; } @@ -605,7 +644,7 @@ pub mod test_exporter { #[cfg(test)] mod tests { use super::*; - use crate::config_file::{OcsfOtlpConfig, OtlpTracesConfig}; + use crate::config_file::{OcsfOtlpConfig, OtlpMetricsConfig, OtlpTracesConfig}; use crate::test_environment::Environment; fn config() -> OtlpConfig { @@ -622,10 +661,20 @@ mod tests { }, }), ocsf: None, + metrics: None, agent: None, } } + fn metrics_config(endpoint: &str) -> OtlpMetricsConfig { + OtlpMetricsConfig { + collector: OtlpCollectorConfig { + endpoint: endpoint.into(), + protocol: OtlpProtocol::Grpc, + }, + } + } + fn ocsf_config(endpoint: &str) -> OcsfOtlpConfig { OcsfOtlpConfig { collector: OtlpCollectorConfig { @@ -703,9 +752,47 @@ mod tests { let export = export.expect("OCSF log export"); assert!(export.tracer_provider().is_none()); assert!(export.ocsf().is_some()); + assert!(export.meter_provider().is_none()); + export.shutdown().await; + + cfg.ocsf = None; + cfg.metrics = Some(metrics_config("http://metrics:4317")); + let (export, error) = OtlpExport::start(Some(&cfg), GatewayResourceAttributes::default()); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + let export = export.expect("metrics export"); + assert_eq!(export.metrics_endpoint(), Some("http://metrics:4317")); + assert!(export.meter_provider().is_some()); + assert!(export.ocsf().is_none()); export.shutdown().await; } + #[tokio::test(flavor = "multi_thread")] + async fn metrics_reach_an_otlp_collector_over_grpc() { + let collector = openshell_otel_test_support::OtlpTestServer::start().await; + let mut cfg = config(); + cfg.traces = None; + cfg.metrics = Some(metrics_config(collector.endpoint())); + let (export, error) = OtlpExport::start( + Some(&cfg), + GatewayResourceAttributes::new(Some("test-gateway"), Some("docker")), + ); + assert!(error.is_none(), "unexpected setup error: {error:?}"); + let export = export.expect("OTLP export"); + let recorder = + crate::gateway_metrics::otlp_recorder(export.meter_provider().expect("metrics export")); + metrics::with_local_recorder(&recorder, || { + crate::gateway_metrics::record_relay_expired(2); + }); + export.shutdown().await; + let received = collector.shutdown().await; + + assert_eq!(received.gateway_names, ["test-gateway"]); + assert_eq!( + received.metric_names, + [crate::gateway_metrics::RELAY_EXPIRED_TOTAL] + ); + } + #[tokio::test] async fn ocsf_records_reach_an_otlp_collector_over_grpc() { use openshell_ocsf::{ConfigStateChangeBuilder, SeverityId, ocsf_emit}; @@ -983,6 +1070,8 @@ mod tests { async fn misconfigured_endpoint_disables_export_without_failing_startup() { let mut bad_ocsf = config(); bad_ocsf.ocsf = Some(ocsf_config("definitely not a url")); + let mut bad_metrics = config(); + bad_metrics.metrics = Some(metrics_config("definitely not a url")); let mut bad_agent = config(); bad_agent.agent = Some(crate::config_file::AgentOtlpConfig { collector: OtlpCollectorConfig { @@ -993,6 +1082,7 @@ mod tests { for cfg in [ with_traces_endpoint("definitely not a url"), bad_ocsf, + bad_metrics, bad_agent, ] { let (export, err) = OtlpExport::start(Some(&cfg), GatewayResourceAttributes::default()); diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 20446834b9..1962289981 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -600,7 +600,7 @@ Allow enough retention and disk space for shipper outages. Graceful shutdown all ## OTLP Export -`[openshell.gateway.otlp]` enables OpenTelemetry export over OTLP. Omit the table to disable export. Its `endpoint` is the collector for every signal the gateway exports and, on its own, enables trace export. `[openshell.gateway.otlp.traces]` sends spans to a different collector. [OCSF log export](#ocsf-log-export) is off until `[openshell.gateway.otlp.ocsf]` sets `enabled = true`, and the [agent trace relay](#agent-trace-relay) is off until `[openshell.gateway.otlp.agent]` sets `enabled = true`. Each can use its own collector. +`[openshell.gateway.otlp]` enables OpenTelemetry export over OTLP. Omit the table to disable export. Its `endpoint` is the collector for every signal the gateway exports and, on its own, enables trace export. `[openshell.gateway.otlp.traces]` sends spans to a different collector. [OCSF log export](#ocsf-log-export) is off until `[openshell.gateway.otlp.ocsf]` sets `enabled = true`, [metrics push](/observability/gateway-metrics#push-metrics-over-otlp) is off until `[openshell.gateway.otlp.metrics]` sets `enabled = true`, and the [agent trace relay](#agent-trace-relay) is off until `[openshell.gateway.otlp.agent]` sets `enabled = true`. Each can use its own collector. The gateway already uses the Rust `tracing` framework for structured logs sent to stdout and the sandbox log stream. Enabling this section adds an OpenTelemetry layer to the same tracing subscriber. It exports span trees to an OTLP collector without exporting, replacing, or redirecting the existing diagnostic log events. @@ -617,7 +617,7 @@ protocol = "http/protobuf" Each enabled signal needs an endpoint, either in its own table or from `endpoint`. A table without its own `endpoint` uses the whole `[openshell.gateway.otlp]` collector, including its `protocol`. A table with its own `endpoint` takes nothing else from `[openshell.gateway.otlp]`, so settings for one collector never reach another, and setting `protocol` in a table without an `endpoint` is a configuration error. Endpoints must be valid URIs without a query string: exporters send no query parameters, so an endpoint with one is a configuration error, and a collector that needs parameters must take them as headers. If an endpoint is malformed, the gateway logs the configuration error and continues with all OTLP export disabled. It does not connect at startup: an unreachable collector produces export failures, never a failure to serve. -`protocol` selects the transport to a collector: `grpc`, the default, or `http/protobuf`. With `http/protobuf`, the endpoint is a base URL and each signal is posted to its path under it, `/v1/traces` or `/v1/logs`, as with `OTEL_EXPORTER_OTLP_ENDPOINT`. HTTP/JSON is not supported. Point a `grpc` endpoint at a collector's gRPC receiver, conventionally port `4317`, and an `http/protobuf` endpoint at its HTTP receiver, conventionally port `4318`. A URI alone cannot distinguish the two, so a mismatched endpoint is accepted at startup and then fails on export. Sandbox supervisors and standalone compute drivers must come from the same release as the gateway to use `http/protobuf`: an older supervisor exports over gRPC, and an older driver rejects the protocol setting. +`protocol` selects the transport to a collector: `grpc`, the default, or `http/protobuf`. With `http/protobuf`, the endpoint is a base URL and each signal is posted to its path under it, `/v1/traces`, `/v1/logs`, or `/v1/metrics`, as with `OTEL_EXPORTER_OTLP_ENDPOINT`. HTTP/JSON is not supported. Point a `grpc` endpoint at a collector's gRPC receiver, conventionally port `4317`, and an `http/protobuf` endpoint at its HTTP receiver, conventionally port `4318`. A URI alone cannot distinguish the two, so a mismatched endpoint is accepted at startup and then fails on export. Sandbox supervisors and standalone compute drivers must come from the same release as the gateway to use `http/protobuf`: an older supervisor exports over gRPC, and an older driver rejects the protocol setting. Use an `https://` endpoint for TLS over `grpc`. The gateway verifies the collector certificate against the system trust roots. An `https://` endpoint with `http/protobuf` is a configuration error. @@ -728,6 +728,8 @@ The gateway honors these variables: | `OTEL_EXPORTER_OTLP_COMPRESSION` | Set to `gzip` to compress exports of every signal. `zstd` is not supported. | | `OTEL_EXPORTER_OTLP_TRACES_HEADERS`, `OTEL_EXPORTER_OTLP_TRACES_COMPRESSION`, `OTEL_EXPORTER_OTLP_TRACES_TIMEOUT` | Exporter transport tuning for spans only. These override the general variables. | | `OTEL_EXPORTER_OTLP_LOGS_HEADERS`, `OTEL_EXPORTER_OTLP_LOGS_COMPRESSION`, `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT` | Exporter transport tuning for OCSF log records only. These override the general variables. | +| `OTEL_EXPORTER_OTLP_METRICS_HEADERS`, `OTEL_EXPORTER_OTLP_METRICS_COMPRESSION`, `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` | Exporter transport tuning for metrics only. These override the general variables. | +| `OTEL_METRIC_EXPORT_INTERVAL`, `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` | How often the gateway pushes metrics, in milliseconds, which defaults to 60000, and whether it sends `cumulative` (the default) or `delta` values. | | `SSL_CERT_FILE`, `SSL_CERT_DIR` | Additional trust roots for `https://` endpoints, such as a private CA. | For Helm deployments, set these variables with `server.extraEnv`. Read credentials from a Secret so they stay out of the values file: @@ -749,13 +751,12 @@ These variables have no effect. Only `[openshell.gateway.otlp]` decides which si | Variable | Reason | |---|---| -| `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` | Enablement has one source, `[openshell.gateway.otlp]`, so an environment variable cannot silently turn export on or redirect it. | +| `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`, `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | Enablement has one source, `[openshell.gateway.otlp]`, so an environment variable cannot silently turn export on or redirect it. | | `OTEL_SDK_DISABLED`, `OTEL_TRACES_EXPORTER`, `OTEL_LOGS_EXPORTER`, `OTEL_METRICS_EXPORTER` | `[openshell.gateway.otlp]` decides which signals are exported. To disable export, omit it. | -| `OTEL_EXPORTER_OTLP_PROTOCOL`, `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`, `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL` | `protocol` in each table selects the transport, and supervisors follow the same setting. | +| `OTEL_EXPORTER_OTLP_PROTOCOL`, `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`, `OTEL_EXPORTER_OTLP_LOGS_PROTOCOL`, `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL` | `protocol` in each table selects the transport, and supervisors follow the same setting. | | `OTEL_EXPORTER_OTLP_INSECURE`, `OTEL_EXPORTER_OTLP_CERTIFICATE`, `OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE`, `OTEL_EXPORTER_OTLP_CLIENT_KEY`, and their per-signal forms | The endpoint scheme selects TLS, and collector certificates are verified against the system trust roots. Client certificates are not supported. | | `OTEL_PROPAGATORS` | The gateway always propagates W3C Trace Context. | | `OTEL_BLRP_SCHEDULE_DELAY`, `OTEL_BLRP_MAX_QUEUE_SIZE`, `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE`, `OTEL_BLRP_EXPORT_TIMEOUT` | OCSF log export uses its own queue and batching, described in [OCSF Log Export](#ocsf-log-export). | -| `OTEL_EXPORTER_OTLP_METRICS_*`, `OTEL_METRIC_EXPORT_INTERVAL` | The gateway serves metrics for Prometheus scraping and does not export them over OTLP. | To sample 10% of traces: diff --git a/docs/observability/gateway-metrics.mdx b/docs/observability/gateway-metrics.mdx index 24ec1b0bd9..24dd4d61e4 100644 --- a/docs/observability/gateway-metrics.mdx +++ b/docs/observability/gateway-metrics.mdx @@ -16,7 +16,8 @@ to alert on saturation, to follow rollouts, and to choose autoscaling signals. The gateway serves `GET /metrics` in the Prometheus text format on its own listener, separate from the API and health ports. When the listener is -disabled, the gateway records no metrics. Enable it in one of these ways: +disabled and OTLP push is off, the gateway records no metrics. Enable it in one +of these ways: - In `gateway.toml`, set `metrics_bind_address` under `[openshell.gateway]`, for example `metrics_bind_address = "0.0.0.0:9090"`. Refer to @@ -137,6 +138,26 @@ Then read the metrics from another terminal: curl -s http://localhost:9090/metrics | grep '^openshell_server_supervisor_sessions' ``` +## Push Metrics over OTLP + +The gateway can also push the same metrics to an OTLP collector, over gRPC or +HTTP/protobuf. It is +off by default and does not replace the scrape endpoint. Enable it in +`gateway.toml`: + +```toml +[openshell.gateway.otlp] +endpoint = "http://otel-collector.observability.svc:4317" +metrics = true +``` + +`metrics = true` sends metrics to the `endpoint`. To send them to another +collector, use a `[openshell.gateway.otlp.metrics]` table with `enabled = true` +and its own `endpoint`, and `protocol = "http/protobuf"` to use OTLP/HTTP. +Counters, gauges, and histograms keep their names and labels, and histograms use the latency buckets in seconds. The gateway pushes +every 60 seconds. For the interval and other export tuning, refer to +[Tuning](/how-it-works/gateways/configuration#tuning). + ## Metric Names Every metric name starts with `openshell_`, followed by the subsystem that diff --git a/docs/observability/overview.mdx b/docs/observability/overview.mdx index cb763427e1..54970d789f 100644 --- a/docs/observability/overview.mdx +++ b/docs/observability/overview.mdx @@ -13,7 +13,7 @@ OpenShell produces three kinds of observability signals: traces, metrics, and OC | Signal | Produced by | Destination | |---|---|---| | Traces | Gateway, compute drivers, sandbox supervisors | OTLP collector | -| Metrics | Gateway | Prometheus scrape | +| Metrics | Gateway | Prometheus scrape, optionally an OTLP collector | | OCSF events, shorthand | Gateway, sandbox supervisors | Gateway console, `openshell logs`, supervisor log file | | OCSF events, JSONL | Gateway, sandbox supervisors | Local file | | OCSF events, OTLP | Gateway, sandbox supervisors | OTLP collector |