Watch
0
0
Fork
You've already forked hyperhive
0

review classifier: export request, verdict, check and run metrics, with a dashboard

The controller counts review requests, verdicts and per-check results and times each run; the classifier service counts retried asks. A provisioned Grafana dashboard charts them.
This commit is contained in:
atlas 2026-10-24 18:50:03 +02:00 • committed by mara
commit eaf26aaadd
13 changed files with 1390 additions and 28 deletions

View file

@ -522,6 +522,38 @@ at warn, with the kind of failure (`token`, `transport` or `status <code>`)
and the start of the response body. It repeats a persisting failure every 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. 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`) ## Agent-emitted custom metrics (`hive-metric`)
Agents can push arbitrary labeled metrics to the same OTEL collector via the Agents can push arbitrary labeled metrics to the same OTEL collector via the

View file

@ -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` runtimes, so the cache hit rate reads the same way on every runtime: `cacheRead`
over the sum of `input`, `cacheRead` and `cacheCreation`. 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 The classifier logs each run's result to its journal, which ships to
VictoriaLogs like every swarm service container's VictoriaLogs like every swarm service container's
([`services.md`](services.md#logs-victorialogs)). ([`services.md`](services.md#logs-victorialogs)).

View file

@ -207,14 +207,16 @@ kept.
- `authelia`, `bao`, `queue`: the identity provider, the secret store and the NATS queue. - `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. - `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. - `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. - `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. - `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. - `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. - `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 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 counters on `controller` appear after the first push webhook, the refused
and failure series on `collector` appear once something fails. 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. **Where the data comes from.** The swarm's OTEL collector, below.

View file

@ -278,6 +278,7 @@ let
"logstore.json" "logstore.json"
"metricstore.json" "metricstore.json"
"queue.json" "queue.json"
"review-classifier.json"
] ]
); );

View file

@ -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"
}
}
}
]
}

View file

@ -382,6 +382,44 @@ let
&& lib.all (m: lib.hasInfix "\"${m}\"" pushed) queried && lib.all (m: lib.hasInfix "\"${m}\"" pushed) queried
&& lib.any (v: v.name == "repo") board.templating.list; && 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 # The standing check on the collector's PRIORITY mapping — whether a line
# ever reaches the log store carrying no severity. It is shipped config # ever reaches the log store carrying no severity. It is shipped config

View file

@ -157,6 +157,8 @@ x509-cert.workspace = true
tokio = { workspace = true, features = ["test-util"] } tokio = { workspace = true, features = ["test-util"] }
# A fresh directory per test `nats-server`, for its store and ports file. # A fresh directory per test `nats-server`, for its store and ports file.
tempfile = "3" 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. # `ServiceExt::oneshot`, to drive a route through the real router in tests.
tower = { version = "0.5", features = ["util"] } tower = { version = "0.5", features = ["util"] }
# A layer that captures `review_classifier::runs`'s log event in a test. # A layer that captures `review_classifier::runs`'s log event in a test.

View file

@ -40,6 +40,7 @@ use self::settings::{Settings, SettingsSource};
use crate::forge::{self, DefaultBranchFiles, PullForReview}; use crate::forge::{self, DefaultBranchFiles, PullForReview};
use crate::routes::{AppState, error_problem}; use crate::routes::{AppState, error_problem};
pub(crate) mod metrics;
pub(crate) mod open_pr; pub(crate) mod open_pr;
pub(crate) mod review; pub(crate) mod review;
pub(crate) mod runs; pub(crate) mod runs;

View file

@ -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<u64>,
verdicts: Counter<u64>,
checks: Counter<u64>,
duration: Histogram<f64>,
}
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<Arc<Instruments>>);
static GLOBAL: OnceLock<Option<Arc<Instruments>>> = 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<Instruments> {
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<R>(&self, name: &str, read: impl FnOnce(&AggregatedMetrics) -> R) -> Option<R> {
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<String, u64> {
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<String, u64> {
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<Item = &'a opentelemetry::KeyValue>) -> String {
let mut all: Vec<String> = attributes
.map(|kv| format!("{}={}", kv.key, kv.value))
.collect();
all.sort();
all.join(",")
}
}

View file

@ -27,7 +27,7 @@ use std::collections::HashSet;
use std::fmt::Write as _; use std::fmt::Write as _;
use std::future::Future; use std::future::Future;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration; use std::time::{Duration, Instant};
use anyhow::Context as _; use anyhow::Context as _;
use async_nats::jetstream::{self, AckKind, consumer::PullConsumer}; use async_nats::jetstream::{self, AckKind, consumer::PullConsumer};
@ -40,6 +40,7 @@ use swarm_wire::review_classifier::{
use tokio::sync::Notify; use tokio::sync::Notify;
use tokio::task::JoinSet; use tokio::task::JoinSet;
use super::metrics::{Choice, Metrics, Request};
use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger}; use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger};
use super::settings::{Settings, SettingsSource}; use super::settings::{Settings, SettingsSource};
use super::{FileSource, PullSource, RunSink, split_repo}; 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 /// 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. /// PR couldn't be read, the run failed, or any classification has no answer.
/// A comment neither approves nor blocks the merge. /// 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 client = state.status.as_ref().map(|s| s.queue_client());
let runs = state.runs.clone(); 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 /// Put `request` in [`REQUEST_STREAM`]. When that fails, the account posts a
@ -337,6 +348,7 @@ async fn accept(
client: Option<&async_nats::Client>, client: Option<&async_nats::Client>,
poster: &impl ReviewPoster, poster: &impl ReviewPoster,
runs: &Runs, runs: &Runs,
metrics: &Metrics,
request: &Requested, request: &Requested,
) { ) {
let repo = format!("{}/{}", request.owner, request.name); let repo = format!("{}/{}", request.owner, request.name);
@ -346,6 +358,7 @@ async fn accept(
}; };
let error = match queued { let error = match queued {
Ok(()) => { Ok(()) => {
metrics.request(Request::Queued);
tracing::info!( tracing::info!(
%repo, pr = request.pr, head_sha = request.head.as_str(), %repo, pr = request.pr, head_sha = request.head.as_str(),
"review request queued" "review request queued"
@ -354,6 +367,8 @@ async fn accept(
} }
Err(e) => format!("{e:#}"), Err(e) => format!("{e:#}"),
}; };
metrics.request(Request::QueueFailed);
metrics.verdict(Choice::NoVerdict);
tracing::warn!( tracing::warn!(
%repo, pr = request.pr, head_sha = request.head.as_str(), %error, %repo, pr = request.pr, head_sha = request.head.as_str(), %error,
"queueing a review request failed; posting a no-verdict comment" "queueing a review request failed; posting a no-verdict comment"
@ -448,6 +463,7 @@ pub(crate) fn spawn_queue(state: &AppState) {
queue, queue,
poster: forge, poster: forge,
runs, runs,
metrics: Metrics::global(),
}; };
consume(&client, TIMING, &reviews, handles).await; consume(&client, TIMING, &reviews, handles).await;
}); });
@ -461,6 +477,7 @@ struct Handles<S, Q, P> {
queue: Q, queue: Q,
poster: Arc<P>, poster: Arc<P>,
runs: Runs, runs: Runs,
metrics: Metrics,
} }
impl<S, Q: Clone, P> Clone for Handles<S, Q, P> { impl<S, Q: Clone, P> Clone for Handles<S, Q, P> {
@ -470,6 +487,7 @@ impl<S, Q: Clone, P> Clone for Handles<S, Q, P> {
queue: self.queue.clone(), queue: self.queue.clone(),
poster: Arc::clone(&self.poster), poster: Arc::clone(&self.poster),
runs: self.runs.clone(), runs: self.runs.clone(),
metrics: self.metrics.clone(),
} }
} }
} }
@ -580,6 +598,7 @@ async fn take<S, Q, P>(
subject = %message.subject, error = %e, subject = %message.subject, error = %e,
"a queued review request did not parse; dropping it" "a queued review request did not parse; dropping it"
); );
handles.metrics.request(Request::Unparsable);
if let Err(e) = message.ack_with(AckKind::Term).await { if let Err(e) = message.ack_with(AckKind::Term).await {
tracing::warn!(error = %e, "dropping an unparsable review request failed"); tracing::warn!(error = %e, "dropping an unparsable review request failed");
} }
@ -588,6 +607,7 @@ async fn take<S, Q, P>(
}; };
let Some(claim) = reviews.claim(&request) else { let Some(claim) = reviews.claim(&request) else {
drop(slot); drop(slot);
handles.metrics.request(Request::Deferred);
tracing::info!( tracing::info!(
repo = %format!("{}/{}", request.owner, request.name), pr = request.pr, repo = %format!("{}/{}", request.owner, request.name), pr = request.pr,
head_sha = request.head.as_str(), head_sha = request.head.as_str(),
@ -606,8 +626,9 @@ async fn take<S, Q, P>(
queue, queue,
poster, poster,
runs, runs,
metrics,
} = &handles; } = &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; in_progress(&message, timing.ack_wait / 3, work).await;
drop(claim); drop(claim);
if let Err(e) = message.ack().await { 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 /// 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. /// new head, never the old head's verdict.
async fn review( async fn review(
files: &impl FileSource, forge: &(impl FileSource + PullSource),
pulls: &impl PullSource,
sink: &(impl RunSink + SettingsSource), sink: &(impl RunSink + SettingsSource),
poster: &impl ReviewPoster, poster: &impl ReviewPoster,
runs: &Runs, runs: &Runs,
metrics: &Metrics,
req: &Requested, req: &Requested,
slot: Slot, 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); let moved = matches!(&pull, Ok(pull) if !req.head.is_empty() && pull.head_sha != req.head);
if moved { if moved {
tracing::info!( tracing::info!(
@ -670,11 +691,14 @@ async fn review(
&& let Some(ruling) = reposted(&earlier) && let Some(ruling) = reposted(&earlier)
{ {
drop(slot); drop(slot);
metrics.request(Request::Reposted);
metrics.verdict(verdict_label(ruling.event));
runs.record(repost(poster, req, &earlier, ruling).await) runs.record(repost(poster, req, &earlier, ruling).await)
.await; .await;
return; 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); drop(slot);
runs.record(run).await; runs.record(run).await;
} }
@ -840,11 +864,21 @@ async fn run(
pull: Result<PullForReview, String>, pull: Result<PullForReview, String>,
sink: &(impl RunSink + SettingsSource), sink: &(impl RunSink + SettingsSource),
poster: &impl ReviewPoster, poster: &impl ReviewPoster,
metrics: &Metrics,
req: &Requested, req: &Requested,
) -> RunRecord { ) -> RunRecord {
let started_unix = unix_now(); let started_unix = unix_now();
let timer = Instant::now();
let (head, narrowed, outcome) = classify(files, pull, sink, req).await; let (head, narrowed, outcome) = classify(files, pull, sink, req).await;
let mut ruling = review_for(&outcome); 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) { if let (Some(reason), Outcome::Ran { .. }) = (narrowed, &outcome) {
let _ = write!( let _ = write!(
ruling.body, ruling.body,
@ -863,13 +897,9 @@ async fn run(
Outcome::Ran { reply, .. } => runs::answer(reply), Outcome::Ran { reply, .. } => runs::answer(reply),
Outcome::NotRun(_) | Outcome::NoRules { .. } => (None, None), Outcome::NotRun(_) | Outcome::NoRules { .. } => (None, None),
}; };
record( let ended = ended(posted_as, reason, sent);
req, metrics.run_duration(ended.outcome, timer.elapsed().as_secs_f64());
head, record(req, head, started_unix, ended, answer)
started_unix,
ended(posted_as, reason, sent),
answer,
)
} }
/// Post `review` on `req`'s PR, naming `head` when there is one. The posted /// 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 swarm_wire::review_classifier::Polarity;
use super::*; use super::*;
use crate::review_classifier::metrics::testing::Recorder;
use crate::review_classifier::runs::tests::FakeDoc; use crate::review_classifier::runs::tests::FakeDoc;
use crate::review_classifier::tests::{Fake, rules_json}; use crate::review_classifier::tests::{Fake, rules_json};
use crate::test_nats::TestNats; use crate::test_nats::TestNats;
@ -1433,7 +1464,15 @@ mod tests {
poster: &impl ReviewPoster, poster: &impl ReviewPoster,
req: &Requested, req: &Requested,
) -> RunRecord { ) -> 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. /// A JSON rules file holding `n` classifications.
@ -1701,7 +1740,16 @@ mod tests {
let request = requested_at(head); let request = requested_at(head);
tokio::spawn(async move { tokio::spawn(async move {
let slot = reviews.slot().await; 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, _nats: TestNats,
client: async_nats::Client, client: async_nats::Client,
rig: Rig, rig: Rig,
recorder: Recorder,
} }
impl Queued { impl Queued {
@ -1866,6 +1915,7 @@ mod tests {
_nats: nats, _nats: nats,
client, client,
rig: Rig::new(), rig: Rig::new(),
recorder: Recorder::new(),
} }
} }
@ -1879,6 +1929,7 @@ mod tests {
Some(&self.client), Some(&self.client),
&*self.rig.poster, &*self.rig.poster,
&self.rig.store, &self.rig.store,
&self.recorder.metrics,
&request, &request,
) )
.await; .await;
@ -1893,12 +1944,14 @@ mod tests {
let gate = self.rig.gate.clone(); let gate = self.rig.gate.clone();
let poster = Arc::clone(&self.rig.poster); let poster = Arc::clone(&self.rig.poster);
let store = self.rig.store.clone(); let store = self.rig.store.clone();
let metrics = self.recorder.metrics.clone();
tokio::spawn(async move { tokio::spawn(async move {
let handles = Handles { let handles = Handles {
forge, forge,
queue: gate, queue: gate,
poster, poster,
runs: store, runs: store,
metrics,
}; };
consume(&client, TEST_TIMING, &reviews, handles).await; consume(&client, TEST_TIMING, &reviews, handles).await;
}) })
@ -2033,6 +2086,7 @@ mod tests {
Some(&q.client), Some(&q.client),
&*q.rig.poster, &*q.rig.poster,
&q.rig.store, &q.rig.store,
&Metrics::off(),
&requested_at("old"), &requested_at("old"),
) )
.await; .await;
@ -2104,6 +2158,14 @@ mod tests {
"{}", "{}",
posted[1].2.body 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); assert_eq!(q.rig.gate.reached(), 1);
} }
@ -2132,12 +2194,21 @@ mod tests {
.expect("both leave the queue"); .expect("both leave the queue");
assert_eq!(q.rig.posted()[0].0, "o/r#8"); assert_eq!(q.rig.posted()[0].0, "o/r#8");
assert_eq!(q.rig.gate.reached(), 1); assert_eq!(q.rig.gate.reached(), 1);
let requests = q.recorder.counter(REQUESTS);
assert_eq!(requests.get("outcome=unparsable"), Some(&1), "{requests:?}");
} }
#[tokio::test] #[tokio::test]
async fn a_request_that_cannot_be_queued_posts_a_no_verdict_comment() { async fn a_request_that_cannot_be_queued_posts_a_no_verdict_comment() {
let rig = Rig::new(); 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(); let posted = rig.posted();
assert_eq!(posted.len(), 1); assert_eq!(posted.len(), 1);
let (target, commit, review) = &posted[0]; let (target, commit, review) = &posted[0];
@ -2960,4 +3031,211 @@ mod tests {
"a run that never ran read no diff: {body}" "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<const N: usize>(pairs: [(&str, u64); N]) -> std::collections::BTreeMap<String, u64> {
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)]));
}
} }

View file

@ -30,7 +30,7 @@ use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
/// Export cadence. Matches every other OTEL exporter in this tree. /// 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 /// 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. /// this module needs, not a full mirror of Forgejo's push-event schema.
@ -109,7 +109,7 @@ fn build() -> Option<Instruments> {
} }
} }
fn build_provider(interval: Duration) -> Result<SdkMeterProvider> { pub(crate) fn build_provider(interval: Duration) -> Result<SdkMeterProvider> {
// Same http/json, endpoint-from-env-only construction as // Same http/json, endpoint-from-env-only construction as
// `hive_jobq_metrics::build_provider` — see that module's doc comment // `hive_jobq_metrics::build_provider` — see that module's doc comment
// for why `with_endpoint` is deliberately never called here. // for why `with_endpoint` is deliberately never called here.
@ -190,7 +190,7 @@ fn resource() -> Resource {
builder.build() builder.build()
} }
fn endpoint() -> Option<String> { pub(crate) fn endpoint() -> Option<String> {
std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
.ok() .ok()
.map(|s| s.trim().to_owned()) .map(|s| s.trim().to_owned())

View file

@ -7,6 +7,10 @@
//! `claude` exports for itself on the claude runtime, so the cache hit rate is //! `claude` exports for itself on the claude runtime, so the cache hit rate is
//! `cacheRead` over the three input types on either runtime. //! `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 //! 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 //! tokens the agent reports, so it does not depend on how, or whether, the
//! agent prices a custom provider. //! agent prices a custom provider.
@ -23,7 +27,7 @@ use opentelemetry::KeyValue;
use opentelemetry::metrics::{Counter, MeterProvider as _}; use opentelemetry::metrics::{Counter, MeterProvider as _};
use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig};
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; 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. /// The currency every price is in.
pub const CURRENCY: &str = "USD"; pub const CURRENCY: &str = "USD";
@ -68,7 +72,6 @@ pub struct Otel;
impl CostSink for Otel { impl CostSink for Otel {
fn record(&self, cost: &Cost) { fn record(&self, cost: &Cost) {
static COUNTERS: OnceLock<Option<(SdkMeterProvider, Counters)>> = OnceLock::new();
let Some((_, counters)) = COUNTERS.get_or_init(build) else { let Some((_, counters)) = COUNTERS.get_or_init(build) else {
return; 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<Option<(SdkMeterProvider, Counters)>> = OnceLock::new();
/// The cost counter, the token counter beside it, and the retry counter.
struct Counters { struct Counters {
cost: Counter<f64>, cost: Counter<f64>,
tokens: Counter<u64>, tokens: Counter<u64>,
retries: Counter<u64>,
} }
/// The counters on `provider`. /// The counters on `provider`.
@ -92,9 +106,21 @@ fn counters(provider: &SdkMeterProvider) -> Counters {
tokens: meter tokens: meter
.u64_counter("hyperhive.review_classifier.token.usage") .u64_counter("hyperhive.review_classifier.token.usage")
.build(), .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. /// Adds `cost` to `counters` with the run's labels.
fn add(counters: &Counters, cost: &Cost) { fn add(counters: &Counters, cost: &Cost) {
let labels = [ 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())]);
}
} }

View file

@ -22,7 +22,7 @@ use swarm_wire::agent_inference::Effort;
use swarm_wire::review_classifier::{CLASSIFICATION_DEADLINE, Reply, Row, Runtime, TryOut}; use swarm_wire::review_classifier::{CLASSIFICATION_DEADLINE, Reply, Row, Runtime, TryOut};
use crate::prompt::Prompt; 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. /// How long `claude` may print nothing before it is stopped.
const IDLE_TIMEOUT: Duration = Duration::from_mins(2); 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, run,
CLASSIFICATION_DEADLINE, CLASSIFICATION_DEADLINE,
RETRY_FLOOR, RETRY_FLOOR,
|| cost::ask_retried(&run.runtime),
) )
.await; .await;
rows.push(aggregate::row(c, outcome)); 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 /// 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 /// 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, /// `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( async fn classify(
model: &impl Model, model: &impl Model,
prompt: &Prompt<'_>, prompt: &Prompt<'_>,
@ -104,6 +105,7 @@ async fn classify(
run: &TryOut, run: &TryOut,
budget: Duration, budget: Duration,
floor: Duration, floor: Duration,
on_retry: impl FnOnce(),
) -> Result<answer::Answer, String> { ) -> Result<answer::Answer, String> {
let started = Instant::now(); let started = Instant::now();
let first = model.ask(prompt, key, run, budget).await?; let first = model.ask(prompt, key, run, budget).await?;
@ -115,6 +117,7 @@ async fn classify(
if left < floor { if left < floor {
return Err(unparsed); return Err(unparsed);
} }
on_retry();
let again = model let again = model
.ask(&prompt.with_json_reminder(), key, run, left) .ask(&prompt.with_json_reminder(), key, run, left)
.await?; .await?;
@ -643,9 +646,27 @@ mod tests {
const FLOOR: Duration = Duration::from_millis(100); const FLOOR: Duration = Duration::from_millis(100);
async fn classify_slow(model: &Slow) -> Result<answer::Answer, String> { async fn classify_slow(model: &Slow) -> Result<answer::Answer, String> {
classify_counting(model).await.0
}
/// The result, and how often `on_retry` ran.
async fn classify_counting(model: &Slow) -> (Result<answer::Answer, String>, usize) {
let run = run(&["c"]); let run = run(&["c"]);
let prompt = Prompt::of("q"); 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] #[tokio::test]