diff --git a/Cargo.toml b/Cargo.toml index 4d420f14..e3bd132b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,7 +38,7 @@ tempfile = "3" toml = "0.9" opentelemetry = "0.30" opentelemetry-otlp = { version = "0.30", default-features = false, features = ["http-json", "reqwest-blocking-client", "internal-logs"] } -opentelemetry_sdk = "0.30" +opentelemetry_sdk = { version = "0.30", features = ["spec_unstable_metrics_views"] } tungstenite = "0.30" [dev-dependencies] diff --git a/docs/vrs/06-observability/open-questions.md b/docs/vrs/06-observability/open-questions.md index c59d8a30..c6947b1e 100644 --- a/docs/vrs/06-observability/open-questions.md +++ b/docs/vrs/06-observability/open-questions.md @@ -2,17 +2,10 @@ Kept minimal; each blocks exactly one delivery slice, not the tree. -- **PR2 metric set.** Resolved by interview (decision record Q5): the RED-minimal - set — counters `reconcile_passes_total{result}`, `task_launches_total{driver}`, - `task_reaps_total{driver}`, `hook_invocations_total{hook,event}`, - `message_deliveries_total{result}`, `crash_loops_total`; histograms - `reconcile_pass_duration_seconds`, `session_start_duration_seconds`. Labels only - from bounded enums (result/driver/hook); ids stay in span attributes. -- **`st2 up ` span coverage.** The catalog reconcile paths are instrumented - (`st2.reconcile_pass` at the `up_loop_until` loop pass and in `up_once`), but the single-file - spec path (`reconcile_pass_specs` / `reconcile_pass_specs_with_sessions`, `src/run.rs`) emits - no spans yet. Same span shape applies; folded into PR2, which needs a pass over that call - chain for metrics anyway. +- **PR2 metric set.** Landed: the RED-minimal set from interview decision Q5, with names, + types, and label enums as specified in [spec.md](spec.md) (`src/metrics.rs`); the + `st2 up ` span coverage folded in as planned — all three reconcile-pass sites now + emit `st2.reconcile_pass` plus the pass counter and duration histogram. - **Remaining R04 resource attributes.** Resolved by source read (dotfiles dev3 `monitoring.nix` transform block): the platform edge stamps `service.namespace`, `sk.site`, `sk.role`, `deployment.environment.name` where absent, and the central diff --git a/docs/vrs/06-observability/spec.md b/docs/vrs/06-observability/spec.md index a9c1d90a..454907f4 100644 --- a/docs/vrs/06-observability/spec.md +++ b/docs/vrs/06-observability/spec.md @@ -44,7 +44,7 @@ One module, `src/telemetry.rs`, owns init and teardown via `Telemetry::init(unit - **Protocol**: HTTP JSON (`http-json` + protobuf-free wire), batch exporter, targeting the local Alloy forwarder at `127.0.0.1:4318` by convention. - **Resource**: `service.name` = `st2-` selected per entrypoint (below; `src/main.rs` - passes `supervisor` or `cli`), `service.version` from `crate::version::machine_version`, and + passes `supervisor`, `hook`, or `cli`), `service.version` from `crate::version::machine_version`, and `host.name` from the existing host detection. The remaining R04 fleet attributes (`service.namespace`, `service.instance.id`, `sk.site`, `sk.role`, `deployment.environment.name`) are not set yet — @@ -59,7 +59,7 @@ One module, `src/telemetry.rs`, owns init and teardown via `Telemetry::init(unit | --- | --- | | Supervisor loop (`st2 up` daemon / systemd unit) | `st2-supervisor` | | One-shot CLI invocations | `st2-cli` | -| Hook executions (`src/hooks.rs`) — not instrumented yet; planned value | `st2-hook` | +| Hook executions (`st2 driver claude-observe`; other hook surfaces not instrumented yet) | `st2-hook` | ## Trace roots @@ -73,20 +73,52 @@ Instrumented in PR1, one root span per unit of work: Not yet instrumented (follow-ups, not PR1 scope): -- Provider session lifecycles (claude / codex / opencode spawn, attach, teardown), exec sidecars - (`src/exec_backend.rs`), and hooks (`src/hooks.rs`). -- The `st2 up ` path (`reconcile_pass_specs` and friends) emits no spans yet — tracked as - an [open question](open-questions.md). +- Provider session lifecycles (claude / codex / opencode spawn, attach, teardown) beyond the + PR2 launch/reap counters, exec sidecars (`src/exec_backend.rs`). Span names follow the central `01-conventions` rules (`span.label` discipline included). Names are registered st2-side; this list plus PR2's metric set is that registry's seed. ## Metrics (PR2) -Exact metric set is open ([open-questions](open-questions.md)). Shape: counters for reconcile -passes, spawns, reaps, hook runs, and errors by kind; histograms for reconcile-pass duration and -provider-session start latency. Same resource attributes, same endpoint, batch exporter shared -with traces. +Landed RED-minimal set per interview decision Q5; every label value comes from a bounded enum, +and identifiers never become metric labels (ids stay in span attributes). `src/metrics.rs` owns +the instruments; every record call early-outs unless a meter provider is installed. + +| Instrument | Type | Labels | +| --- | --- | --- | +| `reconcile_passes_total` | counter | `result` = `pass` \| `fail` | +| `task_launches_total` | counter | `driver` = `codex` \| `claude` \| `opencode` \| `pi` \| `omp` \| `exec` \| `other` | +| `task_reaps_total` | counter | `driver` (same enum as launches) | +| `hook_invocations_total` | counter | `hook` = registry name (`claude-observe`), `event` = bounded Claude hook-event set, unknown → `other` | +| `message_deliveries_total` | counter | `result` = `pass` \| `fail` | +| `crash_loops_total` | counter | — | +| `reconcile_pass_duration_seconds` | histogram | — | +| `session_start_duration_seconds` | histogram | — | + +The duration histograms use seconds-scale explicit buckets (`1ms … 10s`, see +`DURATION_BUCKET_BOUNDARIES` in `src/telemetry.rs`) instead of the SDK's millisecond-tuned +defaults, so sub-second passes and spawns stay distinguishable. + +Scope notes: passes are counted at all three `st2.reconcile_pass` sites (catalog loop pass, +one-shot up, and the single-file spec path — `reconcile_pass_specs_with_sessions`, which now +emits the same root span shape); `fail` means the pass collected errors. Reaps count the +restart path in the launch loop, where driver context exists. Deliveries cover bus deliveries +onto a recipient inbox (`deliver_record`, send + retry paths); ding/native transport outcomes +are separate follow-ups. Hook invocations are observed at the single in-process application +point (`st2 driver claude-observe`); hook scripts the harnesses execute directly are not +visible to st2. The `driver` label is a closed enum resolved by precedence: `exec` task kind +first, then a typed driver declaration, then an observational argv/shell token heuristic +(alphanumeric tokens matched in launch order: `codex`, `claude`, `opencode`, `omp`, `pi`; anything +else → `other`). Because the heuristic inspects arbitrary user work, a hand-authored seat may +be labeled by what its command line merely mentions — the label is diagnostic only and never +influences reconcile decisions. + +The meter provider shares PR1's plumbing: `Telemetry::init` installs an `SdkMeterProvider` +with a `PeriodicReader` + OTLP/HTTP-JSON metric exporter behind the same +`OTEL_EXPORTER_OTLP_ENDPOINT` guard and resource; unset → no provider and the global meter is +a silent no-op (R02 zero-overhead). `Telemetry::shutdown` force-flushes metric points alongside +spans so short-lived CLI runs deliver them. ## Log bridge (PR3) diff --git a/src/claude_session.rs b/src/claude_session.rs index cc7db370..04c682de 100644 --- a/src/claude_session.rs +++ b/src/claude_session.rs @@ -78,6 +78,9 @@ pub fn run_observe( ) -> Result<()> { let agent_dir = message::resolve_agent_dir(catalog_root, identity, &crate::run::detect_host())? .with_context(|| format!("Claude driver agent '{identity}' is not declared"))?; + // Counted only once the invocation has its application target: a hook for an undeclared + // agent errors out before any state is applied and must not inflate `hook_invocations_total`. + crate::metrics::record_hook_invocation("claude-observe", event); let mut raw = String::new(); let _ = std::io::stdin().read_to_string(&mut raw); let payload = serde_json::from_str(&raw).unwrap_or(serde_json::Value::Null); diff --git a/src/lib.rs b/src/lib.rs index a1b0d51c..136684de 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,6 +29,7 @@ pub mod isolate; pub mod materialize; pub mod message; pub mod omp_session; +pub mod metrics; pub mod opencode_session; pub mod park; pub mod pi_channel; diff --git a/src/main.rs b/src/main.rs index 0a7ce97a..5a557273 100644 --- a/src/main.rs +++ b/src/main.rs @@ -992,12 +992,21 @@ fn main() -> Result<()> { command, } = Cli::parse(); - let mut telemetry = - st2::telemetry::Telemetry::init(if matches!(command, Command::Up { once: false, .. }) { + let mut telemetry = st2::telemetry::Telemetry::init( + if matches!(command, Command::Up { once: false, .. }) { "supervisor" + } else if matches!( + command, + // Hook executions are their own process unit: `st2 driver claude-observe` runs per + // Claude hook event and records hook_invocations_total, which the documented + // process-unit contract assigns to `st2-hook`, not `st2-cli`. + Command::Driver(DriverCmd::ClaudeObserve { .. }) + ) { + "hook" } else { "cli" - }); + }, + ); let result = dispatch(command, catalog_path.as_deref()); telemetry.shutdown(); result diff --git a/src/message.rs b/src/message.rs index 206a76b5..2ab71078 100644 --- a/src/message.rs +++ b/src/message.rs @@ -1844,15 +1844,32 @@ fn deliver_record(recipient: &DeliveryEndpoint, record: &SentRecord) -> anyhow:: let (inbox, archive) = recipient.boxes()?; let archived = archive.join(&record.filename); if archived.is_file() { - anyhow::ensure!( - fs::read_to_string(&archived)? == record.rendered_message, - "archived message differs from pending send {}", - record.filename - ); + let same = match fs::read_to_string(&archived) { + Ok(content) => content == record.rendered_message, + // A read failure is a failed delivery too: report it before propagating so + // message_deliveries_total{result="fail"} covers filesystem errors, not just mismatches. + Err(error) => { + crate::metrics::record_message_delivery(true); + return Err(error.into()); + } + }; + if !same { + crate::metrics::record_message_delivery(true); + anyhow::bail!("archived message differs from pending send {}", record.filename); + } + crate::metrics::record_message_delivery(false); return Ok(()); } - materialize_message_once(&inbox, &record.filename, &record.rendered_message)?; - Ok(()) + match materialize_message_once(&inbox, &record.filename, &record.rendered_message) { + Ok(_) => { + crate::metrics::record_message_delivery(false); + Ok(()) + } + Err(error) => { + crate::metrics::record_message_delivery(true); + Err(error) + } + } } fn key_path(root: &Path, to: &str, key: &str) -> anyhow::Result { diff --git a/src/metrics.rs b/src/metrics.rs new file mode 100644 index 00000000..b889dbdc --- /dev/null +++ b/src/metrics.rs @@ -0,0 +1,194 @@ +//! RED-minimal metric set for st2, per the interview Q5 decision recorded in +//! `docs/vrs/06-observability/open-questions.md`. +//! +//! Every label value comes from a bounded enum (`result`, `driver`, hook registry name + +//! normalized event); identifiers never become metric labels — those live in span attributes. +//! +//! Zero-overhead no-op unless a real meter provider is installed by +//! [`crate::telemetry::Telemetry::init`]: every record function checks [`enabled`] first and +//! returns before touching any instrument or allocating a label string. With no provider +//! installed, `opentelemetry::global` hands out a silent no-op meter anyway — this early-out +//! just keeps the disabled case allocation-free. + +use std::sync::LazyLock; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use opentelemetry::global; +use opentelemetry::metrics::{Counter, Histogram, Meter}; + +static ENABLED: AtomicBool = AtomicBool::new(false); + +/// Whether a real meter provider is installed. False → recording is a free no-op. +pub fn enabled() -> bool { + ENABLED.load(Ordering::Relaxed) +} + +/// Flip recording on when [`crate::telemetry::Telemetry::init`] installs the provider. +pub fn set_enabled(enabled: bool) { + ENABLED.store(enabled, Ordering::Relaxed); +} + +static METER: LazyLock = LazyLock::new(|| global::meter("st2")); + +static RECONCILE_PASSES: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("reconcile_passes_total") + .with_description("Reconcile passes by outcome") + .with_unit("1") + .build() +}); +static RECONCILE_PASS_DURATION: LazyLock> = LazyLock::new(|| { + METER + .f64_histogram("reconcile_pass_duration_seconds") + .with_description("Wall-clock duration of one reconcile pass") + .with_unit("s") + .build() +}); +static SESSION_START_DURATION: LazyLock> = LazyLock::new(|| { + METER + .f64_histogram("session_start_duration_seconds") + .with_description("Latency of one task session spawn") + .with_unit("s") + .build() +}); +static TASK_LAUNCHES: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("task_launches_total") + .with_description("Task sessions launched, by driver") + .with_unit("1") + .build() +}); +static TASK_REAPS: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("task_reaps_total") + .with_description("Dead sessions reaped for restart, by driver") + .with_unit("1") + .build() +}); +static HOOK_INVOCATIONS: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("hook_invocations_total") + .with_description("Lifecycle hook invocations applied in-process, by hook and event") + .with_unit("1") + .build() +}); +static MESSAGE_DELIVERIES: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("message_deliveries_total") + .with_description("Bus deliveries onto a recipient inbox, by outcome") + .with_unit("1") + .build() +}); +static CRASH_LOOPS: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("crash_loops_total") + .with_description("Tasks parked as crash-looping past their restart budget") + .with_unit("1") + .build() +}); + +/// One reconcile pass finished. `failed` = the pass collected errors. +pub fn record_reconcile_pass(duration: Duration, failed: bool) { + if !enabled() { + return; + } + RECONCILE_PASS_DURATION.record(duration.as_secs_f64(), &[]); + RECONCILE_PASSES.add( + 1, + &[opentelemetry::KeyValue::new("result", if failed { "fail" } else { "pass" })], + ); +} + +/// One task session spawn succeeded. +pub fn record_session_start(duration: Duration, driver: &'static str) { + if !enabled() { + return; + } + SESSION_START_DURATION.record(duration.as_secs_f64(), &[]); + TASK_LAUNCHES.add(1, &[opentelemetry::KeyValue::new("driver", driver)]); +} + +/// One dead session was reaped so its replacement can start. +pub fn record_task_reap(driver: &'static str) { + if !enabled() { + return; + } + TASK_REAPS.add(1, &[opentelemetry::KeyValue::new("driver", driver)]); +} + +/// One lifecycle-hook invocation reached its single in-process application point. +/// Unknown event names collapse to `other` so the label stays bounded. +pub fn record_hook_invocation(hook: &'static str, event: &str) { + if !enabled() { + return; + } + HOOK_INVOCATIONS.add( + 1, + &[ + opentelemetry::KeyValue::new("hook", hook), + opentelemetry::KeyValue::new("event", normalize_hook_event(event)), + ], + ); +} + +/// One bus delivery attempt onto a recipient inbox finished. `failed` = the attempt errored. +pub fn record_message_delivery(failed: bool) { + if !enabled() { + return; + } + MESSAGE_DELIVERIES.add( + 1, + &[opentelemetry::KeyValue::new("result", if failed { "fail" } else { "pass" })], + ); +} + +/// A task was parked as crash-looping past its restart budget. +pub fn record_crash_loop() { + if !enabled() { + return; + } + CRASH_LOOPS.add(1, &[]); +} + +/// The bounded Claude hook-event vocabulary st2 applies; anything else is `other`. +fn normalize_hook_event(event: &str) -> &'static str { + match event { + "SessionStart" => "SessionStart", + "UserPromptSubmit" => "UserPromptSubmit", + "PreToolUse" => "PreToolUse", + "PostToolUse" => "PostToolUse", + "PermissionRequest" => "PermissionRequest", + "Stop" => "Stop", + "SubagentStop" => "SubagentStop", + "PreCompact" => "PreCompact", + "Notification" => "Notification", + _ => "other", + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn disabled_by_default_and_recording_is_a_no_op() { + // No meter provider installed in unit tests: enabled() is false and every record call + // must return without panicking or touching instruments. + assert!(!enabled()); + set_enabled(false); + record_reconcile_pass(Duration::from_millis(5), false); + record_session_start(Duration::from_millis(5), "exec"); + record_task_reap("codex"); + record_hook_invocation("claude-observe", "SomethingUnheardOf"); + record_message_delivery(true); + record_crash_loop(); + assert!(!enabled()); + } + + #[test] + fn unknown_hook_events_collapse_to_other() { + assert_eq!(normalize_hook_event("SessionStart"), "SessionStart"); + assert_eq!(normalize_hook_event("TotallyNewEvent"), "other"); + } +} diff --git a/src/run.rs b/src/run.rs index 0e84f5b9..cf966b9f 100644 --- a/src/run.rs +++ b/src/run.rs @@ -1315,6 +1315,43 @@ fn stop_live_derived_companions( } } +/// Bounded driver label for lifecycle metrics (`task_launches_total` / `task_reaps_total`). +/// Typed drivers report their own name; legacy routed and hand-authored seats are classified +/// by what their launch actually invokes. Anything unrecognizable collapses to `other`, so +/// the label stays a closed set: `codex|claude|opencode|pi|omp|exec|other`. Observational only — +/// callers gate on [`crate::metrics::enabled`], and it never influences reconcile decisions. +fn driver_label(launch: &crate::reconcile::Launch<'_>, target: &TaskTarget) -> &'static str { + if target.kind == TaskKind::Exec { + return "exec"; + } + if let Some(driver) = &launch.spec.driver { + return driver.name(); + } + // Legacy routed (`deliver "mcp"` → claude-session, ...) or hand-authored seats: inspect + // the launch source by its alphanumeric tokens. + let tokens = |needle: &str| match &target.launch { + TaskLaunch::Argv(argv) => argv + .iter() + .any(|arg| arg.split(|c: char| !c.is_ascii_alphanumeric()).any(|t| t == needle)), + TaskLaunch::Shell(command) => command + .split(|c: char| !c.is_ascii_alphanumeric()) + .any(|t| t == needle), + }; + if tokens("codex") { + "codex" + } else if tokens("claude") { + "claude" + } else if tokens("opencode") { + "opencode" + } else if tokens("omp") { + "omp" + } else if tokens("pi") { + "pi" + } else { + "other" + } +} + fn execute_with_presentation_cursor( plan: &ReconcilePlan, runner: &dyn Runner, @@ -1394,7 +1431,9 @@ fn execute_with_presentation_cursor( let restarting = gc_set.contains(target.pty_id.as_str()); if restarting { match runner.reap_for_restart(&target.pty_id) { - Ok(()) => {} + Ok(()) => { + crate::metrics::record_task_reap(driver_label(launch, target)); + } Err(e) => { report .errors @@ -1407,8 +1446,13 @@ fn execute_with_presentation_cursor( } } } + let spawn_started = Instant::now(); match runner.spawn(target, spec_dir) { Ok(()) => { + crate::metrics::record_session_start( + spawn_started.elapsed(), + driver_label(launch, target), + ); cap.record(&target.pty_id, now); if restarting { report.restarted.push(target.pty_id.clone()); @@ -1724,6 +1768,7 @@ pub fn up_once(root: &Path, this_host: &str, runner: &dyn Runner) -> anyhow::Res let task_context = TaskCompileContext::current(root.to_path_buf())?; let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE); let pass_span = crate::telemetry::PassSpan::start(this_host); + let started = Instant::now(); let report = reconcile_pass( root, this_host, @@ -1734,6 +1779,7 @@ pub fn up_once(root: &Path, this_host: &str, runner: &dyn Runner) -> anyhow::Res &mut PresentationPatchCursor::default(), ); pass_span.finish(report.crash_loops.len(), report.unparked.len()); + crate::metrics::record_reconcile_pass(started.elapsed(), !report.errors.is_empty()); Ok(report) } @@ -1809,6 +1855,8 @@ pub(crate) fn reconcile_pass_specs_with_sessions( debounce: &mut LivenessDebounce, presentation_cursor: &mut PresentationPatchCursor, ) -> UpReport { + let started = Instant::now(); + let pass_span = crate::telemetry::PassSpan::start(this_host); let mut report = UpReport::default(); let now = Instant::now(); debounce.observe(sessions, now); @@ -1816,11 +1864,14 @@ pub(crate) fn reconcile_pass_specs_with_sessions( Ok(plan) => plan, Err(error) => { report.errors.push(error.to_string()); + drop(pass_span); + crate::metrics::record_reconcile_pass(started.elapsed(), true); return report; } }; - report.deferred = debounce.defer_flickers(&mut plan, now); execute_with_presentation_cursor(&plan, runner, cap, presentation_cursor, &mut report); + pass_span.finish(report.crash_loops.len(), report.unparked.len()); + crate::metrics::record_reconcile_pass(started.elapsed(), !report.errors.is_empty()); report } @@ -1978,6 +2029,9 @@ pub fn up_loop_specs( } for cl in &report.crash_loops { if reported_flapping.insert(cl.pty_id.clone()) { + // Counted once per park (the initial transition), not per pass: a task stays + // parked, so per-pass counting would inflate crash_loops_total unboundedly. + crate::metrics::record_crash_loop(); eprintln!( "st2: GAVE UP on '{id}' — crash-looping past its restart{{}} policy (mode=fail); leaving it parked and its last session for inspection. It is reported as parked by `st2 tasks`. Fix the cause, then `st2 unpark {id}` — no supervisor restart needed.", id = cl.pty_id @@ -2222,6 +2276,7 @@ fn up_loop_until( park_channel.grant_requests(&mut cap, &mut pre); let mut report = { let pass_span = crate::telemetry::PassSpan::start(this_host); + let started = Instant::now(); let pass = reconcile_pass( root, this_host, @@ -2232,6 +2287,7 @@ fn up_loop_until( &mut presentation_cursor, ); pass_span.finish(pass.crash_loops.len(), pass.unparked.len()); + crate::metrics::record_reconcile_pass(started.elapsed(), !pass.errors.is_empty()); pass }; pre.absorb(report); @@ -2248,6 +2304,9 @@ fn up_loop_until( park_channel.publish(&cap, &mut report); for cl in &report.crash_loops { if reported_flapping.insert(cl.pty_id.clone()) { + // Counted once per park (the initial transition), not per pass: a task stays + // parked, so per-pass counting would inflate crash_loops_total unboundedly. + crate::metrics::record_crash_loop(); eprintln!( "st2: GAVE UP on '{id}' — crash-looping past its restart{{}} policy (mode=fail); leaving it parked and its last session for inspection. It is reported as parked by `st2 tasks`. Fix the cause, then `st2 unpark {id}` — no supervisor restart needed.", id = cl.pty_id @@ -2340,7 +2399,9 @@ pub fn detect_host() -> String { #[cfg(test)] mod tests { use super::*; - use agent_spec::spec::{AgentSpec, JobType, Task, TaskKind, TaskLifecycle}; + use agent_spec::spec::{ + AgentSpec, Driver, JobType, OmpDriver, Task, TaskKind, TaskLifecycle, + }; use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, BTreeSet}; use std::ffi::OsStr; @@ -2968,6 +3029,59 @@ mod tests { } } + #[test] + fn driver_labels_include_typed_and_argv_omp_but_remain_bounded() { + let legacy_spec = spec_fixture(); + let legacy_launch = Launch { + spec: &legacy_spec, + tasks: Vec::new(), + live_derived: Vec::new(), + }; + let mut omp_argv = target("hetz.demo.agent", "unused"); + omp_argv.launch = TaskLaunch::Argv(vec![ + "st2".into(), + "driver".into(), + "omp-session".into(), + ]); + let mut exec = target("hetz.demo.agent", "codex"); + exec.kind = TaskKind::Exec; + let targets = [ + target("hetz.demo.agent", "codex"), + target("hetz.demo.agent", "claude"), + target("hetz.demo.agent", "opencode"), + target("hetz.demo.agent", "pi"), + omp_argv, + exec, + target("hetz.demo.agent", "unrecognized"), + ]; + let labels = targets + .iter() + .map(|target| driver_label(&legacy_launch, target)) + .collect::>(); + assert_eq!( + labels, + BTreeSet::from(["codex", "claude", "opencode", "pi", "omp", "exec", "other"]) + ); + + let mut typed_spec = spec_fixture(); + typed_spec.driver = Some(Driver::Omp(OmpDriver { + model: None, + effort: None, + prompt: String::new(), + args: Vec::new(), + })); + let typed_launch = Launch { + spec: &typed_spec, + tasks: Vec::new(), + live_derived: Vec::new(), + }; + assert_eq!( + driver_label(&typed_launch, &target("hetz.demo.agent", "claude")), + "omp", + "typed driver identity must take precedence over argv heuristics" + ); + } + #[test] fn debounce_absorbs_a_gc_flicker_but_reaps_a_stable_death() { let t0 = Instant::now(); diff --git a/src/telemetry.rs b/src/telemetry.rs index f0a9f976..3c97b571 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -1,6 +1,6 @@ //! OpenTelemetry export for st2, per `docs/vrs/06-observability/`. //! -//! Zero-overhead no-op unless `OTEL_EXPORTER_OTLP_ENDPOINT` is set; the exporter then ships +//! Zero-overhead no-op unless `OTEL_EXPORTER_OTLP_ENDPOINT` is set; the exporters then ship //! OTLP/HTTP JSON to the fleet's local Alloy forwarder (normally `127.0.0.1:4318`). The process //! model is sync (no tokio), so the blocking reqwest client is mandatory — enabling both //! reqwest client features of `opentelemetry-otlp` 0.30 compiles but fails at runtime with @@ -10,7 +10,8 @@ use std::sync::atomic::{AtomicBool, Ordering}; use opentelemetry::trace::{Span as _, Tracer as _}; use opentelemetry::KeyValue; -use opentelemetry_otlp::WithExportConfig; +use opentelemetry_otlp::{MetricExporter, SpanExporter, WithExportConfig}; +use opentelemetry_sdk::metrics::{Aggregation, Instrument, SdkMeterProvider, Stream}; use opentelemetry_sdk::trace::SdkTracerProvider; static ENABLED: AtomicBool = AtomicBool::new(false); @@ -22,10 +23,34 @@ pub fn enabled() -> bool { ENABLED.load(Ordering::Relaxed) } -/// Guard holding the tracer provider for a process lifetime. Dropping it flushes and shuts the -/// exporter down so short-lived CLI invocations still deliver their spans. +/// Seconds-scale explicit bucket boundaries for the duration histograms. The SDK's default +/// boundaries are millisecond-tuned (`[0, 5, 10, …, 10000]`), so sub-second reconcile passes +/// and session spawns would collapse into the lowest buckets and be indistinguishable. +const DURATION_BUCKET_BOUNDARIES: [f64; 12] = [ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, +]; + +/// View mapping each duration instrument onto seconds-scale explicit buckets (see +/// [`DURATION_BUCKET_BOUNDARIES`]); every other instrument keeps its default aggregation. +fn duration_view(instrument: &Instrument) -> Option { + match instrument.name() { + "reconcile_pass_duration_seconds" | "session_start_duration_seconds" => Stream::builder() + .with_aggregation(Aggregation::ExplicitBucketHistogram { + boundaries: DURATION_BUCKET_BOUNDARIES.into(), + record_min_max: true, + }) + .build() + .ok(), + _ => None, + } +} + +/// Guard holding the tracer and meter providers for a process lifetime. Dropping it flushes +/// and shuts both exporters down so short-lived CLI invocations still deliver their spans +/// and metric points. pub struct Telemetry { - provider: Option, + tracer_provider: Option, + meter_provider: Option, } impl Telemetry { @@ -33,19 +58,21 @@ impl Telemetry { /// follows the central observability contract's process-unit boundary: `st2-`. pub fn init(unit: &str) -> Self { if std::env::var_os("OTEL_EXPORTER_OTLP_ENDPOINT").is_none() { - return Self { provider: None }; + return Self { + tracer_provider: None, + meter_provider: None, + }; } - let exporter = match opentelemetry_otlp::SpanExporter::builder() - .with_http() - .with_protocol(opentelemetry_otlp::Protocol::HttpJson) - .build() - { + let span_exporter = match build_span_exporter() { Ok(exporter) => exporter, // Export setup must never take the runner down: telemetry is best-effort. Err(err) => { eprintln!("st2: otel exporter unavailable, continuing without telemetry: {err}"); - return Self { provider: None }; + return Self { + tracer_provider: None, + meter_provider: None, + }; } }; @@ -58,25 +85,53 @@ impl Telemetry { .with_attribute(KeyValue::new("host.name", crate::run::detect_host())) .build(); - let provider = SdkTracerProvider::builder() - .with_batch_exporter(exporter) + let tracer_provider = SdkTracerProvider::builder() + .with_batch_exporter(span_exporter) + .with_resource(resource.clone()) + .build(); + opentelemetry::global::set_tracer_provider(tracer_provider.clone()); + + // Metrics share endpoint, protocol, and resource with traces. The periodic reader's + // default interval only governs background collection — `shutdown` below force-flushes, + // so short-lived CLI runs still deliver their points. + let meter_exporter = match build_metric_exporter() { + Ok(exporter) => exporter, + Err(err) => { + eprintln!("st2: otel metric exporter unavailable, metrics disabled: {err}"); + return Self { + tracer_provider: Some(tracer_provider), + meter_provider: None, + }; + } + }; + let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(meter_exporter).build(); + let meter_provider = SdkMeterProvider::builder() + .with_reader(reader) + .with_view(duration_view) .with_resource(resource) .build(); - opentelemetry::global::set_tracer_provider(provider.clone()); + opentelemetry::global::set_meter_provider(meter_provider.clone()); + crate::metrics::set_enabled(true); ENABLED.store(true, Ordering::Relaxed); Self { - provider: Some(provider), + tracer_provider: Some(tracer_provider), + meter_provider: Some(meter_provider), } } - /// Whether export is active (endpoint configured and exporter initialized). + /// Whether export is active (endpoint configured and exporters initialized). pub fn enabled(&self) -> bool { - self.provider.is_some() + self.tracer_provider.is_some() } - /// Flush pending spans and stop the exporter. Safe to call multiple times. + /// Flush pending spans and metric points, then stop both exporters. Safe to call multiple times. pub fn shutdown(&mut self) { - if let Some(provider) = self.provider.take() { + if let Some(provider) = self.meter_provider.take() { + let _ = provider.force_flush(); + let _ = provider.shutdown(); + crate::metrics::set_enabled(false); + } + if let Some(provider) = self.tracer_provider.take() { let _ = provider.force_flush(); let _ = provider.shutdown(); } @@ -132,3 +187,19 @@ impl PassSpan { } } } + +/// OTLP/HTTP-JSON span exporter; blocking reqwest client only (module docs). +fn build_span_exporter() -> Result { + SpanExporter::builder() + .with_http() + .with_protocol(opentelemetry_otlp::Protocol::HttpJson) + .build() +} + +/// OTLP/HTTP-JSON metric exporter; same wire and client constraints as traces. +fn build_metric_exporter() -> Result { + MetricExporter::builder() + .with_http() + .with_protocol(opentelemetry_otlp::Protocol::HttpJson) + .build() +} diff --git a/tests/otel_export.rs b/tests/otel_export.rs index a13d4e8b..8b37238c 100644 --- a/tests/otel_export.rs +++ b/tests/otel_export.rs @@ -47,6 +47,40 @@ fn run_with_capture( .expect("run st2 up --once under otelite") } +/// Flatten an OTLP/HTTP-JSON `ExportMetricsServiceRequest` into its metric records. +fn metric_records(line: &str) -> Vec { + let mut records = Vec::new(); + let Ok(value) = serde_json::from_str::(line) else { + return records; + }; + let Some(resource_metrics) = value.get("resourceMetrics").and_then(|v| v.as_array()) else { + return records; + }; + for resource_metric in resource_metrics { + let Some(scope_metrics) = + resource_metric.get("scopeMetrics").and_then(|v| v.as_array()) + else { + continue; + }; + for scope_metric in scope_metrics { + if let Some(batch) = scope_metric.get("metrics").and_then(|v| v.as_array()) { + records.extend(batch.iter().cloned()); + } + } + } + records +} + +/// A data point's string-valued attribute by key, if present. +fn string_attr<'a>(point: &'a serde_json::Value, key: &str) -> Option<&'a str> { + point["attributes"] + .as_array()? + .iter() + .find(|attr| attr["key"].as_str() == Some(key))?["value"] + .get("stringValue") + .and_then(|v| v.as_str()) +} + #[test] fn st2_exports_spans_to_otelite_when_endpoint_is_set() { let Some(otelite) = std::env::var_os("ST2_OTELITE_BIN").map(PathBuf::from) else { @@ -81,6 +115,43 @@ fn st2_exports_spans_to_otelite_when_endpoint_is_set() { traces.contains("st2-cli"), "service.name st2-cli missing from capture:\n{traces}" ); + + // PR2: the same run must deliver metric points over the shared OTLP/HTTP endpoint. + // An empty-catalog `up --once` records exactly one reconcile pass (pass) plus its + // duration histogram sample. + let metrics = + std::fs::read_to_string(cap_dir.join("metrics.ndjson")).expect("metrics.ndjson written"); + let all_metrics: Vec = + metrics.lines().flat_map(metric_records).collect(); + + let passes = all_metrics + .iter() + .find(|m| m["name"].as_str() == Some("reconcile_passes_total")) + .expect("reconcile passes counter missing from capture"); + let pass_point = &passes["sum"]["dataPoints"][0]; + assert_eq!( + string_attr(pass_point, "result"), + Some("pass"), + "reconcile passes counter must carry result=pass:\n{passes}" + ); + + // The view must replace the SDK's millisecond-tuned default boundaries with seconds-scale + // buckets, or every sub-second pass sample collapses into the lowest bucket (P2). + let duration = all_metrics + .iter() + .find(|m| m["name"].as_str() == Some("reconcile_pass_duration_seconds")) + .expect("reconcile pass duration histogram missing from capture"); + let bounds = &duration["histogram"]["dataPoints"][0]["explicitBounds"]; + assert_eq!( + bounds.as_array().map(Vec::len), + Some(12), + "duration histogram must carry the 12 seconds-scale boundaries:\n{duration}" + ); + assert_eq!( + bounds[0].as_f64(), + Some(0.001), + "lowest duration bucket must be 1ms, not the SDK default:\n{duration}" + ); } #[test]