From 9d86ac8a5a13c61eb1da6ac9f036c02c742641a6 Mon Sep 17 00:00:00 2001 From: Greg Clark Date: Fri, 28 Aug 2026 14:29:26 -0400 Subject: [PATCH 1/2] fix(libsy): in-flight algo count metric Signed-off-by: Greg Clark --- .../libsy-llm-client/tests/observability.rs | 126 ++++++++++++++++++ crates/libsy/src/observability.rs | 42 +++++- docs/internal/metrics_reference.md | 29 +++- 3 files changed, 195 insertions(+), 2 deletions(-) diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 39741b867..2538c3d91 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -311,6 +311,34 @@ fn f64_histogram_sum_ms( }) } +/// Most recent value of an `i64` up-down counter for the given attribute set. +/// +/// Unlike the counter and histogram helpers, this takes the value from the last +/// snapshot that carries the metric rather than the max across snapshots: an +/// up-down counter falls as well as rises, so a max would report a past peak and +/// never observe the return to zero. +fn i64_up_down_counter_value( + snapshots: &[ResourceMetrics], + name: &str, + wanted: &[(&str, &str)], +) -> Option { + snapshots + .iter() + .flat_map(|snapshot| snapshot.scope_metrics()) + .filter(|scope| scope.scope().name() == "switchyard") + .flat_map(|scope| scope.metrics()) + .filter(|metric| metric.name() == name) + .filter_map(|metric| match metric.data() { + AggregatedMetrics::I64(MetricData::Sum(sum)) => sum + .data_points() + .filter(|point| attributes_match(point.attributes(), wanted)) + .map(|point| point.value()) + .last(), + _ => None, + }) + .last() +} + /// Latest value of a `u64` observable gauge. fn u64_gauge_value(snapshots: &[ResourceMetrics], name: &str) -> Option { latest_metric_value(snapshots, name, |data| match data { @@ -1436,3 +1464,101 @@ async fn classifier_fail_open_records_each_failure_stage() -> switchyard_libsy:: } Ok(()) } + +#[tokio::test] +async fn in_flight_gauge_reads_a_run_parked_on_an_unanswered_routing_call() +-> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (_, exporter, provider, _, _) = telemetry(); + const ALGO: &str = "obs-in-flight-algo"; + const MODEL: &str = "obs-in-flight-model"; + let algorithm = Arc::new(RoutingCallAlgo { + name: ALGO.to_string(), + target: MODEL.into(), + }); + let stream = algorithm.run_stream(request_with_metadata("obs-session-if", "obs-corr-if")); + tokio::pin!(stream); + let attributes = [("algorithm", ALGO)]; + + // Take the offloaded call and hold it without responding. This is the shape of a + // stalled classifier: the run has started and cannot proceed until the call returns. + let Some(Ok(Step::CallModel(call))) = stream.next().await else { + return Err(test_error("expected an offloaded routing call")); + }; + + let snapshots = flushed_metrics(exporter, provider); + assert_eq!( + i64_up_down_counter_value(&snapshots, "switchyard.algorithms_in_flight", &attributes), + Some(1), + "a run waiting on an unanswered routing call must read as in flight" + ); + // The gap this gauge fills: every other run metric is recorded on resolution, so + // while the call is outstanding they say nothing at all. + assert_eq!( + u64_counter_value(&snapshots, "switchyard.runs", &attributes), + None, + "the run counter must not report a run that has not resolved" + ); + + call.respond(Ok(Response { + llm_response: LlmResponse::Agg(text_response(Some(MODEL.to_string()), "answer")), + metadata: None, + }))?; + while stream.next().await.is_some() {} + + let snapshots = flushed_metrics(exporter, provider); + assert_eq!( + i64_up_down_counter_value(&snapshots, "switchyard.algorithms_in_flight", &attributes), + Some(0), + "the gauge must fall back to zero once the run resolves" + ); + Ok(()) +} + +#[tokio::test] +async fn in_flight_gauge_clears_when_a_run_is_abandoned() -> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (_, exporter, provider, _, _) = telemetry(); + const ALGO: &str = "obs-abandoned-algo"; + const MODEL: &str = "obs-abandoned-model"; + let algorithm = Arc::new(RoutingCallAlgo { + name: ALGO.to_string(), + target: MODEL.into(), + }); + let attributes = [("algorithm", ALGO)]; + + // Drop the step stream while the routing call is still outstanding, the way a + // disconnected client abandons a run. The run task is aborted mid-await and never + // reaches the code that follows it, so only a drop can return the count. + { + let stream = algorithm.run_stream(request_with_metadata("obs-session-ab", "obs-corr-ab")); + tokio::pin!(stream); + let Some(Ok(Step::CallModel(_call))) = stream.next().await else { + return Err(test_error("expected an offloaded routing call")); + }; + assert_eq!( + i64_up_down_counter_value( + &flushed_metrics(exporter, provider), + "switchyard.algorithms_in_flight", + &attributes, + ), + Some(1), + ); + } + + // Let the runtime finish aborting the task it was told to drop. + for _ in 0..10 { + tokio::task::yield_now().await; + } + + assert_eq!( + i64_up_down_counter_value( + &flushed_metrics(exporter, provider), + "switchyard.algorithms_in_flight", + &attributes, + ), + Some(0), + "an abandoned run must not strand the gauge above zero" + ); + Ok(()) +} diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index f43f02bfb..ba0da9821 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -110,13 +110,53 @@ pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span { span } +/// Holds `switchyard.algorithms_in_flight` up by one for as long as it lives. +/// +/// The decrement is a `Drop` rather than a statement after the run's `.await` +/// because a run task is aborted when its step stream is dropped (a disconnected +/// client, a host timeout). A cancelled task never reaches code past the await, +/// so an explicit decrement would strand the gauge above zero for the life of +/// the process. +struct InFlightRun { + algorithm: String, +} + +impl InFlightRun { + fn enter(algorithm: &str) -> Self { + record_algorithms_in_flight(algorithm, 1); + Self { + algorithm: algorithm.to_string(), + } + } +} + +impl Drop for InFlightRun { + fn drop(&mut self) { + record_algorithms_in_flight(&self.algorithm, -1); + } +} + +/// Adds `delta` to the count of algorithm runs that have started and not yet +/// finished. Unlike the run counter and duration histogram, which are recorded +/// once a run resolves, this reads non-zero *during* a run — including one +/// parked on a classifier or judge call that has not come back. +fn record_algorithms_in_flight(algorithm: &str, delta: i64) { + meter() + .i64_up_down_counter("switchyard.algorithms_in_flight") + .build() + .add(delta, &[KeyValue::new("algorithm", algorithm.to_string())]); +} + /// Runs one algorithm task to completion, recording the run counter, duration -/// histogram, span outcome, and failure log when it resolves. +/// histogram, span outcome, and failure log when it resolves. Counts the run as +/// in flight for its whole duration. /// Executes inside the `libsy.run` span its caller instruments the task with. pub(crate) async fn observe_run( algorithm: &str, run: impl Future>, ) -> Result { + // Binding, not `let _ =`: the guard must live until the run resolves. + let _in_flight = InFlightRun::enter(algorithm); let started = Instant::now(); let result = run.await; let duration = started.elapsed(); diff --git a/docs/internal/metrics_reference.md b/docs/internal/metrics_reference.md index 21ef27b11..cbc4857f5 100644 --- a/docs/internal/metrics_reference.md +++ b/docs/internal/metrics_reference.md @@ -59,6 +59,32 @@ Each histogram emits `_bucket`, `_sum`, and `_count` series. Use |---|---|---| | `switchyard_routing_overhead_ms{algorithm}` | histogram | Total run time minus the time spent in successful routed model calls, with overlapping hedged calls counted once. Includes classifier calls, failed routed attempts, target resolution, and decision publication; runs with no successful routed call are not recorded. Measured across the whole run, so it does not reconcile with `switchyard_run_duration_ms`, which times only the algorithm task. | +## In-flight runs + +| Metric | Type | Meaning | +|---|---|---| +| `switchyard_algorithms_in_flight{algorithm}` | gauge | Algorithm runs that have started and not yet finished. | + +Every other run metric is recorded when a run *resolves*, so none of them says +anything while a request is still being worked on. This gauge is the exception: +it rises when a run starts and falls when it ends, including when the run ends +because the client disconnected or the host cancelled it. A run parked on an +internal routing call — a classifier or escalation-judge request that has not +come back — holds the gauge up for as long as it waits, which is what separates +a stalled routing decision from client-side latency. + +Read it alongside the resolution metrics: a gauge that is high while +`switchyard_runs_total` is flat means work is going in and not coming out. + +```promql +# Runs currently in flight, by algorithm +sum by (algorithm) (switchyard_algorithms_in_flight) + +# In-flight work that is not being retired: runs are open but none are completing +sum by (algorithm) (switchyard_algorithms_in_flight) > 0 + and sum by (algorithm) (rate(switchyard_runs_total[5m])) == 0 +``` + ## Classifier fail-open counter | Metric | Type | Meaning | @@ -149,7 +175,7 @@ into label space. | `outcome` | Exactly 3: `success`, `retryable_error`, `other_error`. | Outcome counters | | `code` | Bounded: the known-code allowlist (`200`, `400`, `401`, `403`, `404`, `408`, `409`, `422`, `429`, `500`, `502`, `503`, `504`), plus `none` and the per-class buckets `1xx`/`2xx`/`3xx`/`4xx`/`5xx`/`other`. About 20 values max. | `switchyard_upstream_attempts_total` | | `le` | The configured histogram bucket boundaries. | Histogram buckets | -| `algorithm` | One stable value per configured algorithm. | Routing-overhead histogram | +| `algorithm` | One stable value per configured algorithm. | Routing-overhead histogram, in-flight gauge | | `tier` | Small enumerated set, optional. | Per-endpoint counters and histograms on algorithms that supply it | | `judge_model` | One per configured judge target. | Classifier fail-open counter | | `reason` | Exactly 8 fixed error categories. | Classifier fail-open counter | @@ -161,5 +187,6 @@ into label space. | `model=""` rows appear | A routed-call observation did not include a selected model. | | All counters at 0 after warm-up | Server just started with no traffic, or the scraper is hitting the wrong port. | | `switchyard_routing_overhead_ms_count` stuck at `0` | No successful algorithm run has recorded a successful routed model call. | +| `switchyard_algorithms_in_flight` stuck above zero with no traffic | Runs are parked on an internal routing call that never returns. Check the classifier or judge target's upstream. | | `switchyard_classifier_fail_open_total` rising | The judge target is failing or returning a response the classifier cannot parse. Check `judge_model` and `reason`. | | `switchyard_client_responses_total{outcome="retryable_error"}` rising | Either the upstream is genuinely flaky, or retries are exhausting; compare client responses with retryable upstream attempts. | From 8a1ffff55582a7b83c71685c3adb36bb6cc9f81b Mon Sep 17 00:00:00 2001 From: Greg Clark Date: Mon, 31 Aug 2026 13:29:39 -0400 Subject: [PATCH 2/2] chore: cleanup Signed-off-by: Greg Clark --- crates/libsy/src/observability.rs | 10 +--------- docs/internal/metrics_reference.md | 20 -------------------- 2 files changed, 1 insertion(+), 29 deletions(-) diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index ba0da9821..ab6c4c89d 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -111,12 +111,6 @@ pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span { } /// Holds `switchyard.algorithms_in_flight` up by one for as long as it lives. -/// -/// The decrement is a `Drop` rather than a statement after the run's `.await` -/// because a run task is aborted when its step stream is dropped (a disconnected -/// client, a host timeout). A cancelled task never reaches code past the await, -/// so an explicit decrement would strand the gauge above zero for the life of -/// the process. struct InFlightRun { algorithm: String, } @@ -137,9 +131,7 @@ impl Drop for InFlightRun { } /// Adds `delta` to the count of algorithm runs that have started and not yet -/// finished. Unlike the run counter and duration histogram, which are recorded -/// once a run resolves, this reads non-zero *during* a run — including one -/// parked on a classifier or judge call that has not come back. +/// finished. fn record_algorithms_in_flight(algorithm: &str, delta: i64) { meter() .i64_up_down_counter("switchyard.algorithms_in_flight") diff --git a/docs/internal/metrics_reference.md b/docs/internal/metrics_reference.md index cbc4857f5..464c5d737 100644 --- a/docs/internal/metrics_reference.md +++ b/docs/internal/metrics_reference.md @@ -65,26 +65,6 @@ Each histogram emits `_bucket`, `_sum`, and `_count` series. Use |---|---|---| | `switchyard_algorithms_in_flight{algorithm}` | gauge | Algorithm runs that have started and not yet finished. | -Every other run metric is recorded when a run *resolves*, so none of them says -anything while a request is still being worked on. This gauge is the exception: -it rises when a run starts and falls when it ends, including when the run ends -because the client disconnected or the host cancelled it. A run parked on an -internal routing call — a classifier or escalation-judge request that has not -come back — holds the gauge up for as long as it waits, which is what separates -a stalled routing decision from client-side latency. - -Read it alongside the resolution metrics: a gauge that is high while -`switchyard_runs_total` is flat means work is going in and not coming out. - -```promql -# Runs currently in flight, by algorithm -sum by (algorithm) (switchyard_algorithms_in_flight) - -# In-flight work that is not being retired: runs are open but none are completing -sum by (algorithm) (switchyard_algorithms_in_flight) > 0 - and sum by (algorithm) (rate(switchyard_runs_total[5m])) == 0 -``` - ## Classifier fail-open counter | Metric | Type | Meaning |