diff --git a/docs/scheduler/observability.md b/docs/scheduler/observability.md index 01678185..6927dd4f 100644 --- a/docs/scheduler/observability.md +++ b/docs/scheduler/observability.md @@ -522,6 +522,38 @@ at warn, with the kind of failure (`token`, `transport` or `status `) and the start of the response body. It repeats a persisting failure every 10 minutes with a count, and logs once at info when pushes succeed again. +## Review classifier result metrics (`swarm-controller`) + +`swarm-controller` counts what review requests and review runs come to, from +the same exporter as the VCS metrics above. The classifier service adds the +retry counter beside its cost counters +([`review-classifier.md`](../swarm/review-classifier.md#observability-and-cost)). + +| Metric | Unit | Kind | Attributes | +| ------------------------------------------ | ---- | --------- | ------------------------------------------------------------------------------------ | +| `hyperhive.review_classifier.requests` | — | counter | `outcome`: `queued`, `queue_failed`, `unparsable`, `deferred`, `started`, `reposted` | +| `hyperhive.review_classifier.verdicts` | — | counter | `outcome`: `approve`, `request_changes`, `no_verdict` | +| `hyperhive.review_classifier.checks` | — | counter | `check` (the rule's name), `result`: `match`, `no_match`, `unsure`, `no_answer` | +| `hyperhive.review_classifier.run.duration` | `s` | histogram | `outcome`: the verdict names, or `post_failed` | +| `hyperhive.review_classifier.ask.retries` | — | counter | `runtime`: `api_key`, `acp` | + +A verdict means the classifier decided, before it posts the review: a run +whose post fails still adds to `verdicts`, and shows as `post_failed` in +`run.duration`. A repo with no rules file gets an approval with no model call and +counts as an `approve` verdict and run, with no `checks`. + +`deferred` counts each time the queue hands over a request for a PR whose +review is still going, so a long review adds one per retry wait. `reposted` +is a request for a head already reviewed: it posts the earlier verdict and +adds to `verdicts` but runs no checks, so it adds nothing to `checks` or +`run.duration`. A request the queue refuses posts a comment and counts +as `queue_failed` and a `no_verdict`. `no_answer` is a classification that +failed, so a `no_verdict` run with a `no_answer` check is the usual cause of +a comment instead of a review. The `check` values are the names in a repo's +rules file, at most 10 per repo. + +The `review-classifier` dashboard in Grafana charts these. + ## Agent-emitted custom metrics (`hive-metric`) Agents can push arbitrary labeled metrics to the same OTEL collector via the diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index ac9d97f4..1c76bf9a 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -674,6 +674,13 @@ the types of `claude_code.token.usage`, which `claude` exports on the claude runtimes, so the cache hit rate reads the same way on every runtime: `cacheRead` over the sum of `input`, `cacheRead` and `cacheCreation`. +The controller counts what requests and runs come to as the counters +`hyperhive.review_classifier.requests`, `.verdicts` and `.checks` and the +histogram `.run.duration`, and the service counts each ask it repeats because +an answer failed validation as `hyperhive.review_classifier.ask.retries`. The +[metrics table](../scheduler/observability.md#review-classifier-result-metrics-swarm-controller) +lists their labels, and the `review-classifier` Grafana dashboard charts them. + The classifier logs each run's result to its journal, which ships to VictoriaLogs like every swarm service container's ([`services.md`](services.md#logs-victorialogs)). diff --git a/docs/swarm/services.md b/docs/swarm/services.md index 6b735ae6..b0c7940c 100644 --- a/docs/swarm/services.md +++ b/docs/swarm/services.md @@ -207,14 +207,16 @@ kept. - `authelia`, `bao`, `queue`: the identity provider, the secret store and the NATS queue. - `claude-usage`: Claude use by model, effort, token type and source. Filter by hive, service (agents or the review classifier) and agent. The service filter separates the classifier's API-key spend from the agents. - `controller`: the swarm controller's job graph by state, and the push and commit rates it sees on the forge. +- `review-classifier`: review requests by outcome, the verdict mix, results per check, retried asks, and run duration p50 and p95. - `forge`: what the forge holds, and the forge process's CPU, memory and descriptors. - `gateway`: the gateway nginx on each hive's host: requests per second, connections by state, and accepted against handled connections. - `collector`: what the OTEL collectors accept, refuse, export and queue, with queue depth against capacity. - `logstore`, `metricstore`: health of the log store and the metrics store. A new panel can read "No data" until its source exists: the push and commit -counters on `controller` appear after the first push webhook, and the refused -and failure series on `collector` appear once something fails. +counters on `controller` appear after the first push webhook, the refused +and failure series on `collector` appear once something fails, and the +`review-classifier` panels fill as review requests arrive. **Where the data comes from.** The swarm's OTEL collector, below. diff --git a/nix/host-modules/swarm-grafana.nix b/nix/host-modules/swarm-grafana.nix index 6ac90e76..d6d77012 100644 --- a/nix/host-modules/swarm-grafana.nix +++ b/nix/host-modules/swarm-grafana.nix @@ -278,6 +278,7 @@ let "logstore.json" "metricstore.json" "queue.json" + "review-classifier.json" ] ); diff --git a/nix/host-modules/swarm-grafana/dashboards/review-classifier.json b/nix/host-modules/swarm-grafana/dashboards/review-classifier.json new file mode 100644 index 00000000..ef66fc38 --- /dev/null +++ b/nix/host-modules/swarm-grafana/dashboards/review-classifier.json @@ -0,0 +1,641 @@ +{ + "title": "hyperhive · review classifier results", + "uid": "hyperhive-review-classifier", + "description": "Review requests, verdicts, per-check results, retried asks and run duration.", + "editable": true, + "refresh": "1m", + "schemaVersion": 39, + "tags": ["hyperhive", "swarm", "review-classifier"], + "time": { + "from": "now-24h", + "to": "now" + }, + "timezone": "utc", + "version": 1, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Requests queued", + "description": "Review requests the queue stored in the window.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 0, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.review_classifier.requests\", outcome=\"queued\"}[$__range])) or vector(0)", + "legendFormat": "Requests queued", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "text" + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 2, + "type": "stat", + "title": "Approved", + "description": "Runs that approved, including a re-post of an earlier approval.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 6, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.review_classifier.verdicts\", outcome=\"approve\"}[$__range])) or vector(0)", + "legendFormat": "Approved", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "green" + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 3, + "type": "stat", + "title": "Changes requested", + "description": "Runs that requested changes, including a re-post.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 12, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.review_classifier.verdicts\", outcome=\"request_changes\"}[$__range])) or vector(0)", + "legendFormat": "Changes requested", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "orange" + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 4, + "type": "stat", + "title": "No verdict", + "description": "Runs that posted a comment only: a failed run, a check with no answer, or rules that did not load.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 18, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.review_classifier.verdicts\", outcome=\"no_verdict\"}[$__range])) or vector(0)", + "legendFormat": "No verdict", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "fixed", + "fixedColor": "red" + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 5, + "type": "timeseries", + "title": "Requests by outcome", + "description": "Requests by what became of them: queued, started, reposted, deferred, queue_failed or unparsable.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 4 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (outcome) (increase({__name__=\"hyperhive.review_classifier.requests\"}[$__interval]))", + "legendFormat": "{{outcome}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "bars", + "lineWidth": 1, + "fillOpacity": 60, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0, + "stacking": { + "mode": "normal", + "group": "A" + } + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, + { + "id": 6, + "type": "timeseries", + "title": "Verdicts over time", + "description": "The review event each run chose, including a re-post of an earlier verdict.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 4 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (outcome) (increase({__name__=\"hyperhive.review_classifier.verdicts\"}[$__interval]))", + "legendFormat": "{{outcome}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "bars", + "lineWidth": 1, + "fillOpacity": 60, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0, + "stacking": { + "mode": "normal", + "group": "A" + } + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, + { + "id": 7, + "type": "piechart", + "title": "Verdict mix", + "description": "Share of each verdict in the window.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 12 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (outcome) (increase({__name__=\"hyperhive.review_classifier.verdicts\"}[$__range]))", + "legendFormat": "{{outcome}}", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + } + }, + "overrides": [] + }, + "options": { + "pieType": "donut", + "displayLabels": ["percent"], + "legend": { + "displayMode": "table", + "placement": "right", + "showLegend": true, + "values": ["value"] + }, + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "tooltip": { + "mode": "single", + "sort": "none" + } + } + }, + { + "id": 8, + "type": "table", + "title": "Check results by check", + "description": "Per check and result: match, no_match, unsure or no_answer, over the time range.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 16, + "x": 8, + "y": 12 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (check, result) (increase({__name__=\"hyperhive.review_classifier.checks\"}[$__range]))", + "legendFormat": "", + "instant": true, + "format": "table" + } + ], + "transformations": [ + { + "id": "organize", + "options": { + "excludeByName": { + "Time": true + }, + "renameByName": { + "check": "Check", + "result": "Result", + "Value": "Count" + } + } + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "custom": { + "align": "auto" + } + }, + "overrides": [] + }, + "options": { + "showHeader": true, + "sortBy": [ + { + "displayName": "Check", + "desc": false + } + ] + } + }, + { + "id": 9, + "type": "timeseries", + "title": "Check results over time", + "description": "Classifications by result across every check.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 20 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (result) (increase({__name__=\"hyperhive.review_classifier.checks\"}[$__interval]))", + "legendFormat": "{{result}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "bars", + "lineWidth": 1, + "fillOpacity": 60, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0, + "stacking": { + "mode": "normal", + "group": "A" + } + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, + { + "id": 10, + "type": "timeseries", + "title": "Asks retried", + "description": "Model asks repeated because the first answer failed validation.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 20 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (runtime) (increase({__name__=\"hyperhive.review_classifier.ask.retries\"}[$__interval]))", + "legendFormat": "{{runtime}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "bars", + "lineWidth": 1, + "fillOpacity": 60, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0, + "stacking": { + "mode": "normal", + "group": "A" + } + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, + { + "id": 11, + "type": "timeseries", + "title": "Run duration p50 / p95", + "description": "Time from reading the rules to a posted review, across all outcomes.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 28 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "histogram_quantile(0.5, sum by (le) (rate({__name__=\"hyperhive.review_classifier.run.duration_bucket\"}[$__rate_interval])))", + "legendFormat": "p50" + }, + { + "refId": "B", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "histogram_quantile(0.95, sum by (le) (rate({__name__=\"hyperhive.review_classifier.run.duration_bucket\"}[$__rate_interval])))", + "legendFormat": "p95" + } + ], + "fieldConfig": { + "defaults": { + "unit": "s", + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 10, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0 + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + } + ] +} diff --git a/nix/module-eval/grafana.nix b/nix/module-eval/grafana.nix index bb826163..c60df345 100644 --- a/nix/module-eval/grafana.nix +++ b/nix/module-eval/grafana.nix @@ -382,6 +382,44 @@ let && lib.all (m: lib.hasInfix "\"${m}\"" pushed) queried && lib.any (v: v.name == "repo") board.templating.list; } + { + # The review classifier board ships, and every metric it queries is a name + # the controller or the classifier service records: a renamed metric on + # either side is a board of empty panels. A histogram is queried by its + # `_bucket` series. + name = "the review classifier dashboard ships, and queries the metric names the controller and the classifier record"; + ok = + let + provider = lib.head grafanaWithAuthelia.containers.swarm-grafana.config.services.grafana.provision.dashboards.settings.providers; + shipped = lib.attrNames provider.options.path.entries; + board = builtins.fromJSON ( + builtins.readFile ../host-modules/swarm-grafana/dashboards/review-classifier.json + ); + exprs = lib.concatMap (p: map (t: t.expr or "") (p.targets or [ ])) board.panels; + queried = lib.unique ( + lib.concatMap ( + e: + map (m: lib.removeSuffix "_bucket" (lib.head m)) ( + lib.filter lib.isList (builtins.split ''"(hyperhive\.review_classifier\.[a-z_.]+)"'' e) + ) + ) exprs + ); + recorded = + builtins.readFile ../../swarm-controller/src/review_classifier/metrics.rs + + builtins.readFile ../../swarm-review-classifier/src/cost.rs; + in + lib.elem "review-classifier.json" shipped + && lib.elem "bao.json" shipped + && + lib.sort lib.lessThan queried == [ + "hyperhive.review_classifier.ask.retries" + "hyperhive.review_classifier.checks" + "hyperhive.review_classifier.requests" + "hyperhive.review_classifier.run.duration" + "hyperhive.review_classifier.verdicts" + ] + && lib.all (m: lib.hasInfix "\"${m}\"" recorded) queried; + } { # The standing check on the collector's PRIORITY mapping — whether a line # ever reaches the log store carrying no severity. It is shipped config diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 02b61c7d..f3959680 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -157,6 +157,8 @@ x509-cert.workspace = true tokio = { workspace = true, features = ["test-util"] } # A fresh directory per test `nats-server`, for its store and ports file. tempfile = "3" +# `InMemoryMetricExporter`, to read back what `review_classifier::metrics` records. +opentelemetry_sdk = { workspace = true, features = ["testing"] } # `ServiceExt::oneshot`, to drive a route through the real router in tests. tower = { version = "0.5", features = ["util"] } # A layer that captures `review_classifier::runs`'s log event in a test. diff --git a/swarm-controller/src/review_classifier.rs b/swarm-controller/src/review_classifier.rs index d091d005..03b66278 100644 --- a/swarm-controller/src/review_classifier.rs +++ b/swarm-controller/src/review_classifier.rs @@ -40,6 +40,7 @@ use self::settings::{Settings, SettingsSource}; use crate::forge::{self, DefaultBranchFiles, PullForReview}; use crate::routes::{AppState, error_problem}; +pub(crate) mod metrics; pub(crate) mod open_pr; pub(crate) mod review; pub(crate) mod runs; diff --git a/swarm-controller/src/review_classifier/metrics.rs b/swarm-controller/src/review_classifier/metrics.rs new file mode 100644 index 00000000..7162c16b --- /dev/null +++ b/swarm-controller/src/review_classifier/metrics.rs @@ -0,0 +1,283 @@ +//! OTEL export of what review requests and review runs come to: the counters +//! `hyperhive.review_classifier.requests` (`outcome`), `.verdicts` (`outcome`) +//! and `.checks` (`check`, `result`), and the histogram +//! `hyperhive.review_classifier.run.duration` in seconds (`outcome`). The +//! classifier service exports the retries beside its cost counters, see +//! `swarm-review-classifier`'s `cost` module. +//! +//! Same exporter as [`crate::vcs_metrics`], built on the first record; with no +//! `OTEL_EXPORTER_OTLP_ENDPOINT` every record does nothing. + +use std::sync::{Arc, OnceLock}; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, Histogram, MeterProvider as _}; +use opentelemetry_sdk::metrics::SdkMeterProvider; +use swarm_wire::review_classifier::{Row, Verdict as RowVerdict}; + +use super::runs::RunOutcome; + +/// What became of a review request. +#[derive(Debug, Clone, Copy)] +pub(super) enum Request { + /// Stored in the request queue. + Queued, + /// The queue did not store it; a no-verdict comment was posted. + QueueFailed, + /// Taken from the queue but unreadable, and dropped. + Unparsable, + /// Taken while the PR's review was going; put back in the queue. + Deferred, + /// Taken, and a fresh run began. + Started, + /// Taken for a head already reviewed; its verdict was posted again. + Reposted, +} + +impl Request { + const fn label(self) -> &'static str { + match self { + Self::Queued => "queued", + Self::QueueFailed => "queue_failed", + Self::Unparsable => "unparsable", + Self::Deferred => "deferred", + Self::Started => "started", + Self::Reposted => "reposted", + } + } +} + +/// The review event a run chose. +#[derive(Debug, Clone, Copy)] +pub(super) enum Choice { + Approve, + RequestChanges, + NoVerdict, +} + +impl Choice { + const fn label(self) -> &'static str { + match self { + Self::Approve => "approve", + Self::RequestChanges => "request_changes", + Self::NoVerdict => "no_verdict", + } + } +} + +/// A run's `outcome` label: the verdict names, plus `post_failed`. +const fn run_label(outcome: RunOutcome) -> &'static str { + match outcome { + RunOutcome::Approved => "approve", + RunOutcome::ChangesRequested => "request_changes", + RunOutcome::NoVerdict => "no_verdict", + RunOutcome::PostFailed => "post_failed", + RunOutcome::QueueFull => "queue_full", + RunOutcome::Done => "done", + RunOutcome::Failed => "failed", + } +} + +/// A row's `result` label. +const fn check_result(row: &Row) -> &'static str { + match (row.error.is_some(), row.verdict) { + (true, _) => "no_answer", + (false, RowVerdict::Match) => "match", + (false, RowVerdict::NoMatch) => "no_match", + (false, RowVerdict::NotApplicable) => "unsure", + } +} + +/// Seconds. A run is one model call per classification plus the forge reads +/// and the post. +const DURATION_BOUNDARIES: [f64; 12] = [ + 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0, 90.0, 120.0, 180.0, 300.0, 600.0, +]; + +struct Instruments { + /// Exports only while it lives. + _provider: SdkMeterProvider, + requests: Counter, + verdicts: Counter, + checks: Counter, + duration: Histogram, +} + +impl Instruments { + fn new(provider: &SdkMeterProvider) -> Self { + let meter = provider.meter("hyperhive.review_classifier"); + Self { + _provider: provider.clone(), + requests: meter + .u64_counter("hyperhive.review_classifier.requests") + .build(), + verdicts: meter + .u64_counter("hyperhive.review_classifier.verdicts") + .build(), + checks: meter + .u64_counter("hyperhive.review_classifier.checks") + .build(), + duration: meter + .f64_histogram("hyperhive.review_classifier.run.duration") + .with_unit("s") + .with_boundaries(DURATION_BOUNDARIES.to_vec()) + .build(), + } + } +} + +/// The recorder: a handle on the instruments, or on nothing where no +/// collector is configured. +#[derive(Clone)] +pub(crate) struct Metrics(Option>); + +static GLOBAL: OnceLock>> = OnceLock::new(); + +impl Metrics { + /// The process's recorder, building its exporter on the first call. + pub(crate) fn global() -> Self { + Self(GLOBAL.get_or_init(|| build().map(Arc::new)).clone()) + } + + /// Records nothing. + #[cfg(test)] + pub(super) const fn off() -> Self { + Self(None) + } + + /// Instruments on `provider`. + #[cfg(test)] + pub(super) fn on(provider: &SdkMeterProvider) -> Self { + Self(Some(Arc::new(Instruments::new(provider)))) + } + + pub(super) fn request(&self, request: Request) { + if let Some(inst) = &self.0 { + inst.requests + .add(1, &[KeyValue::new("outcome", request.label())]); + } + } + + pub(super) fn verdict(&self, choice: Choice) { + if let Some(inst) = &self.0 { + inst.verdicts + .add(1, &[KeyValue::new("outcome", choice.label())]); + } + } + + /// One count per classification the run answered or failed to. + pub(super) fn checks(&self, rows: &[Row]) { + let Some(inst) = &self.0 else { return }; + for row in rows { + inst.checks.add( + 1, + &[ + KeyValue::new("check", row.name.clone()), + KeyValue::new("result", check_result(row)), + ], + ); + } + } + + pub(super) fn run_duration(&self, outcome: RunOutcome, seconds: f64) { + if let Some(inst) = &self.0 { + inst.duration + .record(seconds, &[KeyValue::new("outcome", run_label(outcome))]); + } + } +} + +fn build() -> Option { + let endpoint = crate::vcs_metrics::endpoint()?; + let interval = crate::vcs_metrics::EXPORT_INTERVAL; + match crate::vcs_metrics::build_provider(interval) { + Ok(provider) => { + tracing::info!(%endpoint, ?interval, "otel review-classifier metrics: exporter enabled"); + Some(Instruments::new(&provider)) + } + Err(e) => { + tracing::warn!(error = ?e, "otel review-classifier metrics: exporter init failed"); + None + } + } +} + +#[cfg(test)] +pub(super) mod testing { + use std::collections::BTreeMap; + + use opentelemetry_sdk::metrics::data::{ + AggregatedMetrics, MetricData, ResourceMetrics, ScopeMetrics, + }; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader}; + + use super::{Metrics, SdkMeterProvider}; + + /// Instruments on a provider that exports to memory. + pub(in crate::review_classifier) struct Recorder { + pub(in crate::review_classifier) metrics: Metrics, + provider: SdkMeterProvider, + exporter: InMemoryMetricExporter, + } + + impl Recorder { + pub(in crate::review_classifier) fn new() -> Self { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + Self { + metrics: Metrics::on(&provider), + provider, + exporter, + } + } + + /// `read` of the newest export of the metric `name`. + fn export(&self, name: &str, read: impl FnOnce(&AggregatedMetrics) -> R) -> Option { + self.provider.force_flush().expect("flush"); + let all = self.exporter.get_finished_metrics().expect("metrics"); + all.last() + .into_iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .find(|m| m.name() == name) + .map(|m| read(m.data())) + } + + /// The counter `name`'s value per label set, each set as + /// `key=value,key=value` in key order. Empty when nothing was added. + pub(in crate::review_classifier) fn counter(&self, name: &str) -> BTreeMap { + self.export(name, |data| { + let AggregatedMetrics::U64(MetricData::Sum(sum)) = data else { + panic!("not a u64 sum: {data:?}"); + }; + sum.data_points() + .map(|p| (labels(p.attributes()), p.value())) + .collect() + }) + .unwrap_or_default() + } + + /// The histogram `name`'s sample count per label set, as + /// [`Self::counter`] writes them. + pub(in crate::review_classifier) fn histogram(&self, name: &str) -> BTreeMap { + self.export(name, |data| { + let AggregatedMetrics::F64(MetricData::Histogram(h)) = data else { + panic!("not an f64 histogram: {data:?}"); + }; + h.data_points() + .map(|p| (labels(p.attributes()), p.count())) + .collect() + }) + .unwrap_or_default() + } + } + + fn labels<'a>(attributes: impl Iterator) -> String { + let mut all: Vec = attributes + .map(|kv| format!("{}={}", kv.key, kv.value)) + .collect(); + all.sort(); + all.join(",") + } +} diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 343e6ead..f99095ad 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -27,7 +27,7 @@ use std::collections::HashSet; use std::fmt::Write as _; use std::future::Future; use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::Context as _; use async_nats::jetstream::{self, AckKind, consumer::PullConsumer}; @@ -40,6 +40,7 @@ use swarm_wire::review_classifier::{ use tokio::sync::Notify; use tokio::task::JoinSet; +use super::metrics::{Choice, Metrics, Request}; use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger}; use super::settings::{Settings, SettingsSource}; use super::{FileSource, PullSource, RunSink, split_repo}; @@ -101,6 +102,15 @@ impl Event { } } +/// The metrics label of the review `event` a run chose. +const fn verdict_label(event: Event) -> Choice { + match event { + Event::Approve => Choice::Approve, + Event::RequestChanges => Choice::RequestChanges, + Event::Comment => Choice::NoVerdict, + } +} + /// What a run posts when it has no clean verdict: the rules didn't load, the /// PR couldn't be read, the run failed, or any classification has no answer. /// A comment neither approves nor blocks the merge. @@ -328,7 +338,8 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) { }; let client = state.status.as_ref().map(|s| s.queue_client()); let runs = state.runs.clone(); - tokio::spawn(async move { accept(client.as_ref(), &*forge, &runs, &request).await }); + let metrics = Metrics::global(); + tokio::spawn(async move { accept(client.as_ref(), &*forge, &runs, &metrics, &request).await }); } /// Put `request` in [`REQUEST_STREAM`]. When that fails, the account posts a @@ -337,6 +348,7 @@ async fn accept( client: Option<&async_nats::Client>, poster: &impl ReviewPoster, runs: &Runs, + metrics: &Metrics, request: &Requested, ) { let repo = format!("{}/{}", request.owner, request.name); @@ -346,6 +358,7 @@ async fn accept( }; let error = match queued { Ok(()) => { + metrics.request(Request::Queued); tracing::info!( %repo, pr = request.pr, head_sha = request.head.as_str(), "review request queued" @@ -354,6 +367,8 @@ async fn accept( } Err(e) => format!("{e:#}"), }; + metrics.request(Request::QueueFailed); + metrics.verdict(Choice::NoVerdict); tracing::warn!( %repo, pr = request.pr, head_sha = request.head.as_str(), %error, "queueing a review request failed; posting a no-verdict comment" @@ -448,6 +463,7 @@ pub(crate) fn spawn_queue(state: &AppState) { queue, poster: forge, runs, + metrics: Metrics::global(), }; consume(&client, TIMING, &reviews, handles).await; }); @@ -461,6 +477,7 @@ struct Handles { queue: Q, poster: Arc

, runs: Runs, + metrics: Metrics, } impl Clone for Handles { @@ -470,6 +487,7 @@ impl Clone for Handles { queue: self.queue.clone(), poster: Arc::clone(&self.poster), runs: self.runs.clone(), + metrics: self.metrics.clone(), } } } @@ -580,6 +598,7 @@ async fn take( subject = %message.subject, error = %e, "a queued review request did not parse; dropping it" ); + handles.metrics.request(Request::Unparsable); if let Err(e) = message.ack_with(AckKind::Term).await { tracing::warn!(error = %e, "dropping an unparsable review request failed"); } @@ -588,6 +607,7 @@ async fn take( }; let Some(claim) = reviews.claim(&request) else { drop(slot); + handles.metrics.request(Request::Deferred); tracing::info!( repo = %format!("{}/{}", request.owner, request.name), pr = request.pr, head_sha = request.head.as_str(), @@ -606,8 +626,9 @@ async fn take( queue, poster, runs, + metrics, } = &handles; - let work = review(&**forge, &**forge, queue, &**poster, runs, &request, slot); + let work = review(&**forge, queue, &**poster, runs, metrics, &request, slot); in_progress(&message, timing.ack_wait / 3, work).await; drop(claim); if let Err(e) = message.ack().await { @@ -648,15 +669,15 @@ async fn in_progress( /// effort: a request for a head the PR has moved on from gets a review of the /// new head, never the old head's verdict. async fn review( - files: &impl FileSource, - pulls: &impl PullSource, + forge: &(impl FileSource + PullSource), sink: &(impl RunSink + SettingsSource), poster: &impl ReviewPoster, runs: &Runs, + metrics: &Metrics, req: &Requested, slot: Slot, ) { - let pull = read_pull(pulls, req).await; + let pull = read_pull(forge, req).await; let moved = matches!(&pull, Ok(pull) if !req.head.is_empty() && pull.head_sha != req.head); if moved { tracing::info!( @@ -670,11 +691,14 @@ async fn review( && let Some(ruling) = reposted(&earlier) { drop(slot); + metrics.request(Request::Reposted); + metrics.verdict(verdict_label(ruling.event)); runs.record(repost(poster, req, &earlier, ruling).await) .await; return; } - let run = Box::pin(run(files, pull, sink, poster, req)).await; + metrics.request(Request::Started); + let run = Box::pin(run(forge, pull, sink, poster, metrics, req)).await; drop(slot); runs.record(run).await; } @@ -840,11 +864,21 @@ async fn run( pull: Result, sink: &(impl RunSink + SettingsSource), poster: &impl ReviewPoster, + metrics: &Metrics, req: &Requested, ) -> RunRecord { let started_unix = unix_now(); + let timer = Instant::now(); let (head, narrowed, outcome) = classify(files, pull, sink, req).await; let mut ruling = review_for(&outcome); + metrics.verdict(verdict_label(ruling.event)); + if let Outcome::Ran { + reply: Reply::Done { rows, .. }, + .. + } = &outcome + { + metrics.checks(rows); + } if let (Some(reason), Outcome::Ran { .. }) = (narrowed, &outcome) { let _ = write!( ruling.body, @@ -863,13 +897,9 @@ async fn run( Outcome::Ran { reply, .. } => runs::answer(reply), Outcome::NotRun(_) | Outcome::NoRules { .. } => (None, None), }; - record( - req, - head, - started_unix, - ended(posted_as, reason, sent), - answer, - ) + let ended = ended(posted_as, reason, sent); + metrics.run_duration(ended.outcome, timer.elapsed().as_secs_f64()); + record(req, head, started_unix, ended, answer) } /// Post `review` on `req`'s PR, naming `head` when there is one. The posted @@ -1278,6 +1308,7 @@ mod tests { use swarm_wire::review_classifier::Polarity; use super::*; + use crate::review_classifier::metrics::testing::Recorder; use crate::review_classifier::runs::tests::FakeDoc; use crate::review_classifier::tests::{Fake, rules_json}; use crate::test_nats::TestNats; @@ -1433,7 +1464,15 @@ mod tests { poster: &impl ReviewPoster, req: &Requested, ) -> RunRecord { - run(forge, read_pull(forge, req).await, sink, poster, req).await + run( + forge, + read_pull(forge, req).await, + sink, + poster, + &Metrics::off(), + req, + ) + .await } /// A JSON rules file holding `n` classifications. @@ -1701,7 +1740,16 @@ mod tests { let request = requested_at(head); tokio::spawn(async move { let slot = reviews.slot().await; - review(&*forge, &*forge, &gate, &*poster, &store, &request, slot).await; + review( + &*forge, + &gate, + &*poster, + &store, + &Metrics::off(), + &request, + slot, + ) + .await; }) } @@ -1856,6 +1904,7 @@ mod tests { _nats: TestNats, client: async_nats::Client, rig: Rig, + recorder: Recorder, } impl Queued { @@ -1866,6 +1915,7 @@ mod tests { _nats: nats, client, rig: Rig::new(), + recorder: Recorder::new(), } } @@ -1879,6 +1929,7 @@ mod tests { Some(&self.client), &*self.rig.poster, &self.rig.store, + &self.recorder.metrics, &request, ) .await; @@ -1893,12 +1944,14 @@ mod tests { let gate = self.rig.gate.clone(); let poster = Arc::clone(&self.rig.poster); let store = self.rig.store.clone(); + let metrics = self.recorder.metrics.clone(); tokio::spawn(async move { let handles = Handles { forge, queue: gate, poster, runs: store, + metrics, }; consume(&client, TEST_TIMING, &reviews, handles).await; }) @@ -2033,6 +2086,7 @@ mod tests { Some(&q.client), &*q.rig.poster, &q.rig.store, + &Metrics::off(), &requested_at("old"), ) .await; @@ -2104,6 +2158,14 @@ mod tests { "{}", posted[1].2.body ); + let requests = q.recorder.counter(REQUESTS); + assert_eq!(requests.get("outcome=queued"), Some(&2), "{requests:?}"); + assert_eq!(requests.get("outcome=started"), Some(&1), "{requests:?}"); + assert_eq!(requests.get("outcome=reposted"), Some(&1), "{requests:?}"); + assert!( + requests.get("outcome=deferred").is_some_and(|n| *n >= 1), + "the second request waited at least once: {requests:?}" + ); assert_eq!(q.rig.gate.reached(), 1); } @@ -2132,12 +2194,21 @@ mod tests { .expect("both leave the queue"); assert_eq!(q.rig.posted()[0].0, "o/r#8"); assert_eq!(q.rig.gate.reached(), 1); + let requests = q.recorder.counter(REQUESTS); + assert_eq!(requests.get("outcome=unparsable"), Some(&1), "{requests:?}"); } #[tokio::test] async fn a_request_that_cannot_be_queued_posts_a_no_verdict_comment() { let rig = Rig::new(); - accept(None, &*rig.poster, &rig.store, &requested_at("abc123")).await; + accept( + None, + &*rig.poster, + &rig.store, + &Metrics::off(), + &requested_at("abc123"), + ) + .await; let posted = rig.posted(); assert_eq!(posted.len(), 1); let (target, commit, review) = &posted[0]; @@ -2960,4 +3031,211 @@ mod tests { "a run that never ran read no diff: {body}" ); } + + /// Row `name` as the classifier answers it: `verdict`, counted and passed + /// when it matched. + fn named(name: &str, verdict: Verdict, error: Option<&str>) -> Row { + Row { + name: name.to_owned(), + verdict, + ..row(error, error.is_none(), error.is_none()) + } + } + + const REQUESTS: &str = "hyperhive.review_classifier.requests"; + const VERDICTS: &str = "hyperhive.review_classifier.verdicts"; + const CHECKS: &str = "hyperhive.review_classifier.checks"; + const DURATION: &str = "hyperhive.review_classifier.run.duration"; + + /// `pairs` as a count per label set. + fn counts(pairs: [(&str, u64); N]) -> std::collections::BTreeMap { + pairs.into_iter().map(|(k, v)| (k.to_owned(), v)).collect() + } + + /// One review run per reply, each on its own PR head, then a repeat of the + /// last: the metrics name and label what each did. + #[tokio::test] + async fn a_review_records_its_request_verdict_checks_and_duration() { + let recorder = Recorder::new(); + let metrics = &recorder.metrics; + let reviews = Arc::new(Reviews::default()); + let store = Runs::Fake(Arc::default()); + let poster = Posted::default(); + + // A request made at a head the PR has moved on from: reviewed at the + // current head. Four rules, one answer of each kind: no verdict. + let moved = Requested { + head: "def456".to_owned(), + ..request("r") + }; + let rows = vec![ + named("c0", Verdict::Match, None), + named("c1", Verdict::NoMatch, None), + named("c2", Verdict::NotApplicable, None), + named("c3", Verdict::NotApplicable, Some("timed out")), + ]; + let mixed = Fake::new( + Reply::Done { + rows, + overall: Overall::Approve, + }, + vec![rules_of(4)], + ); + review( + &mixed, + &mixed, + &poster, + &store, + metrics, + &moved, + reviews.slot().await, + ) + .await; + + // One rule that matched: the run approves, then a repeat of the + // request posts the same verdict again without a run. + let clean = Fake::new( + Reply::Done { + rows: vec![named("c0", Verdict::Match, None)], + overall: Overall::Approve, + }, + vec![rules_of(1)], + ); + for _ in 0..2 { + review( + &clean, + &clean, + &poster, + &store, + metrics, + &request("r"), + reviews.slot().await, + ) + .await; + } + + // The queue is not configured: the request is refused with a comment. + accept(None, &poster, &store, metrics, &request("r")).await; + + assert_eq!( + recorder.counter(REQUESTS), + counts([ + ("outcome=queue_failed", 1), + ("outcome=reposted", 1), + ("outcome=started", 2), + ]) + ); + assert_eq!( + recorder.counter(VERDICTS), + counts([("outcome=approve", 2), ("outcome=no_verdict", 2)]) + ); + assert_eq!( + recorder.counter(CHECKS), + counts([ + ("check=c0,result=match", 2), + ("check=c1,result=no_match", 1), + ("check=c2,result=unsure", 1), + ("check=c3,result=no_answer", 1), + ]) + ); + assert_eq!( + recorder.histogram(DURATION), + counts([("outcome=approve", 1), ("outcome=no_verdict", 1)]) + ); + } + + /// A run whose rows ask for changes records that verdict, and one whose + /// post the forge refuses records its duration as `post_failed`. + #[tokio::test] + async fn a_run_records_request_changes_and_a_failed_post() { + let recorder = Recorder::new(); + let metrics = &recorder.metrics; + let reply = Reply::Done { + rows: vec![named("c0", Verdict::NoMatch, None)], + overall: Overall::RequestChanges, + }; + let fake = Fake::new(reply, vec![rules_of(1)]); + run( + &fake, + read_pull(&fake, &request("r")).await, + &fake, + &Posted::default(), + metrics, + &request("r"), + ) + .await; + run( + &fake, + read_pull(&fake, &request("r")).await, + &fake, + &Refusing, + metrics, + &request("r"), + ) + .await; + + assert_eq!( + recorder.counter(VERDICTS), + counts([("outcome=request_changes", 2)]) + ); + assert_eq!( + recorder.histogram(DURATION), + counts([("outcome=post_failed", 1), ("outcome=request_changes", 1)]) + ); + } + + /// A repo without rules approves with no checks: it counts as an approve + /// verdict and an approved run, and adds no `checks` row. + #[tokio::test] + async fn a_repo_without_rules_records_an_approve_and_no_checks() { + let recorder = Recorder::new(); + let metrics = &recorder.metrics; + let none = Fake::new( + Reply::Failed { + error: "must not run".to_owned(), + }, + Vec::new(), + ); + run( + &none, + read_pull(&none, &request("r")).await, + &none, + &Posted::default(), + metrics, + &request("r"), + ) + .await; + + assert_eq!(recorder.counter(VERDICTS), counts([("outcome=approve", 1)])); + assert_eq!( + recorder.histogram(DURATION), + counts([("outcome=approve", 1)]) + ); + assert!(recorder.counter(CHECKS).is_empty()); + + // Control: a repo with a rule records its check, so the empty + // `checks` above is the absence of rules and not a lookup that finds + // nothing. + let ruled = Fake::new( + Reply::Done { + rows: vec![named("c0", Verdict::Match, None)], + overall: Overall::Approve, + }, + vec![rules_of(1)], + ); + run( + &ruled, + read_pull(&ruled, &request("r")).await, + &ruled, + &Posted::default(), + metrics, + &request("r"), + ) + .await; + assert_eq!( + recorder.counter(CHECKS), + counts([("check=c0,result=match", 1)]) + ); + assert_eq!(recorder.counter(VERDICTS), counts([("outcome=approve", 2)])); + } } diff --git a/swarm-controller/src/vcs_metrics.rs b/swarm-controller/src/vcs_metrics.rs index e927b9c2..238f4307 100644 --- a/swarm-controller/src/vcs_metrics.rs +++ b/swarm-controller/src/vcs_metrics.rs @@ -30,7 +30,7 @@ use opentelemetry_sdk::Resource; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; /// Export cadence. Matches every other OTEL exporter in this tree. -const EXPORT_INTERVAL: Duration = Duration::from_mins(1); +pub(crate) const EXPORT_INTERVAL: Duration = Duration::from_mins(1); /// One push delivery, as parsed off the webhook body — just the two fields /// this module needs, not a full mirror of Forgejo's push-event schema. @@ -109,7 +109,7 @@ fn build() -> Option { } } -fn build_provider(interval: Duration) -> Result { +pub(crate) fn build_provider(interval: Duration) -> Result { // Same http/json, endpoint-from-env-only construction as // `hive_jobq_metrics::build_provider` — see that module's doc comment // for why `with_endpoint` is deliberately never called here. @@ -190,7 +190,7 @@ fn resource() -> Resource { builder.build() } -fn endpoint() -> Option { +pub(crate) fn endpoint() -> Option { std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") .ok() .map(|s| s.trim().to_owned()) diff --git a/swarm-review-classifier/src/cost.rs b/swarm-review-classifier/src/cost.rs index 8b18f351..2db04f56 100644 --- a/swarm-review-classifier/src/cost.rs +++ b/swarm-review-classifier/src/cost.rs @@ -7,6 +7,10 @@ //! `claude` exports for itself on the claude runtime, so the cache hit rate is //! `cacheRead` over the three input types on either runtime. //! +//! The counter `hyperhive.review_classifier.ask.retries` counts the asks +//! repeated because an answer did not validate, with the run's `runtime` +//! (`api_key` or `acp`); it is exported on either runtime. +//! //! The amount is computed here from the prices the run carries and the //! tokens the agent reports, so it does not depend on how, or whether, the //! agent prices a custom provider. @@ -23,7 +27,7 @@ use opentelemetry::KeyValue; use opentelemetry::metrics::{Counter, MeterProvider as _}; use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; -use swarm_wire::review_classifier::Prices; +use swarm_wire::review_classifier::{Prices, Runtime}; /// The currency every price is in. pub const CURRENCY: &str = "USD"; @@ -68,7 +72,6 @@ pub struct Otel; impl CostSink for Otel { fn record(&self, cost: &Cost) { - static COUNTERS: OnceLock> = OnceLock::new(); let Some((_, counters)) = COUNTERS.get_or_init(build) else { return; }; @@ -76,10 +79,21 @@ impl CostSink for Otel { } } -/// The cost counter, and the token counter beside it. +/// Counts one repeated ask on `runtime`'s model. +pub fn ask_retried(runtime: &Runtime) { + let Some((_, counters)) = COUNTERS.get_or_init(build) else { + return; + }; + add_retry(counters, runtime); +} + +static COUNTERS: OnceLock> = OnceLock::new(); + +/// The cost counter, the token counter beside it, and the retry counter. struct Counters { cost: Counter, tokens: Counter, + retries: Counter, } /// The counters on `provider`. @@ -92,9 +106,21 @@ fn counters(provider: &SdkMeterProvider) -> Counters { tokens: meter .u64_counter("hyperhive.review_classifier.token.usage") .build(), + retries: meter + .u64_counter("hyperhive.review_classifier.ask.retries") + .build(), } } +/// Adds one retry to `counters`, labelled with `runtime`. +fn add_retry(counters: &Counters, runtime: &Runtime) { + let name = match runtime { + Runtime::ApiKey => "api_key", + Runtime::Acp(_) => "acp", + }; + counters.retries.add(1, &[KeyValue::new("runtime", name)]); +} + /// Adds `cost` to `counters` with the run's labels. fn add(counters: &Counters, cost: &Cost) { let labels = [ @@ -293,4 +319,34 @@ mod tests { ] ); } + + #[test] + fn a_retry_adds_one_under_the_counter_name_and_the_runtime() { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let counters = counters(&provider); + add_retry(&counters, &Runtime::ApiKey); + add_retry(&counters, &Runtime::ApiKey); + provider.force_flush().expect("flush"); + + let metrics = exporter.get_finished_metrics().expect("metrics"); + let metric = metrics + .iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .find(|m| m.name() == "hyperhive.review_classifier.ask.retries") + .expect("the retry counter was exported"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("not a u64 sum: {:?}", metric.data()); + }; + let points: Vec<_> = sum.data_points().collect(); + assert_eq!(points.len(), 1); + assert_eq!(points[0].value(), 2); + let labels: Vec<_> = points[0] + .attributes() + .map(|kv| (kv.key.to_string(), kv.value.to_string())) + .collect(); + assert_eq!(labels, [("runtime".to_owned(), "api_key".to_owned())]); + } } diff --git a/swarm-review-classifier/src/run.rs b/swarm-review-classifier/src/run.rs index b2470678..994c7eab 100644 --- a/swarm-review-classifier/src/run.rs +++ b/swarm-review-classifier/src/run.rs @@ -22,7 +22,7 @@ use swarm_wire::agent_inference::Effort; use swarm_wire::review_classifier::{CLASSIFICATION_DEADLINE, Reply, Row, Runtime, TryOut}; use crate::prompt::Prompt; -use crate::{aggregate, answer, prompt}; +use crate::{aggregate, answer, cost, prompt}; /// How long `claude` may print nothing before it is stopped. const IDLE_TIMEOUT: Duration = Duration::from_mins(2); @@ -69,6 +69,7 @@ pub async fn answer(keys: &impl KeySource, model: &impl Model, run: &TryOut) -> run, CLASSIFICATION_DEADLINE, RETRY_FLOOR, + || cost::ask_retried(&run.runtime), ) .await; rows.push(aggregate::row(c, outcome)); @@ -96,7 +97,7 @@ const RETRY_FLOOR: Duration = Duration::from_secs(20); /// answer that does not validate is asked for once more, with a reminder to /// reply with the JSON object alone, and only the time the first ask left of /// `budget` to do it in; with less than `floor` left, or after a failed ask, -/// it is not asked again. +/// it is not asked again. `on_retry` runs just before the second ask. async fn classify( model: &impl Model, prompt: &Prompt<'_>, @@ -104,6 +105,7 @@ async fn classify( run: &TryOut, budget: Duration, floor: Duration, + on_retry: impl FnOnce(), ) -> Result { let started = Instant::now(); let first = model.ask(prompt, key, run, budget).await?; @@ -115,6 +117,7 @@ async fn classify( if left < floor { return Err(unparsed); } + on_retry(); let again = model .ask(&prompt.with_json_reminder(), key, run, left) .await?; @@ -643,9 +646,27 @@ mod tests { const FLOOR: Duration = Duration::from_millis(100); async fn classify_slow(model: &Slow) -> Result { + classify_counting(model).await.0 + } + + /// The result, and how often `on_retry` ran. + async fn classify_counting(model: &Slow) -> (Result, usize) { let run = run(&["c"]); let prompt = Prompt::of("q"); - classify(model, &prompt, "k", &run, BUDGET, FLOOR).await + let retries = std::cell::Cell::new(0); + let result = classify(model, &prompt, "k", &run, BUDGET, FLOOR, || { + retries.set(retries.get() + 1); + }) + .await; + (result, retries.get()) + } + + #[tokio::test] + async fn a_retry_is_counted_and_a_skipped_one_is_not() { + let (_, retries) = classify_counting(&Slow::new(Duration::from_millis(200))).await; + assert_eq!(retries, 1); + let (_, retries) = classify_counting(&Slow::new(BUDGET)).await; + assert_eq!(retries, 0, "no second ask, so no retry"); } #[tokio::test]