From 993a9e5ce69b4b580917069f12f0be074711e266 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 18:26:30 -0400 Subject: [PATCH 01/11] feat(contrib): add pinned single-node OpenObserve reference profile Entire-Checkpoint: 01M0GMGVPS35504XE9KQKPGV14 --- contrib/openobserve/compose.yml | 43 +++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 contrib/openobserve/compose.yml diff --git a/contrib/openobserve/compose.yml b/contrib/openobserve/compose.yml new file mode 100644 index 00000000..2682ba4a --- /dev/null +++ b/contrib/openobserve/compose.yml @@ -0,0 +1,43 @@ +# 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} + ZO_ROOT_USER_PASSWORD: ${ZO_ROOT_USER_PASSWORD:-ChangeMe.Preloop1} + 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: From cba90a3712398eeee5d37a7f72af18d4ba9e1970 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 18:40:43 -0400 Subject: [PATCH 02/11] feat(observability): add bounded OTLP/HTTP JSON exporter and wire lifecycle events --- crates/preloop-observability/src/export.rs | 312 +++++++++++++++++++++ crates/preloop-observability/src/lib.rs | 46 +++ crates/preloop-runner-server/src/main.rs | 6 +- crates/preloop-runner-server/src/state.rs | 31 ++ 4 files changed, 393 insertions(+), 2 deletions(-) create mode 100644 crates/preloop-observability/src/export.rs diff --git a/crates/preloop-observability/src/export.rs b/crates/preloop-observability/src/export.rs new file mode 100644 index 00000000..69d4bb63 --- /dev/null +++ b/crates/preloop-observability/src/export.rs @@ -0,0 +1,312 @@ +//! 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) +} + +fn now_nanos() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) +} + +/// One log record queued for export. +#[derive(Debug, Clone)] +pub struct LogRecord { + pub severity: &'static str, + pub body: String, + pub attributes: Vec<(String, String)>, +} + +/// Handle used by the rest of the process to enqueue telemetry. +#[derive(Debug, Clone)] +pub struct Exporter { + tx: mpsc::Sender, + health: Arc, +} + +impl Exporter { + /// Enqueue a log record. Never blocks; drops on a full queue. + pub fn log(&self, record: LogRecord) { + if self.tx.try_send(record).is_err() { + self.health.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + pub fn health(&self) -> &Arc { + &self.health + } +} + +/// Spawn the export worker. Returns `None` when no endpoint is configured, +/// so the absent-endpoint path opens no socket at all. +pub fn spawn( + endpoint: Option<&str>, + headers: Option<&str>, + service_name: &str, + instance_id: &str, +) -> Option<(Exporter, Arc)> { + let endpoint = endpoint?.trim_end_matches('/').to_string(); + let header_pairs = parse_headers(headers); + let service_name = service_name.to_string(); + let instance_id = instance_id.to_string(); + let health = Arc::new(ExportHealth::default()); + let (tx, mut rx) = mpsc::channel::(QUEUE_CAPACITY); + + let worker_health = health.clone(); + 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; + } + }; + let logs_url = format!("{endpoint}/v1/logs"); + let mut buffer: Vec = Vec::with_capacity(BATCH_MAX); + let mut ticker = tokio::time::interval(FLUSH_INTERVAL); + loop { + tokio::select! { + maybe = rx.recv() => { + match maybe { + Some(record) => { + buffer.push(record); + if buffer.len() >= BATCH_MAX { + flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &mut buffer, &worker_health).await; + } + } + None => { + flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &mut buffer, &worker_health).await; + break; + } + } + } + _ = ticker.tick() => { + flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &mut buffer, &worker_health).await; + } + } + } + }); + + Some(( + Exporter { + tx, + health: health.clone(), + }, + health, + )) +} + +async fn flush( + client: &reqwest::Client, + url: &str, + headers: &[(String, String)], + service_name: &str, + instance_id: &str, + buffer: &mut Vec, + health: &Arc, +) { + if buffer.is_empty() { + return; + } + let batch = std::mem::take(buffer); + let count = batch.len() as u64; + let payload = encode_logs(&batch, service_name, instance_id); + 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() => health.record_success(count), + Ok(response) => { + // Status class only — never the body, which can echo credentials. + tracing::warn!( + failure = "http_status", + 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", "telemetry export failed"); + health.record_failure(); + } + } +} + +fn attr(key: &str, value: &str) -> Value { + json!({"key": key, "value": {"stringValue": value}}) +} + +fn severity_number(severity: &str) -> u8 { + match severity { + "TRACE" => 1, + "DEBUG" => 5, + "INFO" => 9, + "WARN" => 13, + "ERROR" => 17, + _ => 9, + } +} + +/// Encode a batch into the OTLP/HTTP JSON `ExportLogsServiceRequest` shape. +pub fn encode_logs(batch: &[LogRecord], service_name: &str, instance_id: &str) -> Value { + let ts = now_nanos().to_string(); + let records: Vec = batch + .iter() + .map(|record| { + let attributes: Vec = + record.attributes.iter().map(|(k, v)| attr(k, v)).collect(); + json!({ + "timeUnixNano": ts, + "observedTimeUnixNano": ts, + "severityNumber": severity_number(record.severity), + "severityText": record.severity, + "body": {"stringValue": record.body}, + "attributes": attributes, + }) + }) + .collect(); + json!({ + "resourceLogs": [{ + "resource": { + "attributes": [ + attr("service.name", service_name), + attr("service.instance.id", instance_id), + attr("service.version", env!("CARGO_PKG_VERSION")), + ] + }, + "scopeLogs": [{ + "scope": {"name": "preloop-observability"}, + "logRecords": records, + }] + }] + }) +} + +/// Parse `OTEL_EXPORTER_OTLP_HEADERS` (`k1=v1,k2=v2`). +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() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn absent_endpoint_spawns_nothing() { + assert!(spawn(None, None, "preloop", "abc").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())], + }]; + let payload = encode_logs(&batch, "preloop", "inst-1"); + 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"); + } +} diff --git a/crates/preloop-observability/src/lib.rs b/crates/preloop-observability/src/lib.rs index 3da05d42..43cb5ed3 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; @@ -140,6 +142,11 @@ impl ObservabilityConfig { self.otel_headers.is_some() } + /// Raw headers for transport construction. Never logged or in `Debug`. + pub fn otel_headers_raw(&self) -> Option<&str> { + self.otel_headers.as_deref() + } + /// Sanitized endpoint for `Debug`/errors: strips userinfo and query. fn sanitized_endpoint(&self) -> Option { self.otel_endpoint.as_ref().map(|raw| { @@ -383,6 +390,8 @@ struct Inner { heartbeat: TaskHeartbeat, limits: LimitRegistry, metrics: Arc, + vm_registry: Arc, + exporter: Option, is_noop: bool, } @@ -410,6 +419,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,12 +429,22 @@ 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. + let exporter = export::spawn( + config.otel_endpoint_raw(), + config.otel_headers_raw(), + &config.service_name, + &config.instance_id, + ) + .map(|(exporter, _health)| exporter); let handle = Self { inner: Arc::new(Inner { config: Arc::new(config), heartbeat: TaskHeartbeat::default(), limits: LimitRegistry::default(), metrics: Arc::new(metrics::MetricsRegistry::default()), + vm_registry: Arc::new(vm_telemetry::VmTelemetryRegistry::default()), + exporter, is_noop, }), }; @@ -459,6 +480,31 @@ 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)>, + ) { + if let Some(exporter) = &self.inner.exporter { + exporter.log(export::LogRecord { + severity, + body: body.into(), + attributes, + }); + } + } + + /// 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 } diff --git a/crates/preloop-runner-server/src/main.rs b/crates/preloop-runner-server/src/main.rs index 44c46253..7542a45e 100644 --- a/crates/preloop-runner-server/src/main.rs +++ b/crates/preloop-runner-server/src/main.rs @@ -83,7 +83,9 @@ async fn main() -> anyhow::Result<()> { let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); preloop_observability::ObservabilityRuntime::install_fmt_subscriber(observability.config()); - let _observability = observability; + // The handle must reach `ServerConfig`; holding it here alone would leave + // the server on the no-op handle installed by `AppState::new`, so nothing + // would ever export. let _observability_runtime = observability_runtime; let cli = Cli::parse(); @@ -131,7 +133,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, diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs index 1cb3ae6e..00d0a0ee 100644 --- a/crates/preloop-runner-server/src/state.rs +++ b/crates/preloop-runner-server/src/state.rs @@ -841,6 +841,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 @@ -876,6 +886,19 @@ impl AppState { .metrics() .lifecycle .record_job_completed(conclusion, bounded_reason); + self.observability.export_log( + if *status == preloop_gha_protocol::ExecutionStatus::Success { + "INFO" + } else { + "WARN" + }, + "job.completed", + vec![ + ("event.name".to_string(), "job.completed".to_string()), + ("conclusion".to_string(), conclusion.to_string()), + ("reason".to_string(), bounded_reason.to_string()), + ], + ); } NdjsonEvent::JobCompleted { status, .. } if status.is_terminal() => { let conclusion = match status { @@ -889,6 +912,14 @@ impl AppState { .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()), + ], + ); } _ => {} } From e0d00697c8f653cdeee4894f7f7c0052ac2c7fa5 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 18:53:28 -0400 Subject: [PATCH 03/11] fix(server): classify termination reasons and report the host binary version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects surfaced by reading an exported record. The reason label was wrong. The control plane's `reason` is not a code: the starvation sweep builds a prose sentence that interpolates the job's `runs-on` labels. Passing it through would explode metric cardinality and export workflow content, so the previous code bounded it — but it bounded every value, including the common `reason: None`, to "unknown". That labelled "no reason supplied" and "unrecognized string" identically and made a legitimately failed job unexplainable. Now `None` is `unspecified`, exact codes pass through, and prose is classified on its stable leading phrase, so the starvation sentence becomes `no_runner`. The full message stays on the log record; only the metric dimension is bounded. The event name conflated two records. A terminal `JobStatus` is a status transition, not the separate `JobCompleted` event, but both exported `body: "job.completed"`. The transition is now `job.status.terminal`. `service.version` reported the observability crate's version, which is meaningless to an operator. `ObservabilityConfig::with_service_version` now takes the host binary's version and all three binaries pass it. Tests: five cases for the classifier, including a hostile interpolated `runs-on` and a 1,000-string drive asserting the label set stays at two. Entire-Checkpoint: 01M0GP28K4J52DGX9CMDJVCYF5 --- crates/preloop-cli/src/main.rs | 3 +- crates/preloop-observability/src/export.rs | 29 ++-- crates/preloop-observability/src/lib.rs | 14 ++ crates/preloop-runner-server/src/main.rs | 3 +- crates/preloop-runner-server/src/state.rs | 157 ++++++++++++++++----- crates/preloop-runner/src/main.rs | 3 +- 6 files changed, 166 insertions(+), 43 deletions(-) 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 index 69d4bb63..67d3b91b 100644 --- a/crates/preloop-observability/src/export.rs +++ b/crates/preloop-observability/src/export.rs @@ -103,11 +103,13 @@ pub fn spawn( headers: Option<&str>, service_name: &str, instance_id: &str, + service_version: &str, ) -> Option<(Exporter, Arc)> { let endpoint = endpoint?.trim_end_matches('/').to_string(); let header_pairs = parse_headers(headers); let service_name = service_name.to_string(); let instance_id = instance_id.to_string(); + let service_version = service_version.to_string(); let health = Arc::new(ExportHealth::default()); let (tx, mut rx) = mpsc::channel::(QUEUE_CAPACITY); @@ -138,17 +140,17 @@ pub fn spawn( Some(record) => { buffer.push(record); if buffer.len() >= BATCH_MAX { - flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &mut buffer, &worker_health).await; + flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &service_version, &mut buffer, &worker_health).await; } } None => { - flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &mut buffer, &worker_health).await; + flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &service_version, &mut buffer, &worker_health).await; break; } } } _ = ticker.tick() => { - flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &mut buffer, &worker_health).await; + flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &service_version, &mut buffer, &worker_health).await; } } } @@ -169,6 +171,7 @@ async fn flush( headers: &[(String, String)], service_name: &str, instance_id: &str, + service_version: &str, buffer: &mut Vec, health: &Arc, ) { @@ -177,7 +180,7 @@ async fn flush( } let batch = std::mem::take(buffer); let count = batch.len() as u64; - let payload = encode_logs(&batch, service_name, instance_id); + let payload = encode_logs(&batch, service_name, instance_id, service_version); let mut req = client.post(url).json(&payload); for (name, value) in headers { req = req.header(name.as_str(), value.as_str()); @@ -218,7 +221,12 @@ fn severity_number(severity: &str) -> u8 { } /// Encode a batch into the OTLP/HTTP JSON `ExportLogsServiceRequest` shape. -pub fn encode_logs(batch: &[LogRecord], service_name: &str, instance_id: &str) -> Value { +pub fn encode_logs( + batch: &[LogRecord], + service_name: &str, + instance_id: &str, + service_version: &str, +) -> Value { let ts = now_nanos().to_string(); let records: Vec = batch .iter() @@ -241,7 +249,7 @@ pub fn encode_logs(batch: &[LogRecord], service_name: &str, instance_id: &str) - "attributes": [ attr("service.name", service_name), attr("service.instance.id", instance_id), - attr("service.version", env!("CARGO_PKG_VERSION")), + attr("service.version", service_version), ] }, "scopeLogs": [{ @@ -277,7 +285,7 @@ mod tests { #[test] fn absent_endpoint_spawns_nothing() { - assert!(spawn(None, None, "preloop", "abc").is_none()); + assert!(spawn(None, None, "preloop", "abc", "9.9.9").is_none()); } #[test] @@ -300,7 +308,7 @@ mod tests { body: "pool provisioning failed".to_string(), attributes: vec![("event.name".to_string(), "pool.provision".to_string())], }]; - let payload = encode_logs(&batch, "preloop", "inst-1"); + 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); @@ -308,5 +316,10 @@ mod tests { 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" + ); } } diff --git a/crates/preloop-observability/src/lib.rs b/crates/preloop-observability/src/lib.rs index 43cb5ed3..a82e9c14 100644 --- a/crates/preloop-observability/src/lib.rs +++ b/crates/preloop-observability/src/lib.rs @@ -72,6 +72,9 @@ 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 @@ -125,6 +128,7 @@ impl ObservabilityConfig { log_format, rust_log, service_name, + service_version: env!("CARGO_PKG_VERSION").to_string(), instance_id, otel_endpoint, otel_headers, @@ -142,6 +146,13 @@ impl ObservabilityConfig { self.otel_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 + } + /// Raw headers for transport construction. Never logged or in `Debug`. pub fn otel_headers_raw(&self) -> Option<&str> { self.otel_headers.as_deref() @@ -171,6 +182,7 @@ 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", @@ -408,6 +420,7 @@ 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, @@ -435,6 +448,7 @@ impl Observability { config.otel_headers_raw(), &config.service_name, &config.instance_id, + &config.service_version, ) .map(|(exporter, _health)| exporter); let handle = Self { diff --git a/crates/preloop-runner-server/src/main.rs b/crates/preloop-runner-server/src/main.rs index 7542a45e..afd6b67f 100644 --- a/crates/preloop-runner-server/src/main.rs +++ b/crates/preloop-runner-server/src/main.rs @@ -79,7 +79,8 @@ 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()); diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs index 00d0a0ee..a0b10bd1 100644 --- a/crates/preloop-runner-server/src/state.rs +++ b/crates/preloop-runner-server/src/state.rs @@ -561,6 +561,62 @@ 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 stable leading phrase, never the whole + // string, so an interpolated label cannot change the classification. + 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(); @@ -858,29 +914,15 @@ 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() @@ -892,22 +934,19 @@ impl AppState { } else { "WARN" }, - "job.completed", + // 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", vec![ - ("event.name".to_string(), "job.completed".to_string()), + ("event.name".to_string(), "job.status.terminal".to_string()), ("conclusion".to_string(), conclusion.to_string()), ("reason".to_string(), bounded_reason.to_string()), ], ); } 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 @@ -1369,3 +1408,57 @@ 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 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()); From 02a539cc9c2d94dcd80b81bb5e378af01bef52a9 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 19:02:12 -0400 Subject: [PATCH 04/11] feat(observability): export metrics and traces over OTLP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Logs were the only signal reaching a backend; metrics were Prometheus-pull only and traces did not exist at all — the HTTP middleware built a span and dropped it. Metrics. `MetricsRegistry::collect` snapshots every instrument as OTLP-ready families, so export scrapes the same instruments `/metrics` renders rather than maintaining a second set. Counters become cumulative monotonic sums, gauges become gauges, and the internal histogram becomes an explicit-bucket histogram with the implicit `+Inf` bucket OTLP requires. Every cumulative point carries the process start as `startTimeUnixNano`, without which a backend reads a restart as a counter reset. Traces. Add real W3C Trace Context: an inbound `traceparent` is adopted so a caller's trace continues through the control plane, a malformed one starts a new root rather than failing the request, and all-zero ids are rejected per the spec. Spans carry the matched route template and finite surface, never the raw URI. Only 5xx sets `Error` — marking 4xx would make every unauthenticated probe look like an outage. Health and metrics probes are suppressed from trace export; they would swamp the store and explain nothing. Log records now carry `traceId`/`spanId` as OTLP fields, not attributes, so a backend can pivot log to trace. One worker drains logs and spans from a shared bounded queue and scrapes metrics on the same tick, so all three share one client, one batching cadence, and one fail-open path. Verified against a pinned single-node OpenObserve: traces, logs and metrics streams all populated; an injected traceparent arrived as trace_id=4bf92f3577b34da6a3ce929d0e0e4736 with a fresh span id; histogram points carry AGGREGATION_TEMPORALITY_CUMULATIVE with bounded attributes (http_route, preloop_surface) and service_version 0.2.0; public probes absent from spans. 24 crate tests including traceparent adoption, malformed rejection, id uniqueness, OTLP shapes, and the +Inf bucket invariant. Entire-Checkpoint: 01M0GPJ7JVYWNJMPJYVE5QW4P6 --- crates/preloop-observability/src/export.rs | 623 ++++++++++++++++-- crates/preloop-observability/src/lib.rs | 33 +- crates/preloop-observability/src/metrics.rs | 239 +++++++ .../preloop-runner-server/src/http_metrics.rs | 71 +- 4 files changed, 894 insertions(+), 72 deletions(-) diff --git a/crates/preloop-observability/src/export.rs b/crates/preloop-observability/src/export.rs index 67d3b91b..4110678f 100644 --- a/crates/preloop-observability/src/export.rs +++ b/crates/preloop-observability/src/export.rs @@ -61,32 +61,136 @@ fn now_secs() -> u64 { .unwrap_or(0) } -fn now_nanos() -> u128 { +/// 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, +} + +/// 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), } /// Handle used by the rest of the process to enqueue telemetry. #[derive(Debug, Clone)] pub struct Exporter { - tx: mpsc::Sender, + tx: mpsc::Sender, health: Arc, } impl Exporter { /// Enqueue a log record. Never blocks; drops on a full queue. pub fn log(&self, record: LogRecord) { - if self.tx.try_send(record).is_err() { + 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); } } @@ -98,20 +202,27 @@ impl Exporter { /// 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( endpoint: Option<&str>, headers: Option<&str>, service_name: &str, instance_id: &str, service_version: &str, + metrics: Option>, ) -> Option<(Exporter, Arc)> { let endpoint = endpoint?.trim_end_matches('/').to_string(); let header_pairs = parse_headers(headers); - let service_name = service_name.to_string(); - let instance_id = instance_id.to_string(); - let service_version = service_version.to_string(); + 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 (tx, mut rx) = mpsc::channel::(QUEUE_CAPACITY); let worker_health = health.clone(); tokio::spawn(async move { @@ -122,35 +233,51 @@ pub fn spawn( Ok(client) => client, Err(error) => { // Sanitized: never the endpoint or headers. - tracing::warn!( - failure = "client_build", - %error, - "telemetry export disabled" - ); + tracing::warn!(failure = "client_build", %error, "telemetry export disabled"); return; } }; - let logs_url = format!("{endpoint}/v1/logs"); - let mut buffer: Vec = Vec::with_capacity(BATCH_MAX); + // 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 urls = Urls { + logs: format!("{endpoint}/v1/logs"), + traces: format!("{endpoint}/v1/traces"), + metrics: format!("{endpoint}/v1/metrics"), + }; + 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(record) => { - buffer.push(record); - if buffer.len() >= BATCH_MAX { - flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &service_version, &mut buffer, &worker_health).await; + Some(Item::Log(record)) => { + logs.push(record); + if logs.len() >= BATCH_MAX { + flush_logs(&client, &urls, &header_pairs, &resource, &mut logs, &worker_health).await; + } + } + Some(Item::Span(record)) => { + spans.push(record); + if spans.len() >= BATCH_MAX { + flush_spans(&client, &urls, &header_pairs, &resource, &mut spans, &worker_health).await; } } None => { - flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &service_version, &mut buffer, &worker_health).await; + // Channel closed: final drain, then exit. + flush_logs(&client, &urls, &header_pairs, &resource, &mut logs, &worker_health).await; + flush_spans(&client, &urls, &header_pairs, &resource, &mut spans, &worker_health).await; break; } } } _ = ticker.tick() => { - flush(&client, &logs_url, &header_pairs, &service_name, &instance_id, &service_version, &mut buffer, &worker_health).await; + flush_logs(&client, &urls, &header_pairs, &resource, &mut logs, &worker_health).await; + flush_spans(&client, &urls, &header_pairs, &resource, &mut spans, &worker_health).await; + if let Some(registry) = &metrics { + flush_metrics(&client, &urls, &header_pairs, &resource, registry, start_nanos, &worker_health).await; + } } } } @@ -165,23 +292,31 @@ pub fn spawn( )) } -async fn flush( +#[derive(Debug, Clone)] +struct Urls { + logs: String, + traces: String, + metrics: String, +} + +#[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)], - service_name: &str, - instance_id: &str, - service_version: &str, - buffer: &mut Vec, + payload: &Value, + signal: &'static str, + count: u64, health: &Arc, ) { - if buffer.is_empty() { - return; - } - let batch = std::mem::take(buffer); - let count = batch.len() as u64; - let payload = encode_logs(&batch, service_name, instance_id, service_version); - let mut req = client.post(url).json(&payload); + let mut req = client.post(url).json(payload); for (name, value) in headers { req = req.header(name.as_str(), value.as_str()); } @@ -191,6 +326,7 @@ async fn flush( // Status class only — never the body, which can echo credentials. tracing::warn!( failure = "http_status", + signal, status = response.status().as_u16(), "telemetry export failed" ); @@ -199,16 +335,112 @@ async fn flush( Err(_) => { // No error text: reqwest errors embed the URL, which may carry // credentials in userinfo. - tracing::warn!(failure = "transport", "telemetry export failed"); + tracing::warn!(failure = "transport", signal, "telemetry export failed"); health.record_failure(); } } } +async fn flush_logs( + client: &reqwest::Client, + urls: &Urls, + headers: &[(String, String)], + resource: &Resource, + buffer: &mut Vec, + health: &Arc, +) { + 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, &urls.logs, headers, &payload, "logs", count, health).await; +} + +async fn flush_spans( + client: &reqwest::Client, + urls: &Urls, + headers: &[(String, String)], + resource: &Resource, + buffer: &mut Vec, + health: &Arc, +) { + 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, + &urls.traces, + headers, + &payload, + "traces", + count, + health, + ) + .await; +} + +async fn flush_metrics( + client: &reqwest::Client, + urls: &Urls, + headers: &[(String, String)], + resource: &Resource, + registry: &Arc, + start_nanos: u128, + health: &Arc, +) { + 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, + &urls.metrics, + 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, @@ -233,25 +465,25 @@ pub fn encode_logs( .map(|record| { let attributes: Vec = record.attributes.iter().map(|(k, v)| attr(k, v)).collect(); - json!({ + 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": [ - attr("service.name", service_name), - attr("service.instance.id", instance_id), - attr("service.version", service_version), - ] - }, + "resource": {"attributes": resource_attributes(service_name, instance_id, service_version)}, "scopeLogs": [{ "scope": {"name": "preloop-observability"}, "logRecords": records, @@ -279,13 +511,166 @@ fn parse_headers(raw: Option<&str>) -> Vec<(String, 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(None, None, "preloop", "abc", "9.9.9").is_none()); + assert!(spawn(None, None, "preloop", "abc", "9.9.9", None).is_none()); } #[test] @@ -307,6 +692,8 @@ mod tests { severity: "WARN", body: "pool provisioning failed".to_string(), attributes: vec![("event.name".to_string(), "pool.provision".to_string())], + trace_id: None, + span_id: None, }]; let payload = encode_logs(&batch, "preloop", "inst-1", "9.9.9"); let record = &payload["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0]; @@ -323,3 +710,153 @@ mod tests { ); } } + +#[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 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()), + }]; + 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() { + 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, 3, 5], + 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)" + ); + 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 a82e9c14..589cfdfc 100644 --- a/crates/preloop-observability/src/lib.rs +++ b/crates/preloop-observability/src/lib.rs @@ -443,12 +443,16 @@ impl Observability { 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 exporter = export::spawn( config.otel_endpoint_raw(), config.otel_headers_raw(), &config.service_name, &config.instance_id, &config.service_version, + Some(metrics.clone()), ) .map(|(exporter, _health)| exporter); let handle = Self { @@ -456,7 +460,7 @@ impl Observability { 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, @@ -504,16 +508,43 @@ impl Observability { 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()), }); } } + /// 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()) diff --git a/crates/preloop-observability/src/metrics.rs b/crates/preloop-observability/src/metrics.rs index 5943bcc9..02de0788 100644 --- a/crates/preloop-observability/src/metrics.rs +++ b/crates/preloop-observability/src/metrics.rs @@ -566,3 +566,242 @@ 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 { + /// Cumulative bucket counts plus the implicit `+Inf` bucket OTLP requires. + fn otlp_bucket_counts(&self) -> Vec { + let mut counts: Vec = self.buckets.iter().map(|(_, c)| *c).collect(); + counts.push(self.count); + counts + } + + 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-runner-server/src/http_metrics.rs b/crates/preloop-runner-server/src/http_metrics.rs index 9dd9b28d..2283b8e4 100644 --- a/crates/preloop-runner-server/src/http_metrics.rs +++ b/crates/preloop-runner-server/src/http_metrics.rs @@ -56,6 +56,20 @@ pub async fn http_metrics_middleware( 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()), + ) + }); + let span_start = preloop_observability::export::now_nanos(); + let start = Instant::now(); let res = next.run(req).await; let elapsed = start.elapsed(); @@ -66,38 +80,39 @@ pub async fn http_metrics_middleware( if let Some(lbl) = labels { let mut lbl = lbl; lbl.status_class = sc.clone(); - // Record duration only for non-live_logs + let metrics = shared.state.observability.metrics(); + metrics.http.observe_duration(lbl.clone(), elapsed); + metrics.http.dec_active(&lbl); + } + + if let Some(context) = span_context { + // 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.clone()), + ("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: span_start, + 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). - - // 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; - res } From 25aa3f75b2e8f70c1b3ed96e63f18f1a3a0d7685 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 19:17:18 -0400 Subject: [PATCH 05/11] fix(server): keep the reason prose on the log and classify external-host failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Load testing exposed both halves of this. A sustained run produced 21 `reason="unrecognized"` job completions with no way to find out what they were: the previous change claimed the full message stayed on the log record, but only the bounded code was ever attached, so the prose was unrecoverable. Attach it as `reason.detail`. Logs are not a label space, and without it an `unrecognized` classification is a dead end. With the detail visible the path was obvious — a second never-claimable sentence the classifier did not match: no windows runner is registered with this server, so `runs-on: windows-latest` cannot be scheduled built at runtime_scheduling.rs with the platform interpolated. It is a distinct condition from the starvation sweep and gets its own code rather than folding into `no_runner`: the sweep means "no matching runner appeared within the grace window", which more capacity fixes, while this means the server has no runner of that platform class at all and never will until one is registered. Matching is on the invariant phrase, so any interpolated platform classifies. After the fix a mixed load of 40 workflow submits and 640 reads produced only `no_runner` (56) and `no_platform_runner` (8), with zero `unrecognized`, four route templates and two surfaces. Entire-Checkpoint: 01M0GQDWZHCJVWQ5TQB0XEETTB --- crates/preloop-runner-server/src/state.rs | 62 ++++++++++++++++++++--- 1 file changed, 55 insertions(+), 7 deletions(-) diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs index a0b10bd1..b62e2e5d 100644 --- a/crates/preloop-runner-server/src/state.rs +++ b/crates/preloop-runner-server/src/state.rs @@ -597,8 +597,19 @@ fn bounded_termination_reason(value: &str) -> &'static str { "startup_orphan" => return "startup_orphan", _ => {} } - // Prose paths — match on the stable leading phrase, never the whole - // string, so an interpolated label cannot change the classification. + // 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"; } @@ -938,11 +949,22 @@ impl AppState { // separate JobCompleted event; naming both `job.completed` // conflated two distinct records in the log stream. "job.status.terminal", - vec![ - ("event.name".to_string(), "job.status.terminal".to_string()), - ("conclusion".to_string(), conclusion.to_string()), - ("reason".to_string(), bounded_reason.to_string()), - ], + { + 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() => { @@ -1441,6 +1463,32 @@ mod termination_reason_tests { 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"); From d34b9fb7376ef0b479010753915e4d3c1db70682 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 21:25:38 -0400 Subject: [PATCH 06/11] fix(observability): bound metric labels and fix the active-requests gauge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings, all reproduced live against a running server: 1. Unbounded route labels. `normalize_route` short-circuited on `path.contains(':')`, returning the raw path as a label value. A colon is legal inside a path segment, so an unauthenticated 404 like `/evil:1234` created one permanent series per distinct URI — remote memory exhaustion. It now matches only exact entries in the template table, and the parameterized-template branch requires a non-empty child segment so a bare collection path (`/api/v1/runs`) resolves to its own template instead of the single-item one. 2. Unbounded method labels. `req.method().to_string()` copied extension methods (`X-0001`, …) verbatim into the same map. Methods are now allowlisted to the standard set with an `other` bucket. 3. Leaking active-requests gauge. The gauge was keyed on the full label set including `status_class`, which the middleware set to a "2xx" placeholder before the handler ran and overwrote with the real class after — so every non-2xx request incremented one series and decremented another. The `2xx` series grew without bound (160 phantom in-flight after a 4xx storm) and the 4xx/5xx series went negative. The gauge is now keyed on a status-free `ActiveLabels`, and the decrement runs from a drop guard so cancellation, panics, and client disconnects on a long poll release the slot too. 4. Malformed OTLP histograms. Buckets were stored cumulative (Prometheus `le` semantics) and emitted as-if-disjoint; every observation was counted once per bucket and the total exceeded `count`. `otlp_bucket_counts` now differences adjacent cumulative values and uses `count - last` for +Inf, so the counts conserve the total. Label values are also escaped in the Prometheus exposition as defense in depth. Verification: live server after the fix shows 0 phantom active requests, 0 raw evil-route series, methods collapsed to `other`, and every histogram's +Inf bucket equals its declared count. Tests cover the colon escape, the collection-route template match, label escaping, gauge increment/decrement idempotence, and the cumulative-to-disjoint conversion. Entire-Checkpoint: 01M0GYRVWN21MA39T53T6NB03R --- crates/preloop-observability/src/metrics.rs | 142 +++++++++++++++--- .../preloop-runner-server/src/http_metrics.rs | 86 ++++++++--- 2 files changed, 185 insertions(+), 43 deletions(-) diff --git a/crates/preloop-observability/src/metrics.rs b/crates/preloop-observability/src/metrics.rs index 02de0788..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(); @@ -605,11 +696,20 @@ pub struct MetricFamily { } impl Histogram { - /// Cumulative bucket counts plus the implicit `+Inf` bucket OTLP requires. + /// 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 counts: Vec = self.buckets.iter().map(|(_, c)| *c).collect(); - counts.push(self.count); - counts + 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 { diff --git a/crates/preloop-runner-server/src/http_metrics.rs b/crates/preloop-runner-server/src/http_metrics.rs index 2283b8e4..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,27 @@ 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 - }; - - if let Some(lbl) = &labels { - shared.state.observability.metrics().http.inc_active(lbl); - } + }; + shared + .state + .observability + .metrics() + .http + .inc_active(&labels); + ActiveGuard { + shared: shared.clone(), + labels, + } + }); // Adopt an inbound W3C trace so a caller's trace continues through the // control plane; otherwise start a root. Health and metrics probes are @@ -68,7 +82,10 @@ pub async fn http_metrics_middleware( .and_then(|value| value.to_str().ok()), ) }); - let span_start = preloop_observability::export::now_nanos(); + // 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; @@ -77,21 +94,28 @@ 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(); - let metrics = shared.state.observability.metrics(); - metrics.http.observe_duration(lbl.clone(), elapsed); - metrics.http.dec_active(&lbl); + 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) = span_context { + 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.clone()), + ("preloop.surface".to_string(), surface), ("http.response.status_code".to_string(), status.to_string()), ]; shared @@ -100,7 +124,7 @@ pub async fn http_metrics_middleware( .export_span(preloop_observability::export::SpanRecord { context, name: format!("{method} {route}"), - start_nanos: span_start, + 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 @@ -116,3 +140,21 @@ pub async fn http_metrics_middleware( res } + +/// 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, +} + +impl Drop for ActiveGuard { + fn drop(&mut self) { + self.shared + .state + .observability + .metrics() + .http + .dec_active(&self.labels); + } +} From 09f6be94fa76bf7325e5b569deddebd8812ba942 Mon Sep 17 00:00:00 2001 From: Bill Njoroge Date: Thu, 20 Aug 2026 21:25:55 -0400 Subject: [PATCH 07/11] feat(observability): per-signal OTLP endpoints, partial success, shutdown flush Three exporter defects from review, all confirmed in code: 1. Signal-specific endpoints were misrouted. A single endpoint was selected with a fallback chain and `/v1/logs`, `/v1/traces`, `/v1/metrics` were appended unconditionally, so `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT= http://collector/v1/traces` produced `/v1/traces/v1/traces` and routed logs and metrics through the trace URL. Resolution is now per signal: the signal-specific variable wins and is used as-is; the generic base gets the suffix. Headers follow the same per-signal pattern with a generic fallback. 2. `partialSuccess` was ignored. A 2xx with rejected records was recorded as full success. The response body is now parsed for rejected counts per signal and a partial rejection is recorded as a failure. 3. No shutdown flush. The runtime's documented "bounded 2s flush" was a comment with no implementation; buffered records were lost on every clean exit. The worker now selects on a shutdown signal, drains all three signal buffers, and the runtime awaits the join inside the 2s bound. Both binaries invoke it on every exit path. 4. Log timestamps were batch-level. Every record in a flush window got the same export-time timestamp, collapsing intra-batch ordering. Each record now carries its enqueue time. 5. The `mark_exited` heartbeat state was dead (nothing called it; a clean Drop deregisters) and `LimitRegistry::register` took the write lock twice; both cleaned up. Env-mutating config tests now serialize on a shared mutex so they cannot race on process-global `OTEL_*` variables. Verified live against OpenObserve: per-signal URLs resolve exactly (signal-specific as-is, generic suffixed), histograms export disjoint bucket counts that conserve `count`, and spans/logs/metrics all flow. Entire-Checkpoint: 01M0GYSD2YVNH359JTEEHD48H6 --- crates/preloop-observability/src/export.rs | 250 ++++++++++++++++---- crates/preloop-observability/src/lib.rs | 262 +++++++++++++++------ crates/preloop-runner-server/src/main.rs | 11 +- 3 files changed, 401 insertions(+), 122 deletions(-) diff --git a/crates/preloop-observability/src/export.rs b/crates/preloop-observability/src/export.rs index 4110678f..8c6149f8 100644 --- a/crates/preloop-observability/src/export.rs +++ b/crates/preloop-observability/src/export.rs @@ -163,6 +163,10 @@ pub struct LogRecord { /// 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 @@ -173,16 +177,46 @@ pub enum Item { 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); } @@ -198,6 +232,12 @@ impl Exporter { 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, @@ -207,15 +247,15 @@ impl Exporter { /// metrics registry on each tick, so all three signals share one batching /// cadence and one client. pub fn spawn( - endpoint: Option<&str>, - headers: Option<&str>, + targets: &ExportTargets, service_name: &str, instance_id: &str, service_version: &str, metrics: Option>, -) -> Option<(Exporter, Arc)> { - let endpoint = endpoint?.trim_end_matches('/').to_string(); - let header_pairs = parse_headers(headers); +) -> 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(), @@ -223,9 +263,11 @@ pub fn spawn( }; 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(); - tokio::spawn(async move { + let targets = targets.clone(); + let join = tokio::spawn(async move { let client = match reqwest::Client::builder() .timeout(Duration::from_secs(10)) .build() @@ -240,11 +282,6 @@ pub fn spawn( // 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 urls = Urls { - logs: format!("{endpoint}/v1/logs"), - traces: format!("{endpoint}/v1/traces"), - metrics: format!("{endpoint}/v1/metrics"), - }; 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); @@ -255,28 +292,36 @@ pub fn spawn( Some(Item::Log(record)) => { logs.push(record); if logs.len() >= BATCH_MAX { - flush_logs(&client, &urls, &header_pairs, &resource, &mut logs, &worker_health).await; + 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, &urls, &header_pairs, &resource, &mut spans, &worker_health).await; + flush_spans(&client, targets.traces.as_ref(), &resource, &mut spans, &worker_health).await; } } None => { - // Channel closed: final drain, then exit. - flush_logs(&client, &urls, &header_pairs, &resource, &mut logs, &worker_health).await; - flush_spans(&client, &urls, &header_pairs, &resource, &mut spans, &worker_health).await; + // 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, &urls, &header_pairs, &resource, &mut logs, &worker_health).await; - flush_spans(&client, &urls, &header_pairs, &resource, &mut spans, &worker_health).await; + 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, &urls, &header_pairs, &resource, registry, start_nanos, &worker_health).await; + flush_metrics(&client, targets.metrics.as_ref(), &resource, registry, start_nanos, &worker_health).await; } } } @@ -287,16 +332,38 @@ pub fn spawn( Exporter { tx, health: health.clone(), + shutdown: shutdown_tx, }, health, + join, )) } -#[derive(Debug, Clone)] -struct Urls { - logs: String, - traces: String, - metrics: String, +/// 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)] @@ -321,7 +388,29 @@ async fn post( req = req.header(name.as_str(), value.as_str()); } match req.send().await { - Ok(response) if response.status().is_success() => health.record_success(count), + 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!( @@ -343,12 +432,12 @@ async fn post( async fn flush_logs( client: &reqwest::Client, - urls: &Urls, - headers: &[(String, String)], + target: Option<&SignalTarget>, resource: &Resource, buffer: &mut Vec, health: &Arc, ) { + let Some(target) = target else { return }; if buffer.is_empty() { return; } @@ -360,17 +449,26 @@ async fn flush_logs( &resource.instance_id, &resource.service_version, ); - post(client, &urls.logs, headers, &payload, "logs", count, health).await; + post( + client, + &target.url, + &target.headers, + &payload, + "logs", + count, + health, + ) + .await; } async fn flush_spans( client: &reqwest::Client, - urls: &Urls, - headers: &[(String, String)], + target: Option<&SignalTarget>, resource: &Resource, buffer: &mut Vec, health: &Arc, ) { + let Some(target) = target else { return }; if buffer.is_empty() { return; } @@ -384,8 +482,8 @@ async fn flush_spans( ); post( client, - &urls.traces, - headers, + &target.url, + &target.headers, &payload, "traces", count, @@ -396,13 +494,13 @@ async fn flush_spans( async fn flush_metrics( client: &reqwest::Client, - urls: &Urls, - headers: &[(String, String)], + 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; @@ -417,8 +515,8 @@ async fn flush_metrics( ); post( client, - &urls.metrics, - headers, + &target.url, + &target.headers, &payload, "metrics", count, @@ -452,6 +550,18 @@ fn severity_number(severity: &str) -> u8 { } } +/// 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], @@ -459,12 +569,12 @@ pub fn encode_logs( instance_id: &str, service_version: &str, ) -> Value { - let ts = now_nanos().to_string(); 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, @@ -493,7 +603,7 @@ pub fn encode_logs( } /// Parse `OTEL_EXPORTER_OTLP_HEADERS` (`k1=v1,k2=v2`). -fn parse_headers(raw: Option<&str>) -> Vec<(String, String)> { +pub fn parse_headers(raw: Option<&str>) -> Vec<(String, String)> { let Some(raw) = raw else { return Vec::new(); }; @@ -670,7 +780,7 @@ mod tests { #[test] fn absent_endpoint_spawns_nothing() { - assert!(spawn(None, None, "preloop", "abc", "9.9.9", None).is_none()); + assert!(spawn(&ExportTargets::default(), "preloop", "abc", "9.9.9", None).is_none()); } #[test] @@ -694,6 +804,7 @@ mod tests { 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]; @@ -779,6 +890,56 @@ mod signal_tests { 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 { @@ -787,6 +948,7 @@ mod signal_tests { 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]; @@ -820,6 +982,9 @@ mod signal_tests { #[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", @@ -827,7 +992,7 @@ mod signal_tests { count: 5, sum: 0.25, bounds: vec![0.005, 0.01], - bucket_counts: vec![1, 3, 5], + bucket_counts: vec![1, 2, 2], attributes: vec![("http.route".to_string(), "/healthz".to_string())], }], }]; @@ -841,6 +1006,11 @@ mod signal_tests { 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"); } diff --git a/crates/preloop-observability/src/lib.rs b/crates/preloop-observability/src/lib.rs index 589cfdfc..df1b291c 100644 --- a/crates/preloop-observability/src/lib.rs +++ b/crates/preloop-observability/src/lib.rs @@ -77,11 +77,18 @@ pub struct ObservabilityConfig { 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, } @@ -105,23 +112,44 @@ 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 { @@ -130,20 +158,36 @@ impl ObservabilityConfig { 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 @@ -153,14 +197,9 @@ impl ObservabilityConfig { self } - /// Raw headers for transport construction. Never logged or in `Debug`. - pub fn otel_headers_raw(&self) -> Option<&str> { - self.otel_headers.as_deref() - } - /// 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('@') { @@ -184,13 +223,10 @@ impl fmt::Debug for ObservabilityConfig { .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() @@ -222,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 { @@ -234,7 +268,6 @@ impl TaskHeartbeat { HeartbeatEntry { critical, last_beat: Instant::now(), - exited: false, }, ); HeartbeatHandle { @@ -254,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 @@ -269,7 +296,6 @@ impl TaskHeartbeat { name, critical: e.critical, heartbeat_age: e.last_beat.elapsed(), - exited: e.exited, }) .collect() } @@ -279,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); } } @@ -325,7 +350,6 @@ pub struct TaskSnapshot { pub name: &'static str, pub critical: Criticality, pub heartbeat_age: Duration, - pub exited: bool, } // --------------------------------------------------------------------------- @@ -346,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) { @@ -422,8 +441,12 @@ impl Observability { 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 { @@ -446,15 +469,15 @@ impl Observability { // 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 exporter = export::spawn( - config.otel_endpoint_raw(), - config.otel_headers_raw(), + let spawned = export::spawn( + &config.export_targets(), &config.service_name, &config.instance_id, &config.service_version, Some(metrics.clone()), - ) - .map(|(exporter, _health)| exporter); + ); + 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), @@ -466,7 +489,7 @@ impl Observability { is_noop, }), }; - let runtime = ObservabilityRuntime::new(handle.clone()); + let runtime = ObservabilityRuntime::new(handle.clone(), worker); (handle, runtime) } @@ -528,6 +551,9 @@ impl Observability { 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(), }); } } @@ -561,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. @@ -573,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, } } @@ -610,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(); } } @@ -637,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(); @@ -647,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", @@ -670,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()); @@ -678,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(); @@ -751,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-runner-server/src/main.rs b/crates/preloop-runner-server/src/main.rs index afd6b67f..f74efe31 100644 --- a/crates/preloop-runner-server/src/main.rs +++ b/crates/preloop-runner-server/src/main.rs @@ -84,10 +84,10 @@ async fn main() -> anyhow::Result<()> { let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); preloop_observability::ObservabilityRuntime::install_fmt_subscriber(observability.config()); - // The handle must reach `ServerConfig`; holding it here alone would leave - // the server on the no-op handle installed by `AppState::new`, so nothing - // would ever export. - 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 { @@ -159,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(()) } From 6074d32bb9b9683eda2db7f9aaabf91e220145b9 Mon Sep 17 00:00:00 2001 From: Bnjoroge Date: Thu, 20 Aug 2026 21:37:54 -0400 Subject: [PATCH 08/11] fix(observability): serialize unmeasured host usage as absent, not zero MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sample_host is a stub until the cgroup/process sampler lands, so build_fleet_snapshot was emitting cpu_cores: 0.0 and memory_bytes: 0 — a consumer of /api/v1/status could not distinguish an idle fleet from an unmeasured one. VmHostUsage fields are now Option and skipped in JSON when None. Entire-Checkpoint: 01M0GZFAPGDSXX5JP9A2E0T5ZF --- crates/preloop-observability/src/vm_telemetry.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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(), } From c7ba9440dc38635c27bcd098a18cf0468fcc03af Mon Sep 17 00:00:00 2001 From: Bnjoroge Date: Thu, 20 Aug 2026 21:38:22 -0400 Subject: [PATCH 09/11] fix(server): drop the exited heartbeat state from status classification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TaskSnapshot no longer carries exited — a clean Drop deregisters, so the flag could only ever be false. The stale threshold stays a literal here; it is consolidated into one constant in the follow-up review-fixes PR. Entire-Checkpoint: 01M0GZG5SQ5X8W0GSW9T53DGR2 --- crates/preloop-runner-server/src/runs.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) 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, From fd82e91c3900883c7798c47d2bac75a3cdce0827 Mon Sep 17 00:00:00 2001 From: Bnjoroge Date: Thu, 20 Aug 2026 21:38:46 -0400 Subject: [PATCH 10/11] fix(server): stamp ready-queue entry time on jobs Jobs carry an enqueue timestamp so the claim path can measure true queue latency instead of a hardcoded placeholder; the field is serde-defaulted so snapshots persisted before the field existed restore as unknown and are skipped. (The claim-path recording lands with the review fixes.) Entire-Checkpoint: 01M0GZHJA9FSX6WRBH43JB9WFF --- crates/preloop-runner-server/src/models.rs | 15 +++++++++++++++ .../src/runtime_scheduling.rs | 2 ++ 2 files changed, 17 insertions(+) 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/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(), From 56ca02b9947a45626bf68162b580035884bd8d79 Mon Sep 17 00:00:00 2001 From: Bnjoroge Date: Thu, 20 Aug 2026 21:39:28 -0400 Subject: [PATCH 11/11] fix(contrib): require an explicit OpenObserve admin password A known default password on a loopback port is one forwarded-port or one other-local-user away from being public; compose now fails startup until ZO_ROOT_USER_PASSWORD is supplied. Entire-Checkpoint: 01M0GZJ67NNWNJNJ9KEYHPXT78 --- contrib/openobserve/compose.yml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/contrib/openobserve/compose.yml b/contrib/openobserve/compose.yml index 2682ba4a..7a6c3de5 100644 --- a/contrib/openobserve/compose.yml +++ b/contrib/openobserve/compose.yml @@ -17,7 +17,10 @@ services: - "127.0.0.1:5080:5080" environment: ZO_ROOT_USER_EMAIL: ${ZO_ROOT_USER_EMAIL:-admin@preloop.local} - ZO_ROOT_USER_PASSWORD: ${ZO_ROOT_USER_PASSWORD:-ChangeMe.Preloop1} + # 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