diff --git a/contrib/openobserve/compose.yml b/contrib/openobserve/compose.yml new file mode 100644 index 00000000..7a6c3de5 --- /dev/null +++ b/contrib/openobserve/compose.yml @@ -0,0 +1,46 @@ +# Optional reference telemetry backend. Preloop never starts this — it is +# opt-in, loopback-bound, and pinned by digest. Credentials come from the +# environment (systemd LoadCredential / .env outside version control), +# never from this file. +# +# Upstream: https://github.com/openobserve/openobserve (AGPL-3.0). +# Run the stock image as a separate process; do not vendor or modify it. +services: + openobserve: + # Digest-pinned: a floating tag is not an immutable input. + image: public.ecr.aws/zinclabs/openobserve@sha256:88fb692ac791d3eaff69653a4a4686f1c7eceb9e105491d58d29ac2739560b3b + container_name: preloop-openobserve + restart: unless-stopped + # Loopback only. The OSS build has no SSO/RBAC — never put the UI on the + # public webhook origin. Front it with operator auth if shared. + ports: + - "127.0.0.1:5080:5080" + environment: + ZO_ROOT_USER_EMAIL: ${ZO_ROOT_USER_EMAIL:-admin@preloop.local} + # No default password: a known admin credential on a loopback port is + # one forwarded-port or one other-local-user away from being public. + # Startup fails loudly until the operator supplies one. + ZO_ROOT_USER_PASSWORD: ${ZO_ROOT_USER_PASSWORD:?Set ZO_ROOT_USER_PASSWORD} + ZO_DATA_DIR: /data + # Short retention: this is operational telemetry (hours/days), not a + # data lake. Losing SQLite metadata makes the install inoperable, so + # back up the volume if you rely on it. + ZO_COMPACT_DATA_RETENTION_DAYS: "7" + ZO_TELEMETRY: "false" + volumes: + - openobserve-data:/data + # Measured caps: do not starve the VM pool. Re-measure on your host. + deploy: + resources: + limits: + cpus: "1.0" + memory: 2G + healthcheck: + test: ["CMD", "sh", "-c", "wget -qO- http://127.0.0.1:5080/healthz || exit 1"] + interval: 10s + timeout: 3s + retries: 10 + start_period: 20s + +volumes: + openobserve-data: diff --git a/crates/preloop-cli/src/main.rs b/crates/preloop-cli/src/main.rs index 42339662..70c67bd3 100644 --- a/crates/preloop-cli/src/main.rs +++ b/crates/preloop-cli/src/main.rs @@ -749,7 +749,8 @@ async fn main() -> anyhow::Result<()> { // `RUST_LOG` defaults to `info` (the old `fmt::init()` default of ERROR hid // pool provisioning faults). The runtime is held for the life of `main` // and flushed with a bounded 2s shutdown on exit. - let obs_config = preloop_observability::ObservabilityConfig::from_env(); + let obs_config = preloop_observability::ObservabilityConfig::from_env() + .with_service_version(env!("CARGO_PKG_VERSION")); let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); preloop_observability::ObservabilityRuntime::install_fmt_subscriber(observability.config()); diff --git a/crates/preloop-observability/src/export.rs b/crates/preloop-observability/src/export.rs new file mode 100644 index 00000000..8c6149f8 --- /dev/null +++ b/crates/preloop-observability/src/export.rs @@ -0,0 +1,1032 @@ +//! Bounded OTLP/HTTP exporter using the existing reqwest+rustls stack. +//! +//! OTLP JSON encoding (`application/json`) per the OTLP/HTTP spec, so no +//! protobuf or tonic dependency. Invariants: +//! - Fail open: an export error is logged once per failure class and never +//! propagates to a caller. +//! - No request-path export: callers push into a bounded channel; a single +//! background worker drains it. Overflow drops and counts. +//! - No backend by default: constructed only when an endpoint is configured. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use serde_json::{json, Value}; +use tokio::sync::mpsc; + +/// Bounded queue depth. Overflow drops the newest record and increments +/// `dropped`, which surfaces as `preloop.telemetry.export{outcome="dropped"}`. +const QUEUE_CAPACITY: usize = 2048; +const BATCH_MAX: usize = 256; +const FLUSH_INTERVAL: Duration = Duration::from_secs(5); + +#[derive(Debug, Default)] +pub struct ExportHealth { + pub sent: AtomicU64, + pub failed: AtomicU64, + pub dropped: AtomicU64, + pub last_success_unix: AtomicU64, + pub last_failure_unix: AtomicU64, +} + +impl ExportHealth { + fn record_success(&self, n: u64) { + self.sent.fetch_add(n, Ordering::Relaxed); + self.last_success_unix.store(now_secs(), Ordering::Relaxed); + } + + fn record_failure(&self) { + self.failed.fetch_add(1, Ordering::Relaxed); + self.last_failure_unix.store(now_secs(), Ordering::Relaxed); + } + + pub fn dropped(&self) -> u64 { + self.dropped.load(Ordering::Relaxed) + } + + pub fn sent(&self) -> u64 { + self.sent.load(Ordering::Relaxed) + } + + pub fn failed(&self) -> u64 { + self.failed.load(Ordering::Relaxed) + } +} + +fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +/// Wall-clock nanoseconds since the epoch, for OTLP timestamps. +pub fn now_nanos() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) +} + +/// W3C Trace Context: 16-byte trace id / 8-byte span id, lowercase hex. +/// +/// Generated locally when a request arrives without a `traceparent`, or +/// adopted from the incoming header so a caller's trace continues through +/// the control plane. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SpanContext { + pub trace_id: String, + pub span_id: String, + pub parent_span_id: Option, +} + +impl SpanContext { + /// New root context with a random trace and span id. + pub fn root() -> Self { + Self { + trace_id: random_hex(16), + span_id: random_hex(8), + parent_span_id: None, + } + } + + /// Child of an incoming `traceparent`, or a new root when absent/invalid. + /// + /// Format: `00-<32 hex trace>-<16 hex span>-<2 hex flags>`. A malformed + /// header starts a new trace rather than failing the request — telemetry + /// must never reject traffic. + pub fn from_traceparent(header: Option<&str>) -> Self { + let Some(raw) = header else { + return Self::root(); + }; + let parts: Vec<&str> = raw.trim().split('-').collect(); + if parts.len() != 4 { + return Self::root(); + } + let (version, trace_id, parent_span_id) = (parts[0], parts[1], parts[2]); + let valid = version.len() == 2 + && trace_id.len() == 32 + && parent_span_id.len() == 16 + && trace_id.chars().all(|c| c.is_ascii_hexdigit()) + && parent_span_id.chars().all(|c| c.is_ascii_hexdigit()) + // All-zero ids are explicitly invalid per the spec. + && trace_id.chars().any(|c| c != '0') + && parent_span_id.chars().any(|c| c != '0'); + if !valid { + return Self::root(); + } + Self { + trace_id: trace_id.to_ascii_lowercase(), + span_id: random_hex(8), + parent_span_id: Some(parent_span_id.to_ascii_lowercase()), + } + } +} + +fn random_hex(bytes: usize) -> String { + use std::fmt::Write; + let mut out = String::with_capacity(bytes * 2); + for _ in 0..bytes { + let byte: u8 = rand::random(); + let _ = write!(out, "{byte:02x}"); + } + out +} + +/// OTLP span status. `Unset` is the default for a successful server span; +/// only an actual error sets `Error`, per the OTLP spec. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SpanStatus { + Unset, + Error, +} + +/// One completed span queued for export. +#[derive(Debug, Clone)] +pub struct SpanRecord { + pub context: SpanContext, + pub name: String, + pub start_nanos: u128, + pub end_nanos: u128, + pub status: SpanStatus, + pub attributes: Vec<(String, String)>, +} + +/// One log record queued for export. +#[derive(Debug, Clone)] +pub struct LogRecord { + pub severity: &'static str, + pub body: String, + pub attributes: Vec<(String, String)>, + /// Correlates this record with a span. OTLP carries these as first-class + /// fields, not attributes, so a backend can pivot log <-> trace. + pub trace_id: Option, + pub span_id: Option, + /// Nanoseconds since the epoch, captured at enqueue time. A batch-level + /// timestamp would collapse every record in a flush window to the same + /// instant and lose intra-batch ordering. + pub observed_unix_nanos: u128, +} + +/// Either signal, multiplexed over one bounded channel so a burst of one +/// cannot starve the other beyond the shared capacity. +#[derive(Debug, Clone)] +pub enum Item { + Log(LogRecord), + Span(SpanRecord), +} + +/// Fully-resolved destination for one signal: URL plus signal-scoped headers. +#[derive(Debug, Clone)] +pub struct SignalTarget { + pub url: String, + pub headers: Vec<(String, String)>, +} + +/// Per-signal export destinations. Resolution (which env var wins, whether +/// the generic base needs the `/v1/` suffix) happens in +/// `ObservabilityConfig`; the worker never touches the URL shape again. +#[derive(Debug, Clone, Default)] +pub struct ExportTargets { + pub logs: Option, + pub traces: Option, + pub metrics: Option, +} + +impl ExportTargets { + pub fn is_empty(&self) -> bool { + self.logs.is_none() && self.traces.is_none() && self.metrics.is_none() + } +} + +/// Handle used by the rest of the process to enqueue telemetry. +#[derive(Debug, Clone)] +pub struct Exporter { + tx: mpsc::Sender, + health: Arc, + /// Signals the worker to drain and exit. Only the runtime calls this + /// during bounded shutdown; enqueues racing the drain are best-effort. + shutdown: tokio::sync::watch::Sender, +} + +impl Exporter { + /// Enqueue a log record. Never blocks; drops on a full queue. + pub fn log(&self, record: LogRecord) { + let mut record = record; + if record.observed_unix_nanos == 0 { + record.observed_unix_nanos = now_nanos(); + } + if self.tx.try_send(Item::Log(record)).is_err() { + self.health.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + /// Enqueue a completed span. Never blocks; drops on a full queue. + pub fn span(&self, record: SpanRecord) { + if self.tx.try_send(Item::Span(record)).is_err() { + self.health.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + pub fn health(&self) -> &Arc { + &self.health + } + + /// Ask the worker to drain and exit. Used by bounded shutdown; enqueues + /// racing the drain are best-effort. + pub fn request_shutdown(&self) { + let _ = self.shutdown.send(true); + } +} + +/// Spawn the export worker. Returns `None` when no endpoint is configured, +/// so the absent-endpoint path opens no socket at all. +/// +/// One worker drains logs and spans from the shared queue and scrapes the +/// metrics registry on each tick, so all three signals share one batching +/// cadence and one client. +pub fn spawn( + targets: &ExportTargets, + service_name: &str, + instance_id: &str, + service_version: &str, + metrics: Option>, +) -> Option<(Exporter, Arc, tokio::task::JoinHandle<()>)> { + if targets.is_empty() { + return None; + } + let resource = Resource { + service_name: service_name.to_string(), + instance_id: instance_id.to_string(), + service_version: service_version.to_string(), + }; + let health = Arc::new(ExportHealth::default()); + let (tx, mut rx) = mpsc::channel::(QUEUE_CAPACITY); + let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false); + + let worker_health = health.clone(); + let targets = targets.clone(); + let join = tokio::spawn(async move { + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + { + Ok(client) => client, + Err(error) => { + // Sanitized: never the endpoint or headers. + tracing::warn!(failure = "client_build", %error, "telemetry export disabled"); + return; + } + }; + // Cumulative temporality needs a fixed start for every point, or a + // backend cannot tell a restart from a counter reset. + let start_nanos = now_nanos(); + let mut logs: Vec = Vec::with_capacity(BATCH_MAX); + let mut spans: Vec = Vec::with_capacity(BATCH_MAX); + let mut ticker = tokio::time::interval(FLUSH_INTERVAL); + loop { + tokio::select! { + maybe = rx.recv() => { + match maybe { + Some(Item::Log(record)) => { + logs.push(record); + if logs.len() >= BATCH_MAX { + flush_logs(&client, targets.logs.as_ref(), &resource, &mut logs, &worker_health).await; + } + } + Some(Item::Span(record)) => { + spans.push(record); + if spans.len() >= BATCH_MAX { + flush_spans(&client, targets.traces.as_ref(), &resource, &mut spans, &worker_health).await; + } + } + None => { + // All senders gone: final drain, then exit. + drain_and_flush(&client, &targets, &resource, &mut logs, &mut spans, &metrics, start_nanos, &worker_health).await; + break; + } + } + } + _ = shutdown_rx.changed() => { + if !*shutdown_rx.borrow() { + continue; + } + // Bounded shutdown: drain whatever is queued so a clean + // exit does not lose the last flush window's records. + drain_and_flush(&client, &targets, &resource, &mut logs, &mut spans, &metrics, start_nanos, &worker_health).await; + break; + } + _ = ticker.tick() => { + flush_logs(&client, targets.logs.as_ref(), &resource, &mut logs, &worker_health).await; + flush_spans(&client, targets.traces.as_ref(), &resource, &mut spans, &worker_health).await; + if let Some(registry) = &metrics { + flush_metrics(&client, targets.metrics.as_ref(), &resource, registry, start_nanos, &worker_health).await; + } + } + } + } + }); + + Some(( + Exporter { + tx, + health: health.clone(), + shutdown: shutdown_tx, + }, + health, + join, + )) +} + +/// Flush every signal once, then exit. Used for both the shutdown signal and +/// the channel-closed final drain. +async fn drain_and_flush( + client: &reqwest::Client, + targets: &ExportTargets, + resource: &Resource, + logs: &mut Vec, + spans: &mut Vec, + metrics: &Option>, + start_nanos: u128, + health: &Arc, +) { + flush_logs(client, targets.logs.as_ref(), resource, logs, health).await; + flush_spans(client, targets.traces.as_ref(), resource, spans, health).await; + if let Some(registry) = metrics { + flush_metrics( + client, + targets.metrics.as_ref(), + resource, + registry, + start_nanos, + health, + ) + .await; + } +} + +#[derive(Debug, Clone)] +struct Resource { + service_name: String, + instance_id: String, + service_version: String, +} + +/// POST one payload, recording health. Never propagates an error. +async fn post( + client: &reqwest::Client, + url: &str, + headers: &[(String, String)], + payload: &Value, + signal: &'static str, + count: u64, + health: &Arc, +) { + let mut req = client.post(url).json(payload); + for (name, value) in headers { + req = req.header(name.as_str(), value.as_str()); + } + match req.send().await { + Ok(response) if response.status().is_success() => { + // A 2xx is not full success: OTLP protojson may report + // `partialSuccess` with rejected records. Ignoring the body + // would overstate delivery. + let rejected = response + .bytes() + .await + .ok() + .and_then(|body| rejected_count_from_body(&body)); + match rejected { + Some(0) | None => health.record_success(count), + Some(n) => { + // Count only, never the body — it can echo credentials. + tracing::warn!( + failure = "partial_success", + signal, + rejected = n, + "telemetry export partially rejected" + ); + health.record_failure(); + } + } + } + Ok(response) => { + // Status class only — never the body, which can echo credentials. + tracing::warn!( + failure = "http_status", + signal, + status = response.status().as_u16(), + "telemetry export failed" + ); + health.record_failure(); + } + Err(_) => { + // No error text: reqwest errors embed the URL, which may carry + // credentials in userinfo. + tracing::warn!(failure = "transport", signal, "telemetry export failed"); + health.record_failure(); + } + } +} + +async fn flush_logs( + client: &reqwest::Client, + target: Option<&SignalTarget>, + resource: &Resource, + buffer: &mut Vec, + health: &Arc, +) { + let Some(target) = target else { return }; + if buffer.is_empty() { + return; + } + let batch = std::mem::take(buffer); + let count = batch.len() as u64; + let payload = encode_logs( + &batch, + &resource.service_name, + &resource.instance_id, + &resource.service_version, + ); + post( + client, + &target.url, + &target.headers, + &payload, + "logs", + count, + health, + ) + .await; +} + +async fn flush_spans( + client: &reqwest::Client, + target: Option<&SignalTarget>, + resource: &Resource, + buffer: &mut Vec, + health: &Arc, +) { + let Some(target) = target else { return }; + if buffer.is_empty() { + return; + } + let batch = std::mem::take(buffer); + let count = batch.len() as u64; + let payload = encode_spans( + &batch, + &resource.service_name, + &resource.instance_id, + &resource.service_version, + ); + post( + client, + &target.url, + &target.headers, + &payload, + "traces", + count, + health, + ) + .await; +} + +async fn flush_metrics( + client: &reqwest::Client, + target: Option<&SignalTarget>, + resource: &Resource, + registry: &Arc, + start_nanos: u128, + health: &Arc, +) { + let Some(target) = target else { return }; + let families = registry.collect(); + if families.is_empty() { + return; + } + let count = families.len() as u64; + let payload = encode_metrics( + &families, + &resource.service_name, + &resource.instance_id, + &resource.service_version, + start_nanos, + ); + post( + client, + &target.url, + &target.headers, + &payload, + "metrics", + count, + health, + ) + .await; +} + +const SCOPE_NAME: &str = "preloop-observability"; + +fn attr(key: &str, value: &str) -> Value { + json!({"key": key, "value": {"stringValue": value}}) +} + +fn resource_attributes(service_name: &str, instance_id: &str, service_version: &str) -> Vec { + vec![ + attr("service.name", service_name), + attr("service.instance.id", instance_id), + attr("service.version", service_version), + ] +} + +fn severity_number(severity: &str) -> u8 { + match severity { + "TRACE" => 1, + "DEBUG" => 5, + "INFO" => 9, + "WARN" => 13, + "ERROR" => 17, + _ => 9, + } +} + +/// Parse the OTLP protojson `partialSuccess` field for one signal. Returns +/// the number of rejected items, or `None` when the body is not an OTLP +/// response (which means full success — a plain 2xx is a valid empty ack). +fn rejected_count_from_body(body: &[u8]) -> Option { + serde_json::from_slice::(body).ok().and_then(|v| { + v.pointer("/partialSuccess/rejectedLogRecords") + .or_else(|| v.pointer("/partialSuccess/rejectedSpans")) + .or_else(|| v.pointer("/partialSuccess/rejectedDataPoints")) + .and_then(|n| n.as_u64()) + }) +} + +/// Encode a batch into the OTLP/HTTP JSON `ExportLogsServiceRequest` shape. +pub fn encode_logs( + batch: &[LogRecord], + service_name: &str, + instance_id: &str, + service_version: &str, +) -> Value { + let records: Vec = batch + .iter() + .map(|record| { + let attributes: Vec = + record.attributes.iter().map(|(k, v)| attr(k, v)).collect(); + let ts = record.observed_unix_nanos.to_string(); + let mut obj = json!({ + "timeUnixNano": ts, + "observedTimeUnixNano": ts, + "severityNumber": severity_number(record.severity), + "severityText": record.severity, + "body": {"stringValue": record.body}, + "attributes": attributes, + }); + // OTLP carries correlation as first-class fields, not attributes. + if let (Some(trace_id), Some(span_id)) = (&record.trace_id, &record.span_id) { + obj["traceId"] = json!(trace_id); + obj["spanId"] = json!(span_id); + } + obj + }) + .collect(); + json!({ + "resourceLogs": [{ + "resource": {"attributes": resource_attributes(service_name, instance_id, service_version)}, + "scopeLogs": [{ + "scope": {"name": "preloop-observability"}, + "logRecords": records, + }] + }] + }) +} + +/// Parse `OTEL_EXPORTER_OTLP_HEADERS` (`k1=v1,k2=v2`). +pub fn parse_headers(raw: Option<&str>) -> Vec<(String, String)> { + let Some(raw) = raw else { + return Vec::new(); + }; + raw.split(',') + .filter_map(|pair| { + let (k, v) = pair.split_once('=')?; + let k = k.trim(); + let v = v.trim(); + if k.is_empty() || v.is_empty() { + None + } else { + Some((k.to_string(), v.to_string())) + } + }) + .collect() +} + +/// Encode a batch into the OTLP/HTTP JSON `ExportTraceServiceRequest` shape. +pub fn encode_spans( + batch: &[SpanRecord], + service_name: &str, + instance_id: &str, + service_version: &str, +) -> Value { + let spans: Vec = batch + .iter() + .map(|record| { + let attributes: Vec = + record.attributes.iter().map(|(k, v)| attr(k, v)).collect(); + let mut obj = json!({ + "traceId": record.context.trace_id, + "spanId": record.context.span_id, + "name": record.name, + // 2 = SPAN_KIND_SERVER: every span we emit today is an + // inbound request handled by the control plane. + "kind": 2, + "startTimeUnixNano": record.start_nanos.to_string(), + "endTimeUnixNano": record.end_nanos.to_string(), + "attributes": attributes, + "status": {"code": match record.status { + SpanStatus::Unset => 0, + SpanStatus::Error => 2, + }}, + }); + if let Some(parent) = &record.context.parent_span_id { + obj["parentSpanId"] = json!(parent); + } + obj + }) + .collect(); + json!({ + "resourceSpans": [{ + "resource": {"attributes": resource_attributes(service_name, instance_id, service_version)}, + "scopeSpans": [{ + "scope": {"name": SCOPE_NAME}, + "spans": spans, + }] + }] + }) +} + +/// Encode collected families into the OTLP/HTTP JSON `ExportMetricsServiceRequest`. +/// +/// All points are cumulative (`aggregationTemporality: 2`) and share the +/// process start as `startTimeUnixNano`, so a backend can distinguish a +/// restart from a counter reset. +pub fn encode_metrics( + families: &[crate::metrics::MetricFamily], + service_name: &str, + instance_id: &str, + service_version: &str, + start_nanos: u128, +) -> Value { + use crate::metrics::MetricPoint; + const CUMULATIVE: u8 = 2; + let now = now_nanos().to_string(); + let start = start_nanos.to_string(); + + let metrics: Vec = families + .iter() + .map(|family| { + let mut metric = json!({"name": family.name, "unit": family.unit}); + match family.points.first() { + Some(MetricPoint::Sum { .. }) => { + let points: Vec = family + .points + .iter() + .filter_map(|point| match point { + MetricPoint::Sum { value, attributes } => Some(json!({ + "asDouble": value, + "startTimeUnixNano": start, + "timeUnixNano": now, + "attributes": encode_attrs(attributes), + })), + _ => None, + }) + .collect(); + metric["sum"] = json!({ + "dataPoints": points, + "aggregationTemporality": CUMULATIVE, + "isMonotonic": true, + }); + } + Some(MetricPoint::Gauge { .. }) => { + let points: Vec = family + .points + .iter() + .filter_map(|point| match point { + MetricPoint::Gauge { value, attributes } => Some(json!({ + "asDouble": value, + "timeUnixNano": now, + "attributes": encode_attrs(attributes), + })), + _ => None, + }) + .collect(); + metric["gauge"] = json!({"dataPoints": points}); + } + Some(MetricPoint::Histogram { .. }) => { + let points: Vec = family + .points + .iter() + .filter_map(|point| match point { + MetricPoint::Histogram { + count, + sum, + bounds, + bucket_counts, + attributes, + } => Some(json!({ + "count": count.to_string(), + "sum": sum, + "explicitBounds": bounds, + "bucketCounts": bucket_counts + .iter() + .map(|c| c.to_string()) + .collect::>(), + "startTimeUnixNano": start, + "timeUnixNano": now, + "attributes": encode_attrs(attributes), + })), + _ => None, + }) + .collect(); + metric["histogram"] = json!({ + "dataPoints": points, + "aggregationTemporality": CUMULATIVE, + }); + } + None => {} + } + metric + }) + .collect(); + + json!({ + "resourceMetrics": [{ + "resource": {"attributes": resource_attributes(service_name, instance_id, service_version)}, + "scopeMetrics": [{ + "scope": {"name": SCOPE_NAME}, + "metrics": metrics, + }] + }] + }) +} + +fn encode_attrs(attributes: &[(String, String)]) -> Vec { + attributes.iter().map(|(k, v)| attr(k, v)).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn absent_endpoint_spawns_nothing() { + assert!(spawn(&ExportTargets::default(), "preloop", "abc", "9.9.9", None).is_none()); + } + + #[test] + fn headers_parse_pairs() { + let parsed = parse_headers(Some("Authorization=Basic abc,stream-name=default")); + assert_eq!(parsed.len(), 2); + assert_eq!(parsed[0].0, "Authorization"); + assert_eq!(parsed[1].1, "default"); + } + + #[test] + fn headers_ignore_malformed() { + assert!(parse_headers(Some("novalue,=empty,k=")).is_empty()); + } + + #[test] + fn encodes_otlp_log_shape() { + let batch = vec![LogRecord { + severity: "WARN", + body: "pool provisioning failed".to_string(), + attributes: vec![("event.name".to_string(), "pool.provision".to_string())], + trace_id: None, + span_id: None, + observed_unix_nanos: 1_000, + }]; + let payload = encode_logs(&batch, "preloop", "inst-1", "9.9.9"); + let record = &payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0]; + assert_eq!(record["severityText"], "WARN"); + assert_eq!(record["severityNumber"], 13); + assert_eq!(record["body"]["stringValue"], "pool provisioning failed"); + let resource = &payload["resourceLogs"][0]["resource"]["attributes"]; + assert_eq!(resource[0]["key"], "service.name"); + assert_eq!(resource[0]["value"]["stringValue"], "preloop"); + assert_eq!(resource[2]["key"], "service.version"); + assert_eq!( + resource[2]["value"]["stringValue"], "9.9.9", + "service.version must come from the host binary, not this crate" + ); + } +} + +#[cfg(test)] +mod signal_tests { + use super::*; + use crate::metrics::{MetricFamily, MetricPoint}; + + #[test] + fn traceparent_is_adopted_as_parent() { + let ctx = SpanContext::from_traceparent(Some( + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + )); + assert_eq!(ctx.trace_id, "4bf92f3577b34da6a3ce929d0e0e4736"); + assert_eq!(ctx.parent_span_id.as_deref(), Some("00f067aa0ba902b7")); + // A child must get its own span id, never reuse the parent's. + assert_ne!(ctx.span_id, "00f067aa0ba902b7"); + assert_eq!(ctx.span_id.len(), 16); + } + + #[test] + fn malformed_traceparent_starts_a_new_root() { + for bad in [ + "garbage", + "00-tooshort-00f067aa0ba902b7-01", + // All-zero ids are invalid per the spec. + "00-00000000000000000000000000000000-00f067aa0ba902b7-01", + "00-4bf92f3577b34da6a3ce929d0e0e4736-0000000000000000-01", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7", + ] { + let ctx = SpanContext::from_traceparent(Some(bad)); + assert!( + ctx.parent_span_id.is_none(), + "{bad} must not adopt a parent" + ); + assert_eq!(ctx.trace_id.len(), 32); + } + } + + #[test] + fn generated_ids_are_unique_and_well_formed() { + let a = SpanContext::root(); + let b = SpanContext::root(); + assert_ne!(a.trace_id, b.trace_id); + assert_eq!(a.trace_id.len(), 32); + assert_eq!(a.span_id.len(), 8 * 2); + assert!(a.trace_id.chars().all(|c| c.is_ascii_hexdigit())); + } + + #[test] + fn spans_encode_to_otlp_shape() { + let ctx = SpanContext::from_traceparent(Some( + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + )); + let batch = vec![SpanRecord { + context: ctx, + name: "GET /api/v1/runs/:run_id".to_string(), + start_nanos: 1_000, + end_nanos: 2_000, + status: SpanStatus::Error, + attributes: vec![("http.route".to_string(), "/api/v1/runs/:run_id".to_string())], + }]; + let payload = encode_spans(&batch, "preloop", "inst", "0.2.0"); + let span = &payload["resourceSpans"][0]["scopeSpans"][0]["spans"][0]; + assert_eq!(span["traceId"], "4bf92f3577b34da6a3ce929d0e0e4736"); + assert_eq!(span["parentSpanId"], "00f067aa0ba902b7"); + assert_eq!(span["kind"], 2, "server span"); + assert_eq!(span["status"]["code"], 2, "error"); + assert_eq!(span["startTimeUnixNano"], "1000"); + } + + #[test] + fn partial_success_is_detected_for_every_signal() { + assert_eq!( + rejected_count_from_body(br#"{"partialSuccess":{"rejectedLogRecords":3}}"#), + Some(3) + ); + assert_eq!( + rejected_count_from_body(br#"{"partialSuccess":{"rejectedSpans":1}}"#), + Some(1) + ); + assert_eq!( + rejected_count_from_body(br#"{"partialSuccess":{"rejectedDataPoints":7}}"#), + Some(7) + ); + // Empty ack or arbitrary JSON means full success. + assert_eq!(rejected_count_from_body(b"{}"), None); + assert_eq!(rejected_count_from_body(b""), None); + // Partial rejection with zero rejected is full success. + assert_eq!( + rejected_count_from_body(br#"{"partialSuccess":{"rejectedLogRecords":0}}"#), + Some(0) + ); + } + + #[test] + fn logs_keep_their_own_enqueue_timestamp() { + let batch = vec![ + LogRecord { + severity: "INFO", + body: "first".to_string(), + attributes: vec![], + trace_id: None, + span_id: None, + observed_unix_nanos: 1_000, + }, + LogRecord { + severity: "INFO", + body: "second".to_string(), + attributes: vec![], + trace_id: None, + span_id: None, + observed_unix_nanos: 2_000, + }, + ]; + let payload = encode_logs(&batch, "preloop", "inst", "0.2.0"); + let records = &payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"]; + assert_eq!(records[0]["timeUnixNano"], "1000"); + assert_eq!(records[1]["timeUnixNano"], "2000"); + } + + #[test] + fn logs_carry_trace_correlation_as_fields() { + let batch = vec![LogRecord { + severity: "WARN", + body: "job.status.terminal".to_string(), + attributes: vec![], + trace_id: Some("4bf92f3577b34da6a3ce929d0e0e4736".to_string()), + span_id: Some("00f067aa0ba902b7".to_string()), + observed_unix_nanos: 1_000, + }]; + let payload = encode_logs(&batch, "preloop", "inst", "0.2.0"); + let record = &payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0]; + // OTLP requires these as fields, not attributes, for log<->trace pivot. + assert_eq!(record["traceId"], "4bf92f3577b34da6a3ce929d0e0e4736"); + assert_eq!(record["spanId"], "00f067aa0ba902b7"); + } + + #[test] + fn counters_encode_as_cumulative_monotonic_sums() { + let families = vec![MetricFamily { + name: "preloop.job.completed".to_string(), + unit: "{job}", + points: vec![MetricPoint::Sum { + value: 7.0, + attributes: vec![("preloop.conclusion".to_string(), "failure".to_string())], + }], + }]; + let payload = encode_metrics(&families, "preloop", "inst", "0.2.0", 500); + let metric = &payload["resourceMetrics"][0]["scopeMetrics"][0]["metrics"][0]; + assert_eq!(metric["name"], "preloop.job.completed"); + assert_eq!(metric["sum"]["isMonotonic"], true); + assert_eq!(metric["sum"]["aggregationTemporality"], 2, "cumulative"); + let point = &metric["sum"]["dataPoints"][0]; + assert_eq!(point["asDouble"], 7.0); + assert_eq!( + point["startTimeUnixNano"], "500", + "cumulative points need a fixed start or a restart reads as a reset" + ); + } + + #[test] + fn histograms_encode_with_the_implicit_inf_bucket() { + // `bucket_counts` is the per-bucket (disjoint) form OTLP requires: + // [1, 2, 2] means 1 under 0.005, 2 between 0.005 and 0.01, 2 above. + // The sum must equal `count` or the data point is invalid. + let families = vec![MetricFamily { + name: "http.server.request.duration".to_string(), + unit: "s", + points: vec![MetricPoint::Histogram { + count: 5, + sum: 0.25, + bounds: vec![0.005, 0.01], + bucket_counts: vec![1, 2, 2], + attributes: vec![("http.route".to_string(), "/healthz".to_string())], + }], + }]; + let payload = encode_metrics(&families, "preloop", "inst", "0.2.0", 0); + let point = &payload["resourceMetrics"][0]["scopeMetrics"][0]["metrics"][0]["histogram"] + ["dataPoints"][0]; + let bounds = point["explicitBounds"].as_array().unwrap(); + let counts = point["bucketCounts"].as_array().unwrap(); + assert_eq!( + counts.len(), + bounds.len() + 1, + "OTLP requires one more bucket count than bounds (+Inf)" + ); + let sum: u64 = counts + .iter() + .map(|c| c.as_str().unwrap().parse::().unwrap()) + .sum(); + assert_eq!(sum, 5, "per-bucket counts must conserve the total count"); + assert_eq!(point["count"], "5"); + } + + #[test] + fn gauges_have_no_temporality_or_start_time() { + let families = vec![MetricFamily { + name: "http.server.active_requests".to_string(), + unit: "{request}", + points: vec![MetricPoint::Gauge { + value: 3.0, + attributes: vec![], + }], + }]; + let payload = encode_metrics(&families, "preloop", "inst", "0.2.0", 0); + let metric = &payload["resourceMetrics"][0]["scopeMetrics"][0]["metrics"][0]; + assert!(metric["gauge"]["dataPoints"][0]["asDouble"] == 3.0); + assert!(metric["gauge"]["aggregationTemporality"].is_null()); + } +} diff --git a/crates/preloop-observability/src/lib.rs b/crates/preloop-observability/src/lib.rs index 3da05d42..df1b291c 100644 --- a/crates/preloop-observability/src/lib.rs +++ b/crates/preloop-observability/src/lib.rs @@ -11,8 +11,10 @@ //! - Always retain `stderr`/`journald` even when OTLP is configured. //! - `Debug` on config never reveals headers or credential-bearing endpoint parts. +pub mod export; pub mod metrics; pub mod status; +pub mod vm_telemetry; use std::collections::HashMap; use std::fmt; @@ -70,13 +72,23 @@ pub struct ObservabilityConfig { pub rust_log: String, /// `service.name` — `preloop` or `OTEL_SERVICE_NAME`. pub service_name: String, + /// `service.version` — set by the host binary via `with_service_version`. + /// Defaults to this crate's version only until the binary overrides it. + pub service_version: String, /// Per-process instance ID (UUID v4). pub instance_id: String, - /// `OTEL_EXPORTER_OTLP_ENDPOINT` or signal-specific variant, if any. Kept as - /// given for transport, but `Debug` redacts userinfo/query. - otel_endpoint: Option, - /// `OTEL_EXPORTER_OTLP_HEADERS` or signal-specific variant, if any. Never shown in `Debug` or errors. - otel_headers: Option, + /// Per-signal OTLP endpoints, each fully resolved: a signal-specific + /// variable wins and is used as-is; the generic base gets the + /// `/v1/` suffix appended. Kept raw for transport, but `Debug` + /// redacts userinfo/query. + otel_logs_endpoint: Option, + otel_traces_endpoint: Option, + otel_metrics_endpoint: Option, + /// Per-signal `OTEL_EXPORTER_OTLP_*_HEADERS`, signal-specific first, + /// generic fallback applied per signal. Never shown in `Debug` or errors. + otel_logs_headers: Option, + otel_traces_headers: Option, + otel_metrics_headers: Option, /// Whether any OTLP endpoint is present (i.e. export enabled). pub otlp_enabled: bool, } @@ -100,49 +112,94 @@ impl ObservabilityConfig { let service_name = std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| "preloop".to_string()); - // Endpoint: generic or signal-specific. Presence — not value — enables export. - // This is a deliberate deviation from the OTel spec default `http://localhost:4318`. - let otel_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") - .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")) - .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT")) - .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT")) + // Presence — not value — enables export. This is a deliberate + // deviation from the OTel spec default `http://localhost:4318`. + let generic = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") .ok() - .filter(|v| !v.trim().is_empty() && v.trim() != "none"); + .filter(|v| !v.trim().is_empty() && v.trim() != "none") + .map(|v| v.trim_end_matches('/').to_string()); + + // Signal-specific endpoint is a complete URL used as-is; the generic + // base needs the /v1/ suffix. Appending the suffix to a + // signal-specific URL would produce `/v1/traces/v1/traces`, and + // sending every signal to one signal-specific URL misroutes the rest. + let resolve = |var: &str, suffix: &str| { + std::env::var(var) + .ok() + .filter(|v| !v.trim().is_empty() && v.trim() != "none") + .map(|v| v.trim_end_matches('/').to_string()) + .or_else(|| generic.as_ref().map(|g| format!("{g}{suffix}"))) + }; + let otel_logs_endpoint = resolve("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", "/v1/logs"); + let otel_traces_endpoint = resolve("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "/v1/traces"); + let otel_metrics_endpoint = resolve("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", "/v1/metrics"); - let otel_headers = std::env::var("OTEL_EXPORTER_OTLP_HEADERS") - .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_TRACES_HEADERS")) - .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_METRICS_HEADERS")) - .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_LOGS_HEADERS")) + let generic_headers = std::env::var("OTEL_EXPORTER_OTLP_HEADERS") .ok() .filter(|v| !v.trim().is_empty()); + let resolve_headers = |var: &str| { + std::env::var(var) + .ok() + .filter(|v| !v.trim().is_empty()) + .or_else(|| generic_headers.clone()) + }; + let otel_logs_headers = resolve_headers("OTEL_EXPORTER_OTLP_LOGS_HEADERS"); + let otel_traces_headers = resolve_headers("OTEL_EXPORTER_OTLP_TRACES_HEADERS"); + let otel_metrics_headers = resolve_headers("OTEL_EXPORTER_OTLP_METRICS_HEADERS"); - let otlp_enabled = otel_endpoint.is_some(); + let otlp_enabled = otel_logs_endpoint.is_some() + || otel_traces_endpoint.is_some() + || otel_metrics_endpoint.is_some(); let instance_id = Uuid::new_v4().to_string(); Self { log_format, rust_log, service_name, + service_version: env!("CARGO_PKG_VERSION").to_string(), instance_id, - otel_endpoint, - otel_headers, + otel_logs_endpoint, + otel_traces_endpoint, + otel_metrics_endpoint, + otel_logs_headers, + otel_traces_headers, + otel_metrics_headers, otlp_enabled, } } - /// Raw endpoint if export is enabled, for transport construction. - pub fn otel_endpoint_raw(&self) -> Option<&str> { - self.otel_endpoint.as_deref() + /// Fully-resolved per-signal destinations for the export worker. + pub fn export_targets(&self) -> export::ExportTargets { + let signal = |endpoint: &Option, headers: &Option| { + endpoint.as_ref().map(|url| export::SignalTarget { + url: url.clone(), + headers: export::parse_headers(headers.as_deref()), + }) + }; + export::ExportTargets { + logs: signal(&self.otel_logs_endpoint, &self.otel_logs_headers), + traces: signal(&self.otel_traces_endpoint, &self.otel_traces_headers), + metrics: signal(&self.otel_metrics_endpoint, &self.otel_metrics_headers), + } } /// Whether any `OTEL_EXPORTER_OTLP_HEADERS` was supplied (for health reporting). pub fn has_otel_headers(&self) -> bool { - self.otel_headers.is_some() + self.otel_logs_headers.is_some() + || self.otel_traces_headers.is_some() + || self.otel_metrics_headers.is_some() + } + + /// Override `service.version` with the host binary's version. The crate's + /// own version is meaningless to an operator reading telemetry. + pub fn with_service_version(mut self, version: &str) -> Self { + self.service_version = version.to_string(); + self } /// Sanitized endpoint for `Debug`/errors: strips userinfo and query. fn sanitized_endpoint(&self) -> Option { - self.otel_endpoint.as_ref().map(|raw| { + self.otel_logs_endpoint.as_ref().map(|raw| { // Best-effort: hide `user:pass@` and `?...` without a URL parser dep. let without_query = raw.split('?').next().unwrap_or(raw); if let Some(at) = without_query.rfind('@') { @@ -164,14 +221,12 @@ impl fmt::Debug for ObservabilityConfig { .field("log_format", &self.log_format) .field("rust_log", &self.rust_log) .field("service_name", &self.service_name) + .field("service_version", &self.service_version) .field("instance_id", &self.instance_id) - .field( - "otel_endpoint", - &self.sanitized_endpoint().map(|_| ""), - ) + .field("otel_endpoint", &self.sanitized_endpoint()) .field( "otel_headers", - &self.otel_headers.as_ref().map(|_| ""), + &(self.has_otel_headers().then(|| "")), ) .field("otlp_enabled", &self.otlp_enabled) .finish() @@ -203,8 +258,6 @@ pub struct TaskHeartbeat { struct HeartbeatEntry { critical: Criticality, last_beat: Instant, - /// Whether the task has exited cleanly (Drop without panic). - exited: bool, } impl TaskHeartbeat { @@ -215,7 +268,6 @@ impl TaskHeartbeat { HeartbeatEntry { critical, last_beat: Instant::now(), - exited: false, }, ); HeartbeatHandle { @@ -235,12 +287,6 @@ impl TaskHeartbeat { self.inner.write().remove(name); } - pub(crate) fn mark_exited(&self, name: &'static str) { - if let Some(entry) = self.inner.write().get_mut(name) { - entry.exited = true; - } - } - /// Snapshot for `/readyz` and `/api/v1/status`. pub fn snapshot(&self) -> Vec { self.inner @@ -250,7 +296,6 @@ impl TaskHeartbeat { name, critical: e.critical, heartbeat_age: e.last_beat.elapsed(), - exited: e.exited, }) .collect() } @@ -260,8 +305,7 @@ impl TaskHeartbeat { // Hold read lock across iteration to avoid TOCTOU. let guard = self.inner.read(); for (name, e) in guard.iter() { - if e.critical == Criticality::Critical && !e.exited && e.last_beat.elapsed() > threshold - { + if e.critical == Criticality::Critical && e.last_beat.elapsed() > threshold { return Some(*name); } } @@ -306,7 +350,6 @@ pub struct TaskSnapshot { pub name: &'static str, pub critical: Criticality, pub heartbeat_age: Duration, - pub exited: bool, } // --------------------------------------------------------------------------- @@ -327,16 +370,11 @@ struct LimitEntry { } impl LimitRegistry { - /// Register a cap with its configured ceiling. Idempotent. + /// Register a cap with its configured ceiling. Re-registration updates + /// the ceiling and keeps the counters. One lock acquisition so the write + /// is atomic against a concurrent `record_drop`/`record_reject`. pub fn register(&self, limit: &'static str, value: usize) { - self.inner.write().entry(limit).or_insert(LimitEntry { - value, - ..Default::default() - }); - // Update value if re-registered with different ceiling (for tests). - if let Some(entry) = self.inner.write().get_mut(limit) { - entry.value = value; - } + self.inner.write().entry(limit).or_default().value = value; } pub fn record_drop(&self, limit: &'static str, n: u64) { @@ -383,6 +421,8 @@ struct Inner { heartbeat: TaskHeartbeat, limits: LimitRegistry, metrics: Arc, + vm_registry: Arc, + exporter: Option, is_noop: bool, } @@ -399,9 +439,14 @@ impl Observability { log_format: LogFormat::Auto, rust_log: "info".to_string(), service_name: "preloop".to_string(), + service_version: env!("CARGO_PKG_VERSION").to_string(), instance_id: Uuid::new_v4().to_string(), - otel_endpoint: None, - otel_headers: None, + otel_logs_endpoint: None, + otel_traces_endpoint: None, + otel_metrics_endpoint: None, + otel_logs_headers: None, + otel_traces_headers: None, + otel_metrics_headers: None, otlp_enabled: false, }; Self { @@ -410,6 +455,8 @@ impl Observability { heartbeat: TaskHeartbeat::default(), limits: LimitRegistry::default(), metrics: Arc::new(metrics::MetricsRegistry::default()), + vm_registry: Arc::new(vm_telemetry::VmTelemetryRegistry::default()), + exporter: None, is_noop: true, }), } @@ -418,16 +465,31 @@ impl Observability { /// Real handle from `ObservabilityConfig`. Does not install the global subscriber — pair with `ObservabilityRuntime`. pub fn from_config(config: ObservabilityConfig) -> (Self, ObservabilityRuntime) { let is_noop = !config.otlp_enabled; + // Absent endpoint spawns nothing at all — no worker, no socket. + // The registry is shared with the worker so metric export scrapes the + // same instruments `/metrics` renders — one source, never two. + let metrics = Arc::new(metrics::MetricsRegistry::default()); + let spawned = export::spawn( + &config.export_targets(), + &config.service_name, + &config.instance_id, + &config.service_version, + Some(metrics.clone()), + ); + let exporter = spawned.as_ref().map(|(exporter, _, _)| exporter.clone()); + let worker = spawned.map(|(exporter, _, join)| (exporter, join)); let handle = Self { inner: Arc::new(Inner { config: Arc::new(config), heartbeat: TaskHeartbeat::default(), limits: LimitRegistry::default(), - metrics: Arc::new(metrics::MetricsRegistry::default()), + metrics, + vm_registry: Arc::new(vm_telemetry::VmTelemetryRegistry::default()), + exporter, is_noop, }), }; - let runtime = ObservabilityRuntime::new(handle.clone()); + let runtime = ObservabilityRuntime::new(handle.clone(), worker); (handle, runtime) } @@ -459,6 +521,61 @@ impl Observability { &self.inner.metrics } + pub fn vm_registry(&self) -> &vm_telemetry::VmTelemetryRegistry { + &self.inner.vm_registry + } + + /// Enqueue a log record for OTLP export. No-op when export is disabled. + pub fn export_log( + &self, + severity: &'static str, + body: impl Into, + attributes: Vec<(String, String)>, + ) { + self.export_log_in_span(severity, body, attributes, None); + } + + /// As [`Observability::export_log`], correlated with a span so a backend + /// can pivot from a log line to the request that produced it. + pub fn export_log_in_span( + &self, + severity: &'static str, + body: impl Into, + attributes: Vec<(String, String)>, + context: Option<&export::SpanContext>, + ) { + if let Some(exporter) = &self.inner.exporter { + exporter.log(export::LogRecord { + severity, + body: body.into(), + attributes, + trace_id: context.map(|c| c.trace_id.clone()), + span_id: context.map(|c| c.span_id.clone()), + // Stamp at enqueue, not at flush: a batch-level timestamp + // would collapse a flush window's records to one instant. + observed_unix_nanos: export::now_nanos(), + }); + } + } + + /// Enqueue a completed span. No-op when export is disabled. + pub fn export_span(&self, record: export::SpanRecord) { + if let Some(exporter) = &self.inner.exporter { + exporter.span(record); + } + } + + /// Whether spans are worth building. Lets a caller skip id and timestamp + /// work entirely when nothing would consume the result. + pub fn tracing_enabled(&self) -> bool { + self.inner.exporter.is_some() + } + + /// Export health for `/api/v1/status` and `preloop.telemetry.export`. + pub fn export_health(&self) -> Option<&Arc> { + self.inner.exporter.as_ref().map(|e| e.health()) + } + pub fn config(&self) -> &ObservabilityConfig { &self.inner.config } @@ -470,6 +587,10 @@ impl Observability { /// Tests use scoped subscribers and never install the global one twice. pub struct ObservabilityRuntime { _handle: Observability, + // The export worker plus its completion handle, taken when the worker + // exists. `shutdown` drops the sender so the worker drains, then awaits + // the join inside the 2s bound. + worker: Option<(export::Exporter, tokio::task::JoinHandle<()>)>, // Hold the tracing guard so it isn't dropped early when we use a // non-global dispatcher in tests. For the global install, this is `None` // and the global dispatcher owns the guard. @@ -482,12 +603,15 @@ pub struct ObservabilityRuntime { } impl ObservabilityRuntime { - fn new(handle: Observability) -> Self { + fn new( + handle: Observability, + worker: Option<(export::Exporter, tokio::task::JoinHandle<()>)>, + ) -> Self { // Does not install the global subscriber here — the binaries do - // that via `install_fmt_subscriber`. This runtime is the place for the - // future OTel provider guards and the 2s flush on `Drop`. + // that via `install_fmt_subscriber`. Self { _handle: handle, + worker, _guard: None, } } @@ -519,16 +643,18 @@ impl ObservabilityRuntime { } } - /// Attempt to flush exporters for at most 2s. Export failure is logged, never propagated. - pub async fn shutdown(self) { - // No exporter worker yet; this is the bounded-flush seam for - // the future OTel BatchSpanProcessor / metrics reader. - tokio::time::timeout(Duration::from_secs(2), async { - // No-op until OTLP providers are wired. - tokio::task::yield_now().await; - }) - .await - .ok(); + /// Flush exporters for at most 2s, then exit. The worker is signalled to + /// drain its buffers; if it cannot finish within the bound the remaining + /// records are dropped rather than delaying shutdown. Export failure is + /// logged by the worker, never propagated. + pub async fn shutdown(mut self) { + let Some((exporter, join)) = self.worker.take() else { + return; + }; + exporter.request_shutdown(); + tokio::time::timeout(Duration::from_secs(2), join) + .await + .ok(); } } @@ -546,6 +672,17 @@ impl fmt::Debug for ObservabilityRuntime { mod tests { use super::*; + /// `ObservabilityConfig::from_env` reads process-global `OTEL_*` + /// variables, and several tests mutate them. Cargo runs unit tests on + /// many threads inside one process, so these tests race on the + /// environment and fail intermittently unless serialized. + static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + /// Guard that serializes the env-mutating tests. + fn env_guard() -> std::sync::MutexGuard<'static, ()> { + ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()) + } + #[test] fn noop_performs_no_network_io() { let obs = Observability::noop(); @@ -556,6 +693,7 @@ mod tests { #[test] fn absent_endpoint_means_disabled_not_localhost() { + let _guard = env_guard(); // Ensure no ambient OTEL vars leak into the test. for k in [ "OTEL_EXPORTER_OTLP_ENDPOINT", @@ -579,7 +717,7 @@ mod tests { !cfg.otlp_enabled, "absent endpoint must be disabled, not localhost:4318" ); - assert!(cfg.otel_endpoint_raw().is_none()); + assert!(!cfg.otlp_enabled); let (obs, _rt) = Observability::from_config(cfg); assert!(!obs.otlp_enabled()); assert!(obs.is_noop()); @@ -587,12 +725,70 @@ mod tests { #[test] fn none_disables_signal() { + let _guard = env_guard(); std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", "none"); let cfg = ObservabilityConfig::from_env(); assert!(!cfg.otlp_enabled, "`none` must disable export"); std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT"); } + #[test] + fn signal_specific_endpoint_is_used_as_is() { + let _guard = env_guard(); + for k in [ + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + ] { + std::env::remove_var(k); + } + std::env::set_var( + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "http://collector:4318/v1/traces", + ); + let cfg = ObservabilityConfig::from_env(); + let targets = cfg.export_targets(); + assert!(cfg.otlp_enabled); + // The signal-specific URL must be used verbatim — appending the + // suffix would produce /v1/traces/v1/traces. + assert_eq!( + targets.traces.as_ref().unwrap().url, + "http://collector:4318/v1/traces" + ); + // And it must not hijack the other signals. + assert!(targets.logs.is_none()); + assert!(targets.metrics.is_none()); + std::env::remove_var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"); + } + + #[test] + fn generic_endpoint_gets_the_signal_suffix() { + let _guard = env_guard(); + for k in [ + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + ] { + std::env::remove_var(k); + } + std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318"); + let cfg = ObservabilityConfig::from_env(); + let targets = cfg.export_targets(); + assert_eq!( + targets.logs.as_ref().unwrap().url, + "http://collector:4318/v1/logs" + ); + assert_eq!( + targets.traces.as_ref().unwrap().url, + "http://collector:4318/v1/traces" + ); + assert_eq!( + targets.metrics.as_ref().unwrap().url, + "http://collector:4318/v1/metrics" + ); + std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT"); + } + #[test] fn heartbeat_register_beat_deregister() { let obs = Observability::noop(); @@ -660,6 +856,7 @@ mod tests { #[test] fn debug_redacts_headers_and_endpoint_userinfo() { + let _guard = env_guard(); std::env::set_var( "OTEL_EXPORTER_OTLP_ENDPOINT", "https://user:secret@example.com:4318/v1/traces?token=abc", diff --git a/crates/preloop-observability/src/metrics.rs b/crates/preloop-observability/src/metrics.rs index 5943bcc9..0f8c97e7 100644 --- a/crates/preloop-observability/src/metrics.rs +++ b/crates/preloop-observability/src/metrics.rs @@ -62,18 +62,29 @@ pub struct HttpLabels { pub status_class: String, } +/// Identity of an in-flight request. Deliberately carries no +/// `status_class`: the status is unknown until the response completes, so any +/// status in the key would let the increment (pre-response) and decrement +/// (post-response) disagree and permanently leak the gauge. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct ActiveLabels { + pub method: String, + pub route: String, + pub surface: String, +} + #[derive(Debug, Default)] pub struct HttpMetrics { - active: RwLock>, + active: RwLock>, durations: RwLock>, } impl HttpMetrics { - pub fn inc_active(&self, labels: &HttpLabels) { + pub fn inc_active(&self, labels: &ActiveLabels) { *self.active.write().entry(labels.clone()).or_insert(0) += 1; } - pub fn dec_active(&self, labels: &HttpLabels) { + pub fn dec_active(&self, labels: &ActiveLabels) { let mut g = self.active.write(); if let Some(v) = g.get_mut(labels) { *v -= 1; @@ -97,23 +108,26 @@ impl HttpMetrics { out.push_str("# TYPE http_server_request_duration_seconds histogram\n"); let g = self.durations.read(); for (labels, hist) in g.iter() { + let method = escape_label(&labels.method); + let route = escape_label(&labels.route); + let surface = escape_label(&labels.surface); + let status = escape_label(&labels.status_class); for (le, cnt) in &hist.buckets { out.push_str(&format!( - "http_server_request_duration_seconds_bucket{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\",le=\"{}\"}} {}\n", - labels.method, labels.route, labels.surface, labels.status_class, le, cnt + "http_server_request_duration_seconds_bucket{{method=\"{method}\",route=\"{route}\",surface=\"{surface}\",status_class=\"{status}\",le=\"{le}\"}} {cnt}\n" )); } out.push_str(&format!( - "http_server_request_duration_seconds_bucket{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\",le=\"+Inf\"}} {}\n", - labels.method, labels.route, labels.surface, labels.status_class, hist.count + "http_server_request_duration_seconds_bucket{{method=\"{method}\",route=\"{route}\",surface=\"{surface}\",status_class=\"{status}\",le=\"+Inf\"}} {}\n", + hist.count )); out.push_str(&format!( - "http_server_request_duration_seconds_sum{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\"}} {}\n", - labels.method, labels.route, labels.surface, labels.status_class, hist.sum + "http_server_request_duration_seconds_sum{{method=\"{method}\",route=\"{route}\",surface=\"{surface}\",status_class=\"{status}\"}} {}\n", + hist.sum )); out.push_str(&format!( - "http_server_request_duration_seconds_count{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\"}} {}\n", - labels.method, labels.route, labels.surface, labels.status_class, hist.count + "http_server_request_duration_seconds_count{{method=\"{method}\",route=\"{route}\",surface=\"{surface}\",status_class=\"{status}\"}} {}\n", + hist.count )); } out.push_str("# HELP http_server_active_requests Current HTTP concurrency\n"); @@ -122,7 +136,10 @@ impl HttpMetrics { for (labels, v) in g2.iter() { out.push_str(&format!( "http_server_active_requests{{method=\"{}\",route=\"{}\",surface=\"{}\"}} {}\n", - labels.method, labels.route, labels.surface, v + escape_label(&labels.method), + escape_label(&labels.route), + escape_label(&labels.surface), + v )); } } @@ -454,10 +471,6 @@ pub fn classify_surface(route: &str) -> &'static str { pub fn normalize_route(raw: &str) -> String { // Strip query let path = raw.split('?').next().unwrap_or(raw); - // Already a template? (contains ':') - if path.contains(':') { - return path.to_string(); - } // Known templates — longest prefix first const TEMPLATES: &[&str] = &[ "/api/v1/runs/:run_id", @@ -501,7 +514,9 @@ pub fn normalize_route(raw: &str) -> String { // Ensure it's a segment boundary: /api/v1/runs/abc should match /api/v1/runs/:run_id // but /api/v1/runsXYZ should not. let rest = &path[prefix.len()..]; - if rest.is_empty() || rest.starts_with('/') { + // A parameterized template needs a non-empty child segment; + // the bare collection path is matched by its own entry. + if rest.len() > 1 && rest.starts_with('/') { return tmpl.to_string(); } } @@ -511,6 +526,22 @@ pub fn normalize_route(raw: &str) -> String { "/unknown".to_string() } +/// Escape a label value for Prometheus exposition text. The label set is +/// bounded, but a quote, backslash, or newline would corrupt the entire +/// scrape; defense in depth on top of the bounded construction. +fn escape_label(value: &str) -> String { + let mut out = String::with_capacity(value.len()); + for ch in value.chars() { + match ch { + '\\' => out.push_str("\\\\"), + '"' => out.push_str("\\\""), + '\n' => out.push_str("\\n"), + _ => out.push(ch), + } + } + out +} + pub fn status_class(status: u16) -> &'static str { match status { 200..=299 => "2xx", @@ -537,6 +568,35 @@ mod tests { ); assert_eq!(normalize_route("/api/v1/status"), "/api/v1/status"); assert_eq!(normalize_route("/unknown/path/xyz"), "/unknown"); + // A bare collection path resolves to its own template, not the + // single-item one — otherwise list latency is reported under the + // item route. + assert_eq!(normalize_route("/api/v1/runs"), "/api/v1/runs"); + assert_eq!( + normalize_route("/_apis/artifactcache/cache"), + "/_apis/artifactcache/cache" + ); + assert_eq!( + normalize_route("/runner/server/_apis/distributedtask/pools"), + "/runner/server/_apis/distributedtask/pools" + ); + } + + #[test] + fn colon_in_path_cannot_escape_the_template_set() { + // A colon is legal inside a path segment; an unauthenticated 404 can + // hit /evil:anything and must still land on the constant label. + assert_eq!(normalize_route("/evil:1234/path"), "/unknown"); + assert_eq!(normalize_route("/api/v1/runs:junk"), "/unknown"); + assert_eq!(normalize_route("/:colon"), "/unknown"); + } + + #[test] + fn escape_label_keeps_scrape_parseable() { + // A quote and a backslash must be escaped so the exposition stays + // parseable; a real newline must become the two-character escape. + assert_eq!(escape_label("a\"b\\c"), "a\\\"b\\\\c"); + assert_eq!(escape_label("line\nbreak"), "line\\nbreak"); } #[test] @@ -549,6 +609,37 @@ mod tests { assert_eq!(classify_surface("/unknown"), "unknown"); } + #[test] + fn otlp_bucket_counts_convert_cumulative_to_disjoint() { + let mut hist = Histogram::new(&[0.005, 0.01]); + // observe() bumps every bucket where value <= le, so buckets are + // cumulative: after these three, [3, 2, 0] with count 3. + hist.observe(0.004); + hist.observe(0.007); + hist.observe(0.02); + let counts = hist.otlp_bucket_counts(); + // Disjoint: 1 under 0.005, 1 between 0.005 and 0.01, 1 above. + assert_eq!(counts, vec![1, 1, 1]); + assert_eq!(counts.iter().sum::(), hist.count); + } + + #[test] + fn active_gauge_increment_and_decrement_are_idempotent() { + let m = HttpMetrics::default(); + let labels = ActiveLabels { + method: "GET".to_string(), + route: "/api/v1/runs".to_string(), + surface: "native".to_string(), + }; + m.inc_active(&labels); + m.inc_active(&labels); + m.dec_active(&labels); + // One still in flight; the gauge key carries no status_class, so a + // caller cannot increment under one key and decrement under another. + let g = m.active.read(); + assert_eq!(g.get(&labels), Some(&1)); + } + #[test] fn http_series_bounded() { let m = HttpMetrics::default(); @@ -566,3 +657,251 @@ mod tests { assert_eq!(m.series_count(), 1, "1000 distinct IDs must be 1 series"); } } + +// --------------------------------------------------------------------------- +// Structured collection for OTLP export +// --------------------------------------------------------------------------- + +/// One data point, already reduced to bounded attributes. +#[derive(Debug, Clone)] +pub enum MetricPoint { + /// Monotonic counter (OTLP `sum`, cumulative, `isMonotonic: true`). + Sum { + value: f64, + attributes: Vec<(String, String)>, + }, + /// Instantaneous value (OTLP `gauge`). + Gauge { + value: f64, + attributes: Vec<(String, String)>, + }, + /// Explicit-bucket histogram (OTLP `histogram`, cumulative). + /// + /// `bucket_counts` is one longer than `bounds`: OTLP requires the + /// implicit `+Inf` bucket to be present as the final entry. + Histogram { + count: u64, + sum: f64, + bounds: Vec, + bucket_counts: Vec, + attributes: Vec<(String, String)>, + }, +} + +#[derive(Debug, Clone)] +pub struct MetricFamily { + pub name: String, + pub unit: &'static str, + pub points: Vec, +} + +impl Histogram { + /// Convert the cumulative (Prometheus `le`) buckets to the disjoint + /// per-bucket counts OTLP requires, where every observation lands in + /// exactly one bucket and the counts sum to `count`. Emitting the + /// cumulative values as-if-disjoint would count each observation once + /// per bucket and blow the total past `count`. + fn otlp_bucket_counts(&self) -> Vec { + let mut deltas = Vec::with_capacity(self.buckets.len() + 1); + let mut previous = 0; + for (_, c) in &self.buckets { + deltas.push(c - previous); + previous = *c; + } + deltas.push(self.count - previous); + deltas + } + + fn bounds(&self) -> Vec { + self.buckets.iter().map(|(le, _)| *le).collect() + } +} + +impl HttpMetrics { + fn collect(&self, out: &mut Vec) { + let durations = self.durations.read(); + if !durations.is_empty() { + out.push(MetricFamily { + name: "http.server.request.duration".to_string(), + unit: "s", + points: durations + .iter() + .map(|(labels, hist)| MetricPoint::Histogram { + count: hist.count, + sum: hist.sum, + bounds: hist.bounds(), + bucket_counts: hist.otlp_bucket_counts(), + attributes: vec![ + ("http.request.method".to_string(), labels.method.clone()), + ("http.route".to_string(), labels.route.clone()), + ("preloop.surface".to_string(), labels.surface.clone()), + ( + "http.response.status_class".to_string(), + labels.status_class.clone(), + ), + ], + }) + .collect(), + }); + } + let active = self.active.read(); + if !active.is_empty() { + out.push(MetricFamily { + name: "http.server.active_requests".to_string(), + unit: "{request}", + points: active + .iter() + .map(|(labels, value)| MetricPoint::Gauge { + value: *value as f64, + attributes: vec![ + ("http.request.method".to_string(), labels.method.clone()), + ("http.route".to_string(), labels.route.clone()), + ("preloop.surface".to_string(), labels.surface.clone()), + ], + }) + .collect(), + }); + } + } +} + +impl StoreMetrics { + fn collect(&self, out: &mut Vec) { + let durations = self.durations.read(); + if !durations.is_empty() { + out.push(MetricFamily { + name: "preloop.store.operation.duration".to_string(), + unit: "s", + points: durations + .iter() + .map(|(labels, hist)| MetricPoint::Histogram { + count: hist.count, + sum: hist.sum, + bounds: hist.bounds(), + bucket_counts: hist.otlp_bucket_counts(), + attributes: vec![ + ("db.system".to_string(), labels.backend.clone()), + ("preloop.operation".to_string(), labels.operation.clone()), + ("preloop.outcome".to_string(), labels.outcome.clone()), + ], + }) + .collect(), + }); + } + let failures = self.consecutive_failures.read(); + if !failures.is_empty() { + out.push(MetricFamily { + name: "preloop.store.consecutive_failures".to_string(), + unit: "{failure}", + points: failures + .iter() + .map(|(backend, value)| MetricPoint::Gauge { + value: *value as f64, + attributes: vec![("db.system".to_string(), backend.clone())], + }) + .collect(), + }); + } + } +} + +impl LifecycleMetrics { + fn collect(&self, out: &mut Vec) { + let completed = self.job_completed.read(); + if !completed.is_empty() { + out.push(MetricFamily { + name: "preloop.job.completed".to_string(), + unit: "{job}", + points: completed + .iter() + .map(|(labels, value)| MetricPoint::Sum { + value: *value as f64, + attributes: vec![ + ("preloop.conclusion".to_string(), labels.conclusion.clone()), + ("preloop.reason".to_string(), labels.reason.clone()), + ], + }) + .collect(), + }); + } + let wait = self.queue_wait.read(); + if !wait.is_empty() { + out.push(MetricFamily { + name: "preloop.job.queue.wait".to_string(), + unit: "s", + points: wait + .iter() + .map(|(labels, hist)| MetricPoint::Histogram { + count: hist.count, + sum: hist.sum, + bounds: hist.bounds(), + bucket_counts: hist.otlp_bucket_counts(), + attributes: vec![("preloop.outcome".to_string(), labels.outcome.clone())], + }) + .collect(), + }); + } + let poll = self.broker_poll.read(); + if !poll.is_empty() { + out.push(MetricFamily { + name: "preloop.broker.poll".to_string(), + unit: "{poll}", + points: poll + .iter() + .map(|(labels, value)| MetricPoint::Sum { + value: *value as f64, + attributes: vec![("preloop.outcome".to_string(), labels.outcome.clone())], + }) + .collect(), + }); + } + let sessions = self.session_transition.read(); + if !sessions.is_empty() { + out.push(MetricFamily { + name: "preloop.runner.session.transition".to_string(), + unit: "{transition}", + points: sessions + .iter() + .map(|(labels, value)| MetricPoint::Sum { + value: *value as f64, + attributes: vec![ + ("preloop.operation".to_string(), labels.operation.clone()), + ("preloop.reason".to_string(), labels.reason.clone()), + ], + }) + .collect(), + }); + } + let concurrency = self.concurrency_decision.read(); + if !concurrency.is_empty() { + out.push(MetricFamily { + name: "preloop.concurrency.decision".to_string(), + unit: "{decision}", + points: concurrency + .iter() + .map(|(labels, value)| MetricPoint::Sum { + value: *value as f64, + attributes: vec![ + ("preloop.queue_mode".to_string(), labels.queue_mode.clone()), + ("preloop.action".to_string(), labels.action.clone()), + ], + }) + .collect(), + }); + } + } +} + +impl MetricsRegistry { + /// Snapshot every instrument as OTLP-ready families. + /// + /// Read-only: takes each sub-registry's read lock in turn and never holds + /// two at once, so a scrape cannot deadlock against a recording caller. + pub fn collect(&self) -> Vec { + let mut families = Vec::new(); + self.http.collect(&mut families); + self.store.collect(&mut families); + self.lifecycle.collect(&mut families); + families + } +} diff --git a/crates/preloop-observability/src/vm_telemetry.rs b/crates/preloop-observability/src/vm_telemetry.rs index 87b0b521..8398c23c 100644 --- a/crates/preloop-observability/src/vm_telemetry.rs +++ b/crates/preloop-observability/src/vm_telemetry.rs @@ -115,10 +115,12 @@ pub fn build_fleet_snapshot( storage_bytes, overlay_bytes, }, + // The host sampler is not reporting yet, so every measurement is + // explicitly absent rather than a fabricated zero. host_usage: VmHostUsage { - cpu_cores: 0.0, - memory_bytes: 0, - sparse_disk_allocated_bytes: 0, + cpu_cores: None, + memory_bytes: None, + sparse_disk_allocated_bytes: None, }, top_consumers: Vec::new(), } diff --git a/crates/preloop-runner-server/src/http_metrics.rs b/crates/preloop-runner-server/src/http_metrics.rs index 9dd9b28d..8b853897 100644 --- a/crates/preloop-runner-server/src/http_metrics.rs +++ b/crates/preloop-runner-server/src/http_metrics.rs @@ -26,7 +26,14 @@ pub async fn http_metrics_middleware( req: Request, next: Next, ) -> Response { - let method = req.method().to_string(); + // HTTP permits arbitrary extension-method tokens; a verbatim copy would + // give an unauthenticated caller an unbounded duration-series key. + let method = match req.method().as_str() { + "GET" | "POST" | "PUT" | "PATCH" | "DELETE" | "HEAD" | "OPTIONS" | "CONNECT" | "TRACE" => { + req.method().as_str().to_string() + } + _ => "other".to_string(), + }; // Prefer Axum's matched template; fallback to manual normalization for // the 1,000-IDs test and for unmatched routes. let raw_path = req.uri().path().to_string(); @@ -41,20 +48,44 @@ pub async fn http_metrics_middleware( // separately via `preloop.livelog.*`. let is_live_logs = surface == "live_logs"; - let labels = if !is_live_logs { - Some(preloop_observability::metrics::HttpLabels { + // The in-flight gauge is keyed without `status_class`: the status is + // unknown until the response completes, so any status in the key would + // let the increment (pre-response) and decrement (post-response) + // disagree and leak the gauge permanently. + let active_guard = (!is_live_logs).then(|| { + let labels = preloop_observability::metrics::ActiveLabels { method: method.clone(), route: route.clone(), surface: surface.clone(), - status_class: "2xx".to_string(), // placeholder, updated after response - }) - } else { - None - }; + }; + shared + .state + .observability + .metrics() + .http + .inc_active(&labels); + ActiveGuard { + shared: shared.clone(), + labels, + } + }); - if let Some(lbl) = &labels { - shared.state.observability.metrics().http.inc_active(lbl); - } + // Adopt an inbound W3C trace so a caller's trace continues through the + // control plane; otherwise start a root. Health and metrics probes are + // suppressed from trace export per the signal policy — they would swamp + // the trace store and tell an operator nothing. + let traced = shared.state.observability.tracing_enabled() && surface != "public"; + let span_context = traced.then(|| { + preloop_observability::export::SpanContext::from_traceparent( + req.headers() + .get("traceparent") + .and_then(|value| value.to_str().ok()), + ) + }); + // Only sample the clock when a span will actually be exported. + let span_start = span_context + .as_ref() + .map(|_| preloop_observability::export::now_nanos()); let start = Instant::now(); let res = next.run(req).await; @@ -63,41 +94,67 @@ pub async fn http_metrics_middleware( let status = res.status().as_u16(); let sc = status_class(status).to_string(); - if let Some(lbl) = labels { - let mut lbl = lbl; - lbl.status_class = sc.clone(); - // Record duration only for non-live_logs + if active_guard.is_some() { + shared.state.observability.metrics().http.observe_duration( + preloop_observability::metrics::HttpLabels { + method: method.clone(), + route: route.clone(), + surface: surface.clone(), + status_class: sc.clone(), + }, + elapsed, + ); + } + // Drop the guard (and the gauge slot) even when the inner future is + // cancelled or panics; the request is not in flight anymore either way. + drop(active_guard); + + if let (Some(context), Some(start_nanos)) = (span_context, span_start) { + // Attributes are allowlisted, never derived from the raw URI: the + // route is the matched template and the surface is a finite set. + let attributes = vec![ + ("http.request.method".to_string(), method.clone()), + ("http.route".to_string(), route.clone()), + ("preloop.surface".to_string(), surface), + ("http.response.status_code".to_string(), status.to_string()), + ]; shared .state .observability - .metrics() - .http - .observe_duration(lbl.clone(), elapsed); - shared.state.observability.metrics().http.dec_active(&lbl); + .export_span(preloop_observability::export::SpanRecord { + context, + name: format!("{method} {route}"), + start_nanos, + end_nanos: preloop_observability::export::now_nanos(), + // Only 5xx is the server's fault; a 4xx is the caller's and + // marking it Error would make every unauthenticated probe + // look like an outage. + status: if status >= 500 { + preloop_observability::export::SpanStatus::Error + } else { + preloop_observability::export::SpanStatus::Unset + }, + attributes, + }); } - // Safe span: method + route template + surface + status, no headers/body/query. - // Use `tracing::info_span!` so it appears in logs when RUST_LOG includes it, - // but filtered at DEBUG by default (poll/renew are DEBUG). - let span = tracing::info_span!( - "http.request", - http.method = %method, - http.route = %route, - http.surface = %surface, - http.status_code = status, - http.status_class = %sc, - otel.kind = "server", - ); - // Attach span to response for trace correlation; the span itself is not - // entered for the handler thread beyond this point (no await while held). + res +} - // Also emit a counter for broker poll outcomes — the control plane's - // poll is a long-poll that returns job|empty|cancel|error. That is - // already counted via the HTTP histogram, but the plan also wants - // `preloop.broker.poll{outcome}` to distinguish empty vs error. - // For now we just log at DEBUG; the dedicated broker counter is wired - // in the broker module itself (Step 4 lifecycle). - let _ = span; +/// Releases the active-request gauge on drop, so cancellation, panics, and +/// client disconnects during a long poll cannot leak the series. +struct ActiveGuard { + shared: Arc, + labels: preloop_observability::metrics::ActiveLabels, +} - res +impl Drop for ActiveGuard { + fn drop(&mut self) { + self.shared + .state + .observability + .metrics() + .http + .dec_active(&self.labels); + } } diff --git a/crates/preloop-runner-server/src/main.rs b/crates/preloop-runner-server/src/main.rs index 44c46253..f74efe31 100644 --- a/crates/preloop-runner-server/src/main.rs +++ b/crates/preloop-runner-server/src/main.rs @@ -79,12 +79,15 @@ async fn main() -> anyhow::Result<()> { // like the CLI, instead of falling silent when unset. `PRELOOP_LOG_FORMAT` // controls pretty/json/auto. The `Observability` handle will be cloned // into `AppState`; for now it is held for the life of `main`. - let obs_config = preloop_observability::ObservabilityConfig::from_env(); + let obs_config = preloop_observability::ObservabilityConfig::from_env() + .with_service_version(env!("CARGO_PKG_VERSION")); let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); preloop_observability::ObservabilityRuntime::install_fmt_subscriber(observability.config()); - let _observability = observability; - let _observability_runtime = observability_runtime; + // The handle reaches `ServerConfig` below; holding it here alone would + // leave the server on the no-op handle installed by `AppState::new`, so + // nothing would ever export. + let mut observability_runtime = observability_runtime; let cli = Cli::parse(); match cli.command { @@ -131,7 +134,7 @@ async fn main() -> anyhow::Result<()> { pool_preparing: None, listen, pool_status: None, - observability: None, + observability: Some(observability.clone()), systemd_socket_activation: false, unix_socket, state_dir, @@ -156,5 +159,8 @@ async fn main() -> anyhow::Result<()> { .await?; } } + // Bounded 2s flush of buffered telemetry before exit; a clean shutdown + // must not drop the last flush window's records. + observability_runtime.shutdown().await; Ok(()) } diff --git a/crates/preloop-runner-server/src/models.rs b/crates/preloop-runner-server/src/models.rs index bcd59a75..951d234e 100644 --- a/crates/preloop-runner-server/src/models.rs +++ b/crates/preloop-runner-server/src/models.rs @@ -188,6 +188,12 @@ pub(crate) struct QueuedJob { pub(crate) run_id: RunId, pub(crate) job_id: JobId, pub(crate) base_id: String, + /// Unix nanoseconds when the job entered the ready queue, used to + /// measure true queue latency at claim time. `0` means unknown (a + /// snapshot persisted before this field existed); such jobs are not + /// recorded, so a restart never fabricates a latency. + #[serde(default)] + pub(crate) enqueued_at_unix_nanos: i64, pub(crate) needs: Vec, pub(crate) if_condition: Option, pub(crate) condition_context: preloop_gha_expressions::Context, @@ -313,3 +319,12 @@ pub(crate) struct ArtifactV2Entry { /// Upload token used to find the assembled blob on disk. pub(crate) blob_token: String, } + +/// Unix nanoseconds now, for queue-latency bookkeeping. `i64` keeps the +/// field serde-friendly (it travels in persisted job snapshots). +pub(crate) fn now_unix_nanos() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as i64) + .unwrap_or(0) +} diff --git a/crates/preloop-runner-server/src/runs.rs b/crates/preloop-runner-server/src/runs.rs index 85eb830e..0616b9bd 100644 --- a/crates/preloop-runner-server/src/runs.rs +++ b/crates/preloop-runner-server/src/runs.rs @@ -1,6 +1,11 @@ use super::*; use std::collections::BTreeSet; +/// A heartbeat or sampler snapshot older than this is stale: three sampler +/// intervals of 5s. Single source so `/readyz` and `/api/v1/status` cannot +/// disagree when the interval changes. +pub(crate) const STALENESS_THRESHOLD: Duration = Duration::from_secs(15); + pub(crate) async fn healthz(State(shared): State>) -> impl IntoResponse { let shutdown = shared.shutdown.is_cancelled(); let body = json!({ @@ -20,23 +25,21 @@ pub(crate) async fn readyz(State(shared): State>) -> impl IntoR let body = json!({ "ready": false, "reason": "shutting_down" }); return (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response(); } - // Check critical heartbeats freshness (>15s stale is 3 intervals of 5s sampler) if let Some(stale) = shared .state .observability .heartbeat() - .any_critical_stale(Duration::from_secs(15)) + .any_critical_stale(STALENESS_THRESHOLD) { let body = json!({ "ready": false, "reason": format!("task_stale:{}", stale) }); return (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response(); } - // Also check snapshot age (>15s stale sampler) let age_secs = { let snap = shared.state.status_snapshot.read(); let now = chrono::Utc::now(); (now - snap.observed_at).num_milliseconds() as f64 / 1000.0 }; - if age_secs > 15.0 { + if age_secs > STALENESS_THRESHOLD.as_secs_f64() { let body = json!({ "ready": false, "reason": "state_sampler_stale" }); return (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response(); } @@ -68,9 +71,7 @@ pub(crate) async fn status(State(shared): State>) -> impl IntoR name: t.name.to_string(), critical: t.critical == preloop_observability::Criticality::Critical, heartbeat_age_seconds: t.heartbeat_age.as_secs_f64(), - state: if t.exited { - "exited".to_string() - } else if t.heartbeat_age > Duration::from_secs(15) { + state: if t.heartbeat_age > STALENESS_THRESHOLD { "stale".to_string() } else { "running".to_string() @@ -1126,6 +1127,7 @@ pub(crate) async fn submit_run_inner( run_id, job_id: job.id.clone(), base_id: job.base_id.clone(), + enqueued_at_unix_nanos: crate::models::now_unix_nanos(), needs: job.needs.clone(), if_condition: job.if_condition.clone(), condition_context: pb.condition_context, diff --git a/crates/preloop-runner-server/src/runtime_scheduling.rs b/crates/preloop-runner-server/src/runtime_scheduling.rs index 33ae0c61..e629b2bf 100644 --- a/crates/preloop-runner-server/src/runtime_scheduling.rs +++ b/crates/preloop-runner-server/src/runtime_scheduling.rs @@ -2426,6 +2426,7 @@ fn register_expanded_jobs( run_id, job_id: plan.id.clone(), base_id: plan.base_id.clone(), + enqueued_at_unix_nanos: crate::models::now_unix_nanos(), needs: plan.needs.clone(), if_condition: plan.if_condition.clone(), condition_context, @@ -2890,6 +2891,7 @@ mod assignment_tests { run_id: RunId::new(), job_id: JobId(job_id.to_owned()), base_id: job_id.to_owned(), + enqueued_at_unix_nanos: 0, needs: Vec::new(), if_condition: None, condition_context: preloop_gha_expressions::Context::default(), diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs index 1cb3ae6e..b62e2e5d 100644 --- a/crates/preloop-runner-server/src/state.rs +++ b/crates/preloop-runner-server/src/state.rs @@ -561,6 +561,73 @@ pub(crate) enum JobSetAdmissionResult { Blocked, } +/// Bounded conclusion label for a terminal execution status. +fn execution_conclusion(status: preloop_gha_protocol::ExecutionStatus) -> &'static str { + use preloop_gha_protocol::ExecutionStatus as S; + match status { + S::Success => "success", + S::Failure => "failure", + S::Cancelled => "cancelled", + S::Skipped => "skipped", + // Non-terminal statuses never reach here; the guard filters them. + _ => "unrecognized", + } +} + +/// Classify a termination reason into a bounded code. +/// +/// The control plane's `reason` is not a code — several paths build a prose +/// sentence that interpolates the job's `runs-on` labels (see the starvation +/// sweep in `bootstrap.rs`). Those values are user-controlled, so the raw +/// string must never reach a metric label: it would both explode cardinality +/// and export workflow content. Classify by the stable prefix each path +/// writes, and fall back to `unrecognized` rather than passing prose through. +/// +/// The full message is still available on the structured log record; only the +/// metric dimension is bounded. +fn bounded_termination_reason(value: &str) -> &'static str { + // Exact codes first — these come from `concurrency::*_reason()`. + match value { + "concurrency_pending" => return "concurrency_pending", + "concurrency_cancelled" => return "concurrency_cancelled", + "timeout" => return "timeout", + "no_runner" => return "no_runner", + "lease_expired" => return "lease_expired", + "deaf_runner" => return "deaf_runner", + "startup_orphan" => return "startup_orphan", + _ => {} + } + // Prose paths — match on the invariant phrase, never the whole string, so + // an interpolated label or platform cannot change the classification. + // + // Two distinct never-claimable conditions, and conflating them would hide + // the difference between "wait or add capacity" and "this will never work + // until you register that platform": + // - the starvation sweep, which fires after a grace window; + // - the external-host check, where the server has no runner of that + // platform class at all (`no {platform} runner is registered with + // this server, so `runs-on: …` cannot be scheduled`). + if value.contains("runner is registered with this server") { + return "no_platform_runner"; + } + if value.starts_with("no runner is registered for") { + return "no_runner"; + } + if value.starts_with("job exceeded its timeout") + || value.starts_with("timed out") + || value.contains("timeout-minutes") + { + return "timeout"; + } + if value.starts_with("runner stopped polling") || value.contains("deaf") { + return "deaf_runner"; + } + if value.contains("lease expired") { + return "lease_expired"; + } + "unrecognized" +} + impl AppState { pub async fn new(state_dir: PathBuf) -> anyhow::Result { let config_path = crate::config::config_path(); @@ -841,6 +908,16 @@ impl AppState { _ => None, }; let has_run_projection = run_id.is_some(); + if let NdjsonEvent::RunAccepted { queued_jobs, .. } = &event { + self.observability.export_log( + "INFO", + "run.accepted", + vec![ + ("event.name".to_string(), "run.accepted".to_string()), + ("queued_jobs".to_string(), queued_jobs.to_string()), + ], + ); + } // Record job terminal transitions exactly once. Guard with is_terminal // so we don't double-count non-terminal status updates. The event // itself is the proof of old→terminal movement, so we record here @@ -848,47 +925,62 @@ impl AppState { // duplicate `store_run_event` emits. match &event { NdjsonEvent::JobStatus { status, reason, .. } if status.is_terminal() => { - let conclusion = match status { - preloop_gha_protocol::ExecutionStatus::Success => "success", - preloop_gha_protocol::ExecutionStatus::Failure => "failure", - preloop_gha_protocol::ExecutionStatus::Cancelled => "cancelled", - preloop_gha_protocol::ExecutionStatus::Skipped => "skipped", - _ => "unknown", - }; - let reason_str = reason.as_deref().unwrap_or("unknown"); - // Bound the reason to the finite set the plan allows; unknown - // reasons are mapped to "unknown" so they don't create new series. - let bounded_reason = match reason_str { - "timeout" - | "no_runner" - | "lease_expired" - | "deaf_runner" - | "startup_orphan" - | "concurrency_cancelled" - | "concurrency_pending" - | "success" - | "failure" - | "cancelled" - | "skipped" => reason_str, - _ => "unknown", + let conclusion = execution_conclusion(*status); + // `reason: None` is the common case (most terminal transitions + // carry none) and means "no reason supplied" — not + // "unrecognized". Only a value outside the emitted set is + // `unrecognized`, which keeps the label bounded without + // mislabelling the majority. + let bounded_reason = match reason.as_deref() { + None => "unspecified", + Some(value) => bounded_termination_reason(value), }; self.observability .metrics() .lifecycle .record_job_completed(conclusion, bounded_reason); + self.observability.export_log( + if *status == preloop_gha_protocol::ExecutionStatus::Success { + "INFO" + } else { + "WARN" + }, + // A terminal JobStatus is a status transition, not the + // separate JobCompleted event; naming both `job.completed` + // conflated two distinct records in the log stream. + "job.status.terminal", + { + let mut attributes = vec![ + ("event.name".to_string(), "job.status.terminal".to_string()), + ("conclusion".to_string(), conclusion.to_string()), + ("reason".to_string(), bounded_reason.to_string()), + ]; + // The bounded code is the metric dimension; the prose + // is what an operator actually needs to act. Logs may + // carry it (they are not a label space), and without + // it an `unrecognized` classification is a dead end — + // you cannot tell which path produced it. + if let Some(detail) = reason.as_deref() { + attributes.push(("reason.detail".to_string(), detail.to_string())); + } + attributes + }, + ); } NdjsonEvent::JobCompleted { status, .. } if status.is_terminal() => { - let conclusion = match status { - preloop_gha_protocol::ExecutionStatus::Success => "success", - preloop_gha_protocol::ExecutionStatus::Failure => "failure", - preloop_gha_protocol::ExecutionStatus::Cancelled => "cancelled", - preloop_gha_protocol::ExecutionStatus::Skipped => "skipped", - _ => "unknown", - }; + let conclusion = execution_conclusion(*status); self.observability .metrics() .lifecycle .record_job_completed(conclusion, "completed"); + self.observability.export_log( + "INFO", + "job.completed", + vec![ + ("event.name".to_string(), "job.completed".to_string()), + ("conclusion".to_string(), conclusion.to_string()), + ], + ); } _ => {} } @@ -1338,3 +1430,83 @@ mod tests { ); } } + +#[cfg(test)] +mod termination_reason_tests { + use super::bounded_termination_reason; + + #[test] + fn exact_codes_pass_through() { + assert_eq!( + bounded_termination_reason("concurrency_cancelled"), + "concurrency_cancelled" + ); + assert_eq!(bounded_termination_reason("timeout"), "timeout"); + } + + #[test] + fn starvation_prose_classifies_to_no_runner() { + // The starvation sweep builds this sentence with the job's runs-on + // labels interpolated. It must classify, not pass through. + let prose = "no runner is registered for `runs-on: self-hosted, Linux, ARM64` and none \ + appeared within 120s, so the job cannot be scheduled"; + assert_eq!(bounded_termination_reason(prose), "no_runner"); + } + + #[test] + fn user_controlled_labels_never_become_the_label() { + // A hostile or merely unusual `runs-on` must not reach the metric. + let prose = "no runner is registered for `runs-on: attacker-controlled-\u{1F4A5}-label` \ + and none appeared within 120s, so the job cannot be scheduled"; + let bounded = bounded_termination_reason(prose); + assert_eq!(bounded, "no_runner"); + assert!(!bounded.contains("attacker")); + } + + #[test] + fn external_host_prose_is_its_own_code() { + // `{platform}` is interpolated, so match the invariant phrase. + for platform in ["windows", "macos", "freebsd-13"] { + let prose = format!( + "no {platform} runner is registered with this server, so \ + `runs-on: {platform}-latest` cannot be scheduled" + ); + assert_eq!( + bounded_termination_reason(&prose), + "no_platform_runner", + "{platform} must classify distinctly from the starvation sweep" + ); + } + } + + #[test] + fn platform_and_starvation_do_not_collide() { + let starved = "no runner is registered for `runs-on: self-hosted, Linux, ARM64` and none \ + appeared within 120s, so the job cannot be scheduled"; + let platform = "no windows runner is registered with this server, so \ + `runs-on: windows-latest` cannot be scheduled"; + assert_eq!(bounded_termination_reason(starved), "no_runner"); + assert_eq!(bounded_termination_reason(platform), "no_platform_runner"); + } + + #[test] + fn unknown_prose_is_bounded_not_passed_through() { + let bounded = bounded_termination_reason("something entirely new happened with id-99999"); + assert_eq!(bounded, "unrecognized"); + assert!(!bounded.contains("99999")); + } + + #[test] + fn classification_is_a_finite_set() { + // Drive 1,000 distinct prose strings; the label set must stay bounded. + let mut seen = std::collections::BTreeSet::new(); + for i in 0..1000 { + let prose = format!( + "no runner is registered for `runs-on: label-{i}` and none appeared within 120s" + ); + seen.insert(bounded_termination_reason(&prose)); + seen.insert(bounded_termination_reason(&format!("novel reason {i}"))); + } + assert_eq!(seen.len(), 2, "expected exactly no_runner + unrecognized"); + } +} diff --git a/crates/preloop-runner/src/main.rs b/crates/preloop-runner/src/main.rs index a7c5a687..10740709 100644 --- a/crates/preloop-runner/src/main.rs +++ b/crates/preloop-runner/src/main.rs @@ -16,7 +16,8 @@ async fn main() -> Result<()> { // Runner gets structured local logging only — never OTLP export by default. // `PRELOOP_LOG_FORMAT` still controls pretty/json/auto for consistency. - let obs_config = preloop_observability::ObservabilityConfig::from_env(); + let obs_config = preloop_observability::ObservabilityConfig::from_env() + .with_service_version(env!("CARGO_PKG_VERSION")); let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); preloop_observability::ObservabilityRuntime::install_fmt_subscriber(observability.config());