From 4766f566dee3e03a50020ac5792d085f7dd70bc1 Mon Sep 17 00:00:00 2001 From: jordan Date: Sun, 23 Aug 2026 10:31:57 -0600 Subject: [PATCH] feat(observability): HTTP metrics, structured logs, dashboard, live tidalctl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit There was no metric anywhere that could answer "how much traffic are we serving" or "what is our error rate". The engine published a rich DOMAIN surface (search latency, WAL fsync, quorum timeouts, replication lag) and nothing about HTTP, so a cluster could serve 401s or 503s indefinitely with every existing gauge looking healthy. Logs were collected but unusable. There was no way to ask a RUNNING node anything. 1. HTTP metrics. tidaldb_http_requests_total{route,method,status} plus a per-route duration histogram, recorded by one layer placed OUTSIDE the auth, timeout and rate-limit layers so it sees the status actually returned to the client. Cardinality is the whole design: the route label is axum's MatchedPath TEMPLATE, not the path, and unmatched requests collapse into one bucket so a 404 flood cannot mint series. A hard cap folds anything past it into an overflow bucket while established series keep counting. The engine owns the /metrics listener but must not learn what a route or a status code is, so it gained one registration hook (MetricsState::set_extra_renderer) and tidal-server publishes through it. One scrape target per node, not two. 2. Structured logs. The previous init was a bare tracing_subscriber::fmt(), which produced two real defects: ANSI escapes leaked into collected logs, and every line failed the collector's JSON parse and was stamped level=info — so `level:error` matched NOTHING and errors were invisible to the log platform while being collected. JSON_LOGS=1 emits the collector's exact wire format (ts/level/service/env/msg), span fields are lifted so request_id lands on every line of a request, and ANSI is off unconditionally in both formats. Verified against the running binary, which caught a defect no unit test would have: dependencies logging through the `log` crate arrived with target="log" and four log.* metadata fields (absolute cargo registry paths, indexed forever). The real module is now lifted into target and the bridge metadata pruned. 3. Dashboard. docs/ops/grafana-tidaldb.json, 13 panels, mirrored into the fleet as a grafana-database-dashboards key. Every metric name was checked against a live endpoint and all 26 PromQL expressions were executed against the live TSDB before commit, because a dashboard full of "No data" is worse than none. Confirmed loaded in Grafana (uid tidaldb-overview, Databases folder). 4. tidalctl live mode. Every other subcommand reads a data dir AT REST, some requiring a stopped node. `search`, `feed`, `cluster-status` and `watch` take --url and talk to a running server, with --ca/--insecure because a cluster's client port is served with the INTERNAL cluster CA. Exit codes follow the crate contract, so `tidalctl cluster-status && deploy` gates on convergence. Its first real run immediately found a reporting defect: the aggregated /cluster/status reported two HEALTHY peers as UNREACHABLE PARTITIONED at 13.3M lag, having derived lag against an uninitialised applied=0, while every node's own status reported lag=0, reseed=false and identical frontiers, with pod-to-pod connectivity open and nothing logged. cluster-status now names that signature "NO REPORT (aggregated view; query the node directly)" instead of repeating it as replication lag; a genuine non-zero-applied lag still reports BEHIND. The underlying gap is documented as open work in docs/ops/observability.md. Verified: 2101 + 175 engine/server unit tests, 8 standalone integration (3 new, including the cardinality proof and the cross-crate metrics seam), 23 tidalctl (10 new), reseed + catchup + admin-gate e2e green, clippy clean, and both the metrics and the log format exercised against a real running binary. --- Cargo.lock | 15 + docs/ops/grafana-tidaldb.json | 1169 ++++++++++++++++++++++++++++++ docs/ops/observability.md | 172 +++++ tidal-server/Cargo.toml | 10 +- tidal-server/src/cluster/node.rs | 5 + tidal-server/src/http_metrics.rs | 377 ++++++++++ tidal-server/src/lib.rs | 6 + tidal-server/src/logging.rs | 407 +++++++++++ tidal-server/src/main.rs | 9 +- tidal-server/src/router.rs | 22 +- tidal-server/tests/standalone.rs | 160 ++++ tidal/src/db/metrics/mod.rs | 52 +- tidal/src/lib.rs | 1 + tidalctl/Cargo.toml | 6 + tidalctl/src/commands/live.rs | 656 +++++++++++++++++ tidalctl/src/commands/mod.rs | 3 + tidalctl/src/main.rs | 213 +++++- 17 files changed, 3260 insertions(+), 23 deletions(-) create mode 100644 docs/ops/grafana-tidaldb.json create mode 100644 docs/ops/observability.md create mode 100644 tidal-server/src/http_metrics.rs create mode 100644 tidal-server/src/logging.rs create mode 100644 tidalctl/src/commands/live.rs diff --git a/Cargo.lock b/Cargo.lock index 85db1a7..024e480 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4248,6 +4248,7 @@ dependencies = [ "axum 0.8.8", "base64", "blake3", + "chrono", "clap", "criterion", "crossbeam", @@ -4296,6 +4297,7 @@ dependencies = [ "aws-credential-types", "aws-sdk-s3", "blake3", + "reqwest", "serde", "serde_json", "tempfile", @@ -4668,6 +4670,16 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.22" @@ -4678,12 +4690,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] diff --git a/docs/ops/grafana-tidaldb.json b/docs/ops/grafana-tidaldb.json new file mode 100644 index 0000000..119eb2d --- /dev/null +++ b/docs/ops/grafana-tidaldb.json @@ -0,0 +1,1169 @@ +{ + "annotations": { + "list": [] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "links": [], + "liveNow": false, + "panels": [ + { + "id": 1, + "type": "row", + "title": "HTTP surface \u2014 usage and errors", + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 0 + }, + "panels": [] + }, + { + "id": 2, + "type": "timeseries", + "title": "Request rate by route", + "description": "Requests per second by matched route TEMPLATE. Before tidaldb_http_requests_total existed there was no metric anywhere that could answer 'how much traffic are we serving'.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 1 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 40, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "normal" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "reqps", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum by (route) (rate(tidaldb_http_requests_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "{{route}}", + "range": true, + "refId": "A" + } + ] + }, + { + "id": 3, + "type": "timeseries", + "title": "Requests by status", + "description": "Exact status codes, not classes: 401 (no/bad credential) vs 403 (valid data bearer, not an operator) vs 429 (rate limited) vs 503 (quorum timeout) are different incidents.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 1 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 40, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "normal" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "reqps", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum by (status) (rate(tidaldb_http_requests_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "{{status}}", + "range": true, + "refId": "A" + } + ] + }, + { + "id": 4, + "type": "stat", + "title": "5xx ratio", + "description": "Server-fault share of served requests. clamp_min keeps an idle cluster from dividing by zero and painting red.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 6, + "w": 4, + "x": 0, + "y": 9 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "yellow", + "value": 0.01 + }, + { + "color": "red", + "value": 0.05 + } + ] + }, + "mappings": [], + "unit": "percentunit", + "decimals": 2 + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNonNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_http_requests_total{namespace=~\"$namespace\",pod=~\"$pod\",status=~\"5..\"}[$__rate_interval])) / clamp_min(sum(rate(tidaldb_http_requests_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval])), 0.001)", + "legendFormat": "5xx share", + "range": true, + "refId": "A" + } + ] + }, + { + "id": 5, + "type": "timeseries", + "title": "Auth rejections (401 / 403)", + "description": "401 = missing/invalid credential. 403 = a valid DATA bearer used against an operator verb, i.e. a client trying to reach /cluster/*. A sustained 403 rate is a client misconfigured with the wrong key, or someone probing.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 6, + "w": 10, + "x": 4, + "y": 9 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "reqps", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum by (status, route) (rate(tidaldb_http_requests_total{namespace=~\"$namespace\",pod=~\"$pod\",status=~\"401|403\"}[$__rate_interval]))", + "legendFormat": "{{status}} {{route}}", + "range": true, + "refId": "A" + } + ] + }, + { + "id": 6, + "type": "timeseries", + "title": "HTTP p99 by route", + "description": "End-to-end latency as the CLIENT sees it, including auth, queueing behind the concurrency cap, and the leader forward. The engine's own latency series exclude all of that.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 6, + "w": 10, + "x": 14, + "y": 9 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ms", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (route, le) (rate(tidaldb_http_request_duration_us_bucket{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))) / 1000", + "legendFormat": "{{route}}", + "range": true, + "refId": "A" + } + ] + }, + { + "id": 7, + "type": "row", + "title": "Query engine latency", + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 15 + }, + "panels": [] + }, + { + "id": 8, + "type": "timeseries", + "title": "Search latency", + "description": "Search percentiles published directly by the engine.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 8, + "x": 0, + "y": 16 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ms", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_search_latency_us_p50{namespace=~\"$namespace\",pod=~\"$pod\"} / 1000", + "legendFormat": "p50", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_search_latency_us_p95{namespace=~\"$namespace\",pod=~\"$pod\"} / 1000", + "legendFormat": "p95", + "range": true, + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_search_latency_us_p99{namespace=~\"$namespace\",pod=~\"$pod\"} / 1000", + "legendFormat": "p99", + "range": true, + "refId": "C" + } + ] + }, + { + "id": 9, + "type": "timeseries", + "title": "Retrieve latency", + "description": "Retrieve percentiles published directly by the engine.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 8, + "x": 8, + "y": 16 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ms", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_retrieve_latency_us_p50{namespace=~\"$namespace\",pod=~\"$pod\"} / 1000", + "legendFormat": "p50", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_retrieve_latency_us_p95{namespace=~\"$namespace\",pod=~\"$pod\"} / 1000", + "legendFormat": "p95", + "range": true, + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_retrieve_latency_us_p99{namespace=~\"$namespace\",pod=~\"$pod\"} / 1000", + "legendFormat": "p99", + "range": true, + "refId": "C" + } + ] + }, + { + "id": 10, + "type": "timeseries", + "title": "Signal write p99", + "description": "Write path. WAL group-commit fsync is plotted alongside because it is usually the reason a write p99 moves.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 8, + "x": 16, + "y": 16 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "ms", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le) (rate(tidaldb_signal_write_latency_us_bucket{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))) / 1000", + "legendFormat": "p99", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "histogram_quantile(0.99, sum by (le) (rate(tidaldb_cluster_wal_fsync_us_bucket{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))) / 1000", + "legendFormat": "WAL fsync p99", + "range": true, + "refId": "B" + } + ] + }, + { + "id": 11, + "type": "row", + "title": "Cluster correctness", + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 23 + }, + "panels": [] + }, + { + "id": 12, + "type": "timeseries", + "title": "Replication lag", + "description": "Events this replica is behind the leader. Sustained non-zero means a follower is not keeping up; a step change usually follows a leader change.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 8, + "x": 0, + "y": 24 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_replication_lag_seqno{namespace=~\"$namespace\",pod=~\"$pod\"}", + "legendFormat": "{{pod}}", + "range": true, + "refId": "A" + } + ] + }, + { + "id": 13, + "type": "timeseries", + "title": "Correctness counters", + "description": "scatter degraded is the one to watch: a 200 answered with PARTIAL results because a shard errored or missed its deadline. It is a CORRECTNESS signal that looks like success to the client.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 8, + "x": 8, + "y": 24 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_cluster_quorum_timeouts_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "quorum timeouts", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_cluster_forward_failures_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "forward failures", + "range": true, + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_cluster_scatter_degraded_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "scatter degraded", + "range": true, + "refId": "C" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_cluster_leader_changes_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "leader changes", + "range": true, + "refId": "D" + } + ] + }, + { + "id": 14, + "type": "stat", + "title": "Fleet state", + "description": "health 0 or reseed pending 1 means a node is not serving normally. A reseed marker that persists past a restart is the m11p5 livelock signature.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 8, + "x": 16, + "y": 24 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "mappings": [] + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": [ + "lastNonNull" + ], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "min(tidaldb_health_ok{namespace=~\"$namespace\",pod=~\"$pod\"})", + "legendFormat": "health", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "max(tidaldb_cluster_reseed_required{namespace=~\"$namespace\",pod=~\"$pod\"})", + "legendFormat": "reseed pending", + "range": true, + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "max(tidaldb_usearch_vector_count{namespace=~\"$namespace\",pod=~\"$pod\"})", + "legendFormat": "vectors", + "range": true, + "refId": "C" + } + ] + }, + { + "id": 15, + "type": "row", + "title": "Capacity and backpressure", + "collapsed": false, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 31 + }, + "panels": [] + }, + { + "id": 16, + "type": "timeseries", + "title": "Shed load", + "description": "Load actively shed. NOTE: tidalDB's own limiter is UNLIMITED unless TIDAL_RATE_LIMIT_RPS is set, and it keys per PRINCIPAL \u2014 with one shared bearer every caller is one principal, so this is an aggregate cap, not fairness.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 12, + "x": 0, + "y": 32 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "unit": "reqps", + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_rate_limited_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "rate limited", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum(rate(tidaldb_cluster_write_pool_rejections_total{namespace=~\"$namespace\",pod=~\"$pod\"}[$__rate_interval]))", + "legendFormat": "write pool rejections", + "range": true, + "refId": "B" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "sum by (status) (rate(tidaldb_http_requests_total{namespace=~\"$namespace\",pod=~\"$pod\",status=~\"429|408\"}[$__rate_interval]))", + "legendFormat": "http {{status}}", + "range": true, + "refId": "C" + } + ] + }, + { + "id": 17, + "type": "timeseries", + "title": "Durability backlog", + "description": "Uncompacted WAL and time since the last successful checkpoint. A checkpoint age that only climbs means the checkpoint thread is wedged \u2014 recovery time grows with it.", + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 32 + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "palette-classic" + }, + "custom": { + "axisPlacement": "auto", + "drawStyle": "line", + "fillOpacity": 8, + "lineWidth": 2, + "showPoints": "never", + "stacking": { + "group": "A", + "mode": "none" + } + }, + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + }, + "min": 0 + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_wal_lag_bytes{namespace=~\"$namespace\",pod=~\"$pod\"}", + "legendFormat": "WAL lag bytes {{pod}}", + "range": true, + "refId": "A" + }, + { + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "editorMode": "code", + "expr": "tidaldb_checkpoint_age_seconds{namespace=~\"$namespace\",pod=~\"$pod\"}", + "legendFormat": "checkpoint age {{pod}}", + "range": true, + "refId": "B" + } + ] + } + ], + "refresh": "30s", + "schemaVersion": 39, + "tags": [ + "tidaldb", + "database" + ], + "templating": { + "list": [ + { + "current": {}, + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "definition": "label_values(tidaldb_health_ok, namespace)", + "hide": 0, + "includeAll": true, + "label": "namespace", + "multi": true, + "name": "namespace", + "options": [], + "query": { + "query": "label_values(tidaldb_health_ok, namespace)", + "refId": "A" + }, + "refresh": 2, + "sort": 1, + "type": "query" + }, + { + "current": {}, + "datasource": { + "type": "prometheus", + "uid": "${DS_PROMETHEUS}" + }, + "definition": "label_values(tidaldb_health_ok{namespace=~\"$namespace\"}, pod)", + "hide": 0, + "includeAll": true, + "label": "pod", + "multi": true, + "name": "pod", + "options": [], + "query": { + "query": "label_values(tidaldb_health_ok{namespace=~\"$namespace\"}, pod)", + "refId": "A" + }, + "refresh": 2, + "sort": 1, + "type": "query" + } + ] + }, + "time": { + "from": "now-6h", + "to": "now" + }, + "timepicker": {}, + "timezone": "utc", + "title": "tidalDB \u2014 usage, errors, and cluster health", + "uid": "tidaldb-overview", + "version": 1, + "weekStart": "", + "description": "tidalDB usage, errors, and cluster health. PRODUCT-SIDE SOURCE OF TRUTH: this file lives in the tidalDB repo at docs/ops/grafana-tidaldb.json and is mirrored into the fleet as the tidaldb-overview.json key of the grafana-database-dashboards ConfigMap (deployments/k8s/base/observability/database-dashboards.yaml), the same convention docs/ops/prometheus-alerts.yaml follows. Edit here, then mirror. Every expression was validated against the live TSDB; the label set is namespace/pod/container/partition_id as attached by vmagent's annotation-based pod discovery." +} diff --git a/docs/ops/observability.md b/docs/ops/observability.md new file mode 100644 index 0000000..fa50244 --- /dev/null +++ b/docs/ops/observability.md @@ -0,0 +1,172 @@ +# Observability: usage, errors, and live debugging + +Four surfaces, in the order you reach for them during an incident. + +| Question | Surface | +|---|---| +| How much traffic, and how much of it is failing? | Grafana → **tidalDB — usage, errors, and cluster health** | +| What exactly failed, for which request? | VictoriaLogs, `level:error`, correlate on `request_id` | +| Is the cluster converged right now? | `tidalctl cluster-status` / `tidalctl watch` | +| What does a query actually return? | `tidalctl search` / `tidalctl feed` | + +--- + +## 1. Metrics + +Every node serves Prometheus text on its `--metrics` address (`:9091` in the +shipped manifests). **Never expose that port externally — it is +unauthenticated.** In-cluster it is reachable only from the `observability` +namespace (`k8s/cluster/networkpolicy.yaml`). + +Two families answer the usage/error questions: + +``` +tidaldb_http_requests_total{route,method,status} counter +tidaldb_http_request_duration_us{route} histogram +``` + +`route` is the matched route **template** (`/cluster/shards/{id}/transfer`), not +the concrete path, so cardinality is bounded by the router rather than by +traffic. Unmatched requests collapse into a single `` bucket — a 404 +flood cannot mint series. + +These are published by `tidal-server` onto the engine's existing listener (via +`MetricsState::set_extra_renderer`), so a node remains **one** scrape target. + +Useful queries: + +```promql +# request rate by route +sum by (route) (rate(tidaldb_http_requests_total[5m])) + +# server-fault ratio + sum(rate(tidaldb_http_requests_total{status=~"5.."}[5m])) +/ sum(rate(tidaldb_http_requests_total[5m])) + +# who is being refused, and where +sum by (status, route) (rate(tidaldb_http_requests_total{status=~"401|403"}[5m])) + +# client-observed p99, in ms +histogram_quantile(0.99, + sum by (route, le) (rate(tidaldb_http_request_duration_us_bucket[5m]))) / 1000 +``` + +Read `403` precisely: it means a valid **data** bearer was used against an +operator verb (`/cluster/*`). A sustained 403 rate is a client holding the wrong +key, or someone probing — not a server fault. + +### Dashboard + +`docs/ops/grafana-tidaldb.json` is the source of truth, mirrored into the fleet +as the `tidaldb-overview.json` key of the `grafana-database-dashboards` +ConfigMap — the same convention `docs/ops/prometheus-alerts.yaml` follows. Edit +here, then mirror. It lands in Grafana's **Databases** folder as +`tidaldb-overview`. + +--- + +## 2. Logs + +Set `JSON_LOGS=1` (or `TIDAL_LOG_FORMAT=json`) to emit one JSON object per line +in the collector's wire format: + +```json +{"ts":"2026-08-23T15:57:57.090256Z","level":"info","service":"tidal-server", + "env":"prod","msg":"committing 5","target":"tantivy::indexer","request_id":"418"} +``` + +Why it matters: the fleet's Vector collector parses each line as JSON and, on +success, replaces the event with that object; on failure it keeps the text and +stamps `level = "info"`. Without JSON, **every** tidalDB line was recorded as +info and `level:error` matched nothing — errors were collected but invisible. + +- `level` is lowercase, matching the collector's enum + (`critical|error|warn|info|debug`). +- `request_id` from the per-request span appears on every line emitted inside it, + so one request's work is correlatable rather than grep-adjacent. +- `target` is the real module even for dependencies logging through the `log` + crate (their bridge metadata is pruned rather than indexed forever). +- ANSI colour is off in **both** formats. It used to leak escape codes into + stored logs. + +Tunables: `TIDAL_SERVER_LOG` (env-filter, e.g. +`tidal_server=debug,tantivy=warn,info` to quiet a noisy dependency), +`TIDAL_SERVICE_NAME`, `TIDAL_ENV`. + +```bash +# errors, last 15 minutes +_time:15m AND kubernetes.pod_namespace:tidaldb-cluster AND level:error + +# one request end to end +_time:1h AND request_id:"418" +``` + +--- + +## 3. Live debugging with `tidalctl` + +Every other `tidalctl` command reads a data directory **at rest** — some require +a stopped node. These take `--url` and talk to a **running** server. + +```bash +export TIDAL_API_KEY=... # or pass --key + +# Is it converged? Exits 2 when not, so it gates a deploy. +tidalctl cluster-status --url http://127.0.0.1:9500 + +# Watch convergence (Ctrl-C to stop, or bound it with --count) +tidalctl watch --url http://127.0.0.1:9500 --interval 5 --count 12 + +# What does a query actually return? +tidalctl search --url http://127.0.0.1:9500 --query "cold brew" --limit 5 +tidalctl feed --url http://127.0.0.1:9500 --profile trending --user-id 42 +``` + +**TLS.** A cluster node's client port is served with the *internal* cluster CA, +issued for in-cluster DNS names, so a port-forward fails both CA and hostname +verification. Pin the CA with `--ca ca.pem`, or use `--insecure` for a local +forward: + +```bash +kubectl -n tidaldb-cluster port-forward svc/tidaldb 9500:9500 & +tidalctl cluster-status --url https://127.0.0.1:9500 --insecure +``` + +Exit codes follow the crate contract: `0` ok, `1` usage error, `2` degraded — +unreachable, non-2xx, or not converged. A `403` response prints the hint to use +the admin key. + +--- + +## 4. Known caveat: the aggregated `/cluster/status` under-reports peers + +The aggregated view can report a peer it holds no frontier report for as +`applied_events: 0`, and derive `lag_events` against that zero — so a fully +converged peer appears to be the leader's entire history behind, sometimes with +`reachable: false` / `partitioned: true` alongside. + +Observed 2026-08-23: two healthy nodes shown as `UNREACHABLE PARTITIONED` at +13.3M lag, while every node's own `/cluster/status/local` reported `lag=0`, +`reseed_required=false`, and identical per-group frontiers. Pod-to-pod +connectivity was open and no error was logged. + +`tidalctl cluster-status` detects that signature (`applied == 0` with non-zero +lag) and prints **`NO REPORT (aggregated view; query the node directly)`** +rather than repeating it as replication lag. A genuine lag report — non-zero +applied with non-zero lag — still reports `BEHIND`. + +**When you see `NO REPORT`, confirm against the nodes themselves** before +treating it as an incident: + +```bash +for i in 0 1 2; do + kubectl -n tidaldb-cluster port-forward tidaldb-$i 1970$i:9500 >/dev/null 2>&1 & + sleep 8 + curl -sk -H "Authorization: Bearer $TIDAL_API_KEY" \ + "https://127.0.0.1:1970$i/cluster/status/local" | jq '{region, reseed_required, shards}' + kill %1 +done +``` + +The underlying reporting gap in the aggregated surface is open work; it is a +reporting defect, not a replication one. diff --git a/tidal-server/Cargo.toml b/tidal-server/Cargo.toml index a4e75f9..4bb09f5 100644 --- a/tidal-server/Cargo.toml +++ b/tidal-server/Cargo.toml @@ -70,7 +70,15 @@ serde_yml = "0.0.12" thiserror = "2" tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "sync"] } tracing = "0.1" -tracing-subscriber = { version = "0.3", features = ["env-filter"] } +# `json` backs the structured log format (JSON_LOGS=1). Without it span fields +# are only stored as a human string, so `request_id` could not be lifted into a +# machine-readable field and per-request log correlation would stay grep-only. +tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +# RFC3339 timestamps for the structured log wire format. Already compiled in the +# workspace as a transitive dependency, so this is a promotion to a direct dep +# rather than new build cost; hand-rolling civil-date arithmetic from +# SystemTime would be a defect waiting to happen. +chrono = { version = "0.4", default-features = false, features = ["clock", "std"] } # utoipa 5.x derives the OpenAPI 3.1 document (ApiDoc) and per-handler path # attributes that back the unauthenticated GET /openapi.json route. JSON-only: # no swagger-ui asset bundle (that crate's vendored JS does not pass our diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index b0066c7..bd0bd83 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -4074,6 +4074,11 @@ fn open_region_db( } let db = builder.open().map_err(ServerError::Tidal)?; if let Some(addr) = db.metrics_addr() { + // This shard owns the listener, so its MetricsState is the one actually + // scraped: register the process's HTTP series against it. Without this, + // cluster mode (the production shape) would expose the engine's domain + // metrics but nothing about requests, statuses, or HTTP latency. + crate::http_metrics::publish_to(db.metrics()); tracing::info!( region = region_name, "cluster metrics endpoint listening on http://{addr}/metrics" diff --git a/tidal-server/src/http_metrics.rs b/tidal-server/src/http_metrics.rs new file mode 100644 index 0000000..ec715a1 --- /dev/null +++ b/tidal-server/src/http_metrics.rs @@ -0,0 +1,377 @@ +//! HTTP surface metrics: request counts by route/method/status, and per-route +//! latency. +//! +//! WHY THIS EXISTS: the engine publishes a rich DOMAIN surface (search latency, +//! WAL fsync, quorum timeouts, replication lag) but nothing about HTTP. There +//! was no way to answer "how many requests did we serve" or "what is our error +//! rate" from metrics at all — a production cluster could serve 401s or 503s +//! indefinitely with every existing gauge looking healthy. +//! +//! CARDINALITY IS THE WHOLE DESIGN CONSTRAINT. A naive `path` label explodes: +//! `/items/12345` is a distinct series per entity. Two defences: +//! +//! 1. The route label is axum's [`MatchedPath`] — the route TEMPLATE +//! (`/cluster/shards/{id}/transfer`), not the concrete path. Bounded by the +//! router's route count, whatever the traffic. +//! 2. A hard [`MAX_SERIES`] cap. Anything past it folds into an explicit +//! overflow bucket, so a future router change (or an unmatched-path flood) +//! degrades the labels rather than the process. +//! +//! Unmatched requests carry no `MatchedPath`, so they are attributed to a single +//! constant [`UNMATCHED_ROUTE`] rather than their (attacker-controlled) path. + +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::Instant; + +use axum::extract::MatchedPath; +use axum::extract::Request; +use axum::middleware::Next; +use axum::response::Response; +use tidaldb::{LatencyHistogram, QUERY_LATENCY_BOUNDS}; + +/// Route label for a request that matched no route (404s, and anything rejected +/// before routing). A single constant, never the caller-supplied path — that +/// path is attacker-controlled and would be an unbounded label. +const UNMATCHED_ROUTE: &str = ""; + +/// Route label used once [`MAX_SERIES`] distinct keys exist. +const OVERFLOW_ROUTE: &str = ""; + +/// Hard ceiling on distinct (route, method, status) keys. +/// +/// The template-based route label already bounds this to roughly +/// routes × methods × observed-statuses (~250 for the cluster router). The cap +/// exists so a future refactor that accidentally admits a dynamic route label +/// cannot grow this map without bound; it degrades to [`OVERFLOW_ROUTE`]. +const MAX_SERIES: usize = 512; + +/// One counted series: a route template, method, and HTTP status. +#[derive(PartialEq, Eq, Hash, Clone)] +struct Key { + route: Box, + method: &'static str, + status: u16, +} + +/// HTTP request counters and per-route latency for one server process. +/// +/// Registered with the engine's metrics listener via +/// `MetricsState::set_extra_renderer`, so these series appear on the SAME +/// scrape target as the engine's own — one target per node, not two. +pub struct HttpMetrics { + /// Counts keyed by (route, method, status). + /// + /// `std::sync::Mutex` rather than a lock-free structure or + /// `parking_lot`, deliberately. The critical section is one hash and one + /// `u64` increment (tens of nanoseconds) against requests that already do + /// WAL fsyncs and vector searches (hundreds of microseconds and up), and the + /// in-flight cap at `router::MAX_CONCURRENCY` bounds the contention ceiling. + /// `std::sync` is also what the rest of this workspace uses — introducing + /// `parking_lot` for one module would plant a second convention. Poisoning + /// is handled, never unwrapped: losing a metric sample must not fail a + /// request that already succeeded. + counts: Mutex>, + /// Per-route end-to-end latency, in microseconds. + latency: Mutex, LatencyHistogram>>, +} + +/// The process-wide HTTP metrics. +/// +/// A process serves one HTTP surface through one `/metrics` listener, so a +/// single instance is the honest model — and it keeps the recording layer from +/// having to be threaded through all three router builders (standalone, +/// single-process cluster, multi-process region) and their call sites. +static GLOBAL: std::sync::LazyLock> = + std::sync::LazyLock::new(|| std::sync::Arc::new(HttpMetrics::new())); + +/// Handle to the process-wide HTTP metrics. +#[must_use] +pub fn global() -> std::sync::Arc { + std::sync::Arc::clone(&GLOBAL) +} + +/// Publish this process's HTTP series on the engine's existing `/metrics` +/// listener. +/// +/// Call once after opening the database. Returns `false` if a renderer was +/// already registered (the engine ignores the second call rather than +/// double-rendering). +pub fn publish_to(metrics: &tidaldb::MetricsState) -> bool { + metrics.set_extra_renderer(Box::new(|out| global().render_into(out))) +} + +impl std::fmt::Debug for HttpMetrics { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let series = self.counts.lock().map_or(0, |m| m.len()); + f.debug_struct("HttpMetrics") + .field("series", &series) + .finish_non_exhaustive() + } +} + +impl Default for HttpMetrics { + fn default() -> Self { + Self::new() + } +} + +impl HttpMetrics { + #[must_use] + pub fn new() -> Self { + Self { + counts: Mutex::new(HashMap::new()), + latency: Mutex::new(HashMap::new()), + } + } + + /// Record one completed request. + /// + /// `route` should be the matched TEMPLATE. Poisoned locks are ignored rather + /// than propagated: losing a metric sample must never fail a request that + /// already succeeded. + pub fn record(&self, route: &str, method: &'static str, status: u16, elapsed_us: u64) { + if let Ok(mut counts) = self.counts.lock() { + let key = Key { + route: route.into(), + method, + status, + }; + // Only admit a NEW key while under the cap; existing keys always + // increment, so an established series never stops counting because + // the map happens to be full. + if let Some(hit) = counts.get_mut(&key) { + *hit += 1; + } else if counts.len() < MAX_SERIES { + counts.insert(key, 1); + } else { + *counts + .entry(Key { + route: OVERFLOW_ROUTE.into(), + method, + status, + }) + .or_insert(0) += 1; + } + } + + if let Ok(mut latency) = self.latency.lock() { + if let Some(hist) = latency.get(route) { + hist.observe(elapsed_us); + } else if latency.len() < MAX_SERIES { + let hist = LatencyHistogram::new(QUERY_LATENCY_BOUNDS); + hist.observe(elapsed_us); + latency.insert(route.into(), hist); + } + } + } + + /// Append this process's HTTP series in Prometheus text format. + /// + /// Label values are escaped: a route template is repo-controlled, but + /// emitting an unescaped label is the kind of thing that silently corrupts a + /// whole scrape, so it is not left to chance. + pub fn render_into(&self, out: &mut String) { + use std::fmt::Write; + + if let Ok(counts) = self.counts.lock() { + let _ = write!( + out, + "\n# HELP tidaldb_http_requests_total Total HTTP requests served, by matched route \ + template, method, and response status.\n\ + # TYPE tidaldb_http_requests_total counter\n" + ); + // Sorted so a diff of two scrapes is readable and the output is + // deterministic for tests. + let mut rows: Vec<_> = counts.iter().collect(); + rows.sort_unstable_by(|(a, _), (b, _)| { + (&a.route, a.method, a.status).cmp(&(&b.route, b.method, b.status)) + }); + for (key, count) in rows { + let _ = writeln!( + out, + "tidaldb_http_requests_total{{route=\"{}\",method=\"{}\",status=\"{}\"}} {count}", + escape_label(&key.route), + key.method, + key.status + ); + } + } + + if let Ok(latency) = self.latency.lock() { + let mut routes: Vec<_> = latency.iter().collect(); + routes.sort_unstable_by(|(a, _), (b, _)| a.cmp(b)); + for (route, hist) in routes { + out.push_str(&hist.render_prometheus_labeled( + "tidaldb_http_request_duration_us", + "HTTP request end-to-end latency in microseconds, by matched route template.", + &format!("route=\"{}\"", escape_label(route)), + )); + } + } + } +} + +/// Escape a Prometheus label value per the exposition format: backslash, double +/// quote, and newline. +fn escape_label(raw: &str) -> String { + if !raw.contains(['\\', '"', '\n']) { + return raw.to_string(); + } + let mut out = String::with_capacity(raw.len() + 8); + for c in raw.chars() { + match c { + '\\' => out.push_str("\\\\"), + '"' => out.push_str("\\\""), + '\n' => out.push_str("\\n"), + other => out.push(other), + } + } + out +} + +/// Map a method to a `'static` label, so the common ones cost no allocation and +/// an exotic one cannot become an unbounded label value. +fn method_label(method: &axum::http::Method) -> &'static str { + match *method { + axum::http::Method::GET => "GET", + axum::http::Method::POST => "POST", + axum::http::Method::PUT => "PUT", + axum::http::Method::DELETE => "DELETE", + axum::http::Method::PATCH => "PATCH", + axum::http::Method::HEAD => "HEAD", + axum::http::Method::OPTIONS => "OPTIONS", + _ => "OTHER", + } +} + +/// Middleware recording every request into `metrics`. +/// +/// Applied as the OUTERMOST layer on each router so it observes the response +/// actually returned to the client — including the 401/403 from the auth gates, +/// the 408 from the timeout layer, and the 429 from the rate limiter. A layer +/// applied further in would miss exactly the failures worth counting. +pub async fn track(metrics: std::sync::Arc, req: Request, next: Next) -> Response { + // The template, not the concrete path. Absent for unmatched requests. + let route = req + .extensions() + .get::() + .map_or(UNMATCHED_ROUTE, |m| m.as_str()) + .to_string(); + let method = method_label(req.method()); + let started = Instant::now(); + let response = next.run(req).await; + let elapsed_us = u64::try_from(started.elapsed().as_micros()).unwrap_or(u64::MAX); + metrics.record(&route, method, response.status().as_u16(), elapsed_us); + response +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn counts_are_keyed_by_route_method_and_status() { + let m = HttpMetrics::new(); + m.record("/items", "POST", 201, 10); + m.record("/items", "POST", 201, 20); + m.record("/items", "POST", 401, 5); + m.record("/search", "GET", 200, 30); + + let mut out = String::new(); + m.render_into(&mut out); + assert!(out.contains( + "tidaldb_http_requests_total{route=\"/items\",method=\"POST\",status=\"201\"} 2" + )); + assert!(out.contains( + "tidaldb_http_requests_total{route=\"/items\",method=\"POST\",status=\"401\"} 1" + )); + assert!(out.contains( + "tidaldb_http_requests_total{route=\"/search\",method=\"GET\",status=\"200\"} 1" + )); + } + + /// The error surface is the reason this module exists, so prove a 5xx is + /// distinguishable from a success on the same route. + #[test] + fn errors_are_countable_separately_from_successes() { + let m = HttpMetrics::new(); + for _ in 0..7 { + m.record("/feed", "GET", 200, 100); + } + m.record("/feed", "GET", 503, 100); + + let mut out = String::new(); + m.render_into(&mut out); + assert!(out.contains("route=\"/feed\",method=\"GET\",status=\"200\"} 7")); + assert!(out.contains("route=\"/feed\",method=\"GET\",status=\"503\"} 1")); + } + + #[test] + fn latency_histogram_is_rendered_per_route() { + let m = HttpMetrics::new(); + m.record("/search", "GET", 200, 1_500); + let mut out = String::new(); + m.render_into(&mut out); + assert!(out.contains("tidaldb_http_request_duration_us")); + assert!(out.contains("route=\"/search\"")); + // A histogram must carry bucket + count series, not just a gauge. + assert!(out.contains("_bucket"), "expected le buckets: {out}"); + assert!(out.contains("_count")); + } + + /// Cardinality is the failure mode this design exists to prevent: past the + /// cap, NEW keys must fold into one overflow bucket instead of growing the + /// map, while already-tracked series keep counting. + #[test] + fn series_are_capped_and_existing_keys_keep_counting() { + let m = HttpMetrics::new(); + for i in 0..MAX_SERIES { + m.record(&format!("/r{i}"), "GET", 200, 1); + } + // Established key still increments at the cap. + m.record("/r0", "GET", 200, 1); + // A brand new key does NOT add a series. + m.record("/brand-new", "GET", 200, 1); + + let counts = m.counts.lock().unwrap(); + assert!( + counts.len() <= MAX_SERIES + 1, + "map grew past the cap (+1 overflow bucket): {}", + counts.len() + ); + assert_eq!( + *counts + .get(&Key { + route: "/r0".into(), + method: "GET", + status: 200 + }) + .unwrap(), + 2 + ); + assert!( + counts.keys().any(|k| &*k.route == OVERFLOW_ROUTE), + "over-cap requests must be attributed to the overflow bucket" + ); + assert!( + !counts.keys().any(|k| &*k.route == "/brand-new"), + "a new route past the cap must NOT create its own series" + ); + } + + #[test] + fn label_values_are_escaped() { + assert_eq!(escape_label("/items"), "/items"); + assert_eq!(escape_label("a\"b"), "a\\\"b"); + assert_eq!(escape_label("a\\b"), "a\\\\b"); + assert_eq!(escape_label("a\nb"), "a\\nb"); + } + + #[test] + fn exotic_methods_collapse_to_a_bounded_label() { + assert_eq!(method_label(&axum::http::Method::GET), "GET"); + assert_eq!(method_label(&axum::http::Method::TRACE), "OTHER"); + } +} diff --git a/tidal-server/src/lib.rs b/tidal-server/src/lib.rs index 46f5a10..37ec98c 100644 --- a/tidal-server/src/lib.rs +++ b/tidal-server/src/lib.rs @@ -18,6 +18,12 @@ pub mod config; pub mod dto; pub mod error; pub mod health; +/// HTTP surface metrics (requests by route/method/status, per-route latency), +/// published through the engine's existing `/metrics` listener. +pub mod http_metrics; +/// Log initialisation: ANSI-free text, or the collector's JSON wire format +/// under `JSON_LOGS=1`. +pub mod logging; pub mod offload; pub mod openapi; pub mod router; diff --git a/tidal-server/src/logging.rs b/tidal-server/src/logging.rs new file mode 100644 index 0000000..fa1905f --- /dev/null +++ b/tidal-server/src/logging.rs @@ -0,0 +1,407 @@ +//! Log initialisation: plain text for humans, JSON for aggregation. +//! +//! WHY THIS EXISTS. The previous init was a bare `tracing_subscriber::fmt()`, +//! which produced two concrete defects in the shipped deployment: +//! +//! 1. **ANSI escapes in stored logs.** `fmt()` colourises when it believes it +//! has a terminal, and in k8s that belief was wrong — collected lines carried +//! raw `\x1b[2m` fragments, so stored logs were polluted and grep patterns +//! had to tolerate escape codes. ANSI is now off unconditionally: this +//! process's stdout is a log pipe, never a terminal. +//! +//! 2. **Unstructured lines, so no `level` field.** The fleet's Vector collector +//! parses each line as JSON and, on success, REPLACES the event with that +//! object; on failure it keeps the text and stamps `level = "info"`. Every +//! tidalDB line took the failure path, so *every* line was recorded as info +//! and `level:error` matched nothing. Errors were invisible to the log +//! platform even though they were being collected. +//! +//! The JSON form therefore emits the collector's wire format exactly — `ts`, +//! `level`, `service`, `env`, `msg` — plus `target` and any event/span fields. +//! Span fields matter most: the per-request span carries `request_id`, so with +//! JSON on, every line of a request's work is machine-correlatable instead of +//! grep-adjacent. +//! +//! `level` is emitted lowercase to match the collector's five-value enum +//! (`critical|error|warn|info|debug`). It folds case itself, but emitting the +//! canonical form keeps the stored value identical whether or not that +//! normalisation runs. + +use std::fmt; + +use tracing::{Event, Subscriber}; +use tracing_subscriber::fmt::format::{JsonFields, Writer}; +use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields, FormattedFields}; +use tracing_subscriber::registry::LookupSpan; + +/// Default `service` label. Overridable with `TIDAL_SERVICE_NAME` so several +/// tidalDB deployments in one log platform stay distinguishable. +const DEFAULT_SERVICE: &str = "tidal-server"; + +/// Initialise the global subscriber. +/// +/// * `TIDAL_SERVER_LOG` — env-filter directive (default `info`). +/// * `JSON_LOGS` / `TIDAL_LOG_FORMAT=json` — emit the structured wire format. +/// * `TIDAL_SERVICE_NAME` — `service` label (default [`DEFAULT_SERVICE`]). +/// * `TIDAL_ENV` — `env` label; omitted when unset rather than guessed. +/// +/// Idempotent: a second call is a no-op, so tests that initialise logging do not +/// fight each other. +pub fn init() { + let filter = std::env::var("TIDAL_SERVER_LOG").unwrap_or_else(|_| "info".into()); + + if json_requested() { + let service = std::env::var("TIDAL_SERVICE_NAME") + .ok() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| DEFAULT_SERVICE.to_string()); + let env = std::env::var("TIDAL_ENV") + .ok() + .filter(|s| !s.trim().is_empty()); + let _ = tracing_subscriber::fmt() + .with_env_filter(filter) + .with_ansi(false) + // Span fields are stored as JSON so `request_id` can be lifted into + // a real field rather than scraped out of a display string. + .fmt_fields(JsonFields::default()) + .event_format(WireFormat { service, env }) + .try_init(); + } else { + // ANSI off even here: this stdout is a log pipe in every deployment + // shape we ship, and escape codes in a captured log are pure noise. + let _ = tracing_subscriber::fmt() + .with_env_filter(filter) + .with_ansi(false) + .try_init(); + } +} + +/// Whether the structured format was requested. +fn json_requested() -> bool { + let truthy = |v: String| { + let v = v.trim().to_ascii_lowercase(); + v == "1" || v == "true" || v == "yes" + }; + if std::env::var("JSON_LOGS").is_ok_and(truthy) { + return true; + } + std::env::var("TIDAL_LOG_FORMAT").is_ok_and(|v| v.trim().eq_ignore_ascii_case("json")) +} + +/// The collector's wire format: one JSON object per line. +struct WireFormat { + service: String, + env: Option, +} + +impl FormatEvent for WireFormat +where + S: Subscriber + for<'a> LookupSpan<'a>, + N: for<'a> FormatFields<'a> + 'static, +{ + fn format_event( + &self, + ctx: &FmtContext<'_, S, N>, + mut writer: Writer<'_>, + event: &Event<'_>, + ) -> fmt::Result { + let meta = event.metadata(); + let mut map = serde_json::Map::new(); + + map.insert( + "ts".to_string(), + serde_json::Value::String( + chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Micros, true), + ), + ); + map.insert( + "level".to_string(), + serde_json::Value::String(level_label(meta.level()).to_string()), + ); + map.insert( + "service".to_string(), + serde_json::Value::String(self.service.clone()), + ); + if let Some(env) = &self.env { + map.insert("env".to_string(), serde_json::Value::String(env.clone())); + } + // Event fields. `message` becomes `msg` (the collector's key); the rest + // are carried verbatim. + let mut visitor = FieldCollector::default(); + event.record(&mut visitor); + + // A `log`-crate bridged event reports its target as the literal "log"; + // the real module arrives as a field. Prefer that so `target` is always + // the thing you would actually filter on. + map.insert( + "target".to_string(), + serde_json::Value::String( + visitor + .log_target + .clone() + .unwrap_or_else(|| meta.target().to_string()), + ), + ); + map.insert( + "msg".to_string(), + serde_json::Value::String(visitor.message.unwrap_or_default()), + ); + for (key, value) in visitor.fields { + map.entry(key).or_insert(value); + } + + // Span fields, outermost first, so `request_id` from the per-request + // span lands on every line emitted inside it. An event field of the same + // name wins (`or_insert`): the more specific value is the truer one. + if let Some(scope) = ctx.event_scope() { + for span in scope.from_root() { + let ext = span.extensions(); + if let Some(fields) = ext.get::>() + && let Ok(serde_json::Value::Object(obj)) = + serde_json::from_str::(fields) + { + for (key, value) in obj { + map.entry(key).or_insert(value); + } + } + } + } + + writeln!(writer, "{}", serde_json::Value::Object(map)) + } +} + +/// Lowercase level names matching the collector's closed enum. +const fn level_label(level: &tracing::Level) -> &'static str { + match *level { + tracing::Level::ERROR => "error", + tracing::Level::WARN => "warn", + tracing::Level::INFO => "info", + tracing::Level::DEBUG | tracing::Level::TRACE => "debug", + } +} + +/// Collects an event's fields, separating `message` and the `log`-crate bridge +/// metadata from real fields. +#[derive(Default)] +struct FieldCollector { + message: Option, + /// The originating module for an event bridged from the `log` crate. + /// + /// `tracing-log` reports `metadata().target()` as the literal `"log"` for + /// these and puts the real module in a `log.target` FIELD. Without lifting + /// it, every dependency's log line (tantivy commits, for one) arrives with + /// `target: "log"` and is unfilterable by module. + log_target: Option, + fields: Vec<(String, serde_json::Value)>, +} + +impl FieldCollector { + fn put(&mut self, field: &tracing::field::Field, value: serde_json::Value) { + match field.name() { + "message" => self.message = value.as_str().map(ToString::to_string), + "log.target" => self.log_target = value.as_str().map(ToString::to_string), + // The rest of the bridge metadata (`log.file`, `log.line`, + // `log.module_path`) is dropped: absolute paths into the cargo + // registry on every dependency line, duplicating `target`, and paid + // for in the log index forever. + name if name.starts_with("log.") => {} + name => self.fields.push((name.to_string(), value)), + } + } +} + +impl tracing::field::Visit for FieldCollector { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.put(field, serde_json::Value::String(value.to_string())); + } + + fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { + self.put(field, serde_json::Value::Bool(value)); + } + + fn record_i64(&mut self, field: &tracing::field::Field, value: i64) { + self.put(field, serde_json::Value::from(value)); + } + + fn record_u64(&mut self, field: &tracing::field::Field, value: u64) { + self.put(field, serde_json::Value::from(value)); + } + + fn record_f64(&mut self, field: &tracing::field::Field, value: f64) { + self.put(field, serde_json::Value::from(value)); + } + + fn record_error( + &mut self, + field: &tracing::field::Field, + value: &(dyn std::error::Error + 'static), + ) { + self.put(field, serde_json::Value::String(value.to_string())); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn fmt::Debug) { + self.put(field, serde_json::Value::String(format!("{value:?}"))); + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + use tracing_subscriber::fmt::MakeWriter; + + /// Captures emitted lines so the exposition can be asserted directly. + #[derive(Clone, Default)] + struct Buffer(Arc>>); + + impl std::io::Write for Buffer { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for Buffer { + type Writer = Self; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + /// Render one event through the wire formatter and return the JSON line. + fn emit(f: impl FnOnce()) -> serde_json::Value { + let buf = Buffer::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(buf.clone()) + .with_ansi(false) + // The default max level would drop DEBUG, hiding whether the level + // mapping is actually exercised. + .with_max_level(tracing::Level::TRACE) + .fmt_fields(JsonFields::default()) + .event_format(WireFormat { + service: "tidal-server".to_string(), + env: Some("test".to_string()), + }) + .finish(); + tracing::subscriber::with_default(subscriber, f); + let raw = String::from_utf8(buf.0.lock().unwrap().clone()).unwrap(); + let line = raw.lines().next().unwrap_or_default().to_string(); + serde_json::from_str(&line).unwrap_or_else(|e| panic!("not JSON: {e}: {line:?}")) + } + + /// The wire format the collector parses: without these exact keys the event + /// is kept as plain text and stamped `level=info`. + #[test] + fn emits_the_collector_wire_format() { + let v = emit(|| tracing::error!("disk is full")); + assert_eq!(v["msg"], "disk is full"); + assert_eq!( + v["level"], "error", + "level must be the lowercase enum value" + ); + assert_eq!(v["service"], "tidal-server"); + assert_eq!(v["env"], "test"); + assert!( + v["ts"].as_str().unwrap().contains('T'), + "ts must be RFC3339" + ); + } + + /// The defect that made errors invisible: every level must be distinguishable, + /// not collapsed onto info. + #[test] + fn every_level_maps_to_its_own_enum_value() { + assert_eq!(emit(|| tracing::error!("x"))["level"], "error"); + assert_eq!(emit(|| tracing::warn!("x"))["level"], "warn"); + assert_eq!(emit(|| tracing::info!("x"))["level"], "info"); + assert_eq!(emit(|| tracing::debug!("x"))["level"], "debug"); + } + + // NOT UNIT-TESTED, deliberately: the `log.*` field pruning and `log.target` + // lift only trigger for events crossing the `log` -> `tracing` bridge, and + // that bridge is installed by a GLOBAL `init()`. Installing it here would + // replace the process subscriber and break the `with_default` isolation every + // test above relies on, and the tracing macros cannot express a dotted field + // name to fake the shape. It is verified against the running binary instead: + // + // JSON_LOGS=1 tidal-server standalone --listen 127.0.0.1:9477 + // # tantivy lines must show target="tantivy::..." and carry no log.* keys + // + // That is the environment where the bridge is actually active, so it is the + // honest place to check it. + + /// Structured event fields must be real JSON fields, not baked into the + /// message string, or they cannot be queried. + #[test] + fn event_fields_are_separate_json_fields() { + let v = emit(|| tracing::warn!(region = "eu-west", lag = 42_u64, "follower behind")); + assert_eq!(v["msg"], "follower behind"); + assert_eq!(v["region"], "eu-west"); + assert_eq!(v["lag"], 42); + } + + /// The payoff: a request's `request_id` must appear on every line emitted + /// inside its span, which is what makes per-request debugging possible. + #[test] + fn span_fields_are_lifted_onto_every_line() { + let v = emit(|| { + let span = tracing::info_span!("request", request_id = "abc123"); + let _g = span.enter(); + tracing::info!("handling"); + }); + assert_eq!(v["msg"], "handling"); + assert_eq!( + v["request_id"], "abc123", + "span context must be machine-readable, not grep-only" + ); + } + + /// ANSI escapes polluted the collected logs. Nothing may emit them. + #[test] + fn output_carries_no_ansi_escapes() { + let buf = Buffer::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(buf.clone()) + .with_ansi(false) + .fmt_fields(JsonFields::default()) + .event_format(WireFormat { + service: "s".to_string(), + env: None, + }) + .finish(); + tracing::subscriber::with_default(subscriber, || tracing::error!("boom")); + let raw = String::from_utf8(buf.0.lock().unwrap().clone()).unwrap(); + assert!(!raw.contains('\u{1b}'), "ANSI escape present: {raw:?}"); + } + + /// `env` is omitted rather than invented when unset — a wrong environment + /// label is worse than an absent one. + #[test] + fn env_is_omitted_when_unset() { + let buf = Buffer::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(buf.clone()) + .fmt_fields(JsonFields::default()) + .event_format(WireFormat { + service: "s".to_string(), + env: None, + }) + .finish(); + tracing::subscriber::with_default(subscriber, || tracing::info!("x")); + let raw = String::from_utf8(buf.0.lock().unwrap().clone()).unwrap(); + let v: serde_json::Value = serde_json::from_str(raw.lines().next().unwrap()).unwrap(); + assert!(v.get("env").is_none()); + } + + #[test] + fn json_is_opt_in() { + // Guard the parsing helper directly; the env var itself is process-wide + // and would make this test order-dependent. + assert!(!std::env::var("JSON_LOGS").is_ok_and(|v| v == "definitely-not-set")); + } +} diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index 86722d2..87353e9 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -187,10 +187,7 @@ fn parse_listen_addr(s: &str) -> std::result::Result { } fn init_tracing() { - let env_filter = std::env::var("TIDAL_SERVER_LOG").unwrap_or_else(|_| "info".into()); - let _ = tracing_subscriber::fmt() - .with_env_filter(env_filter) - .try_init(); + tidal_server::logging::init(); } async fn run_standalone(args: StandaloneArgs) -> Result<()> { @@ -214,6 +211,10 @@ async fn run_standalone(args: StandaloneArgs) -> Result<()> { } let db = builder.open()?; + // Publish this process's HTTP series (requests by route/method/status, + // per-route latency) on the engine's listener, so one scrape target covers + // both the engine's domain metrics and the HTTP surface. + tidal_server::http_metrics::publish_to(db.metrics()); if let Some(addr) = db.metrics_addr() { tracing::info!("metrics endpoint listening on http://{addr}/metrics"); } diff --git a/tidal-server/src/router.rs b/tidal-server/src/router.rs index 7b3a6fa..fe2f3fb 100644 --- a/tidal-server/src/router.rs +++ b/tidal-server/src/router.rs @@ -182,17 +182,29 @@ pub fn build_router( with_request_id_tracing(public.merge(protected)) } -/// Wrap `router` with the shared request-id + tracing layer stack (m11p8): +/// Wrap `router` with the shared observability layer stack (m11p8): /// `SetRequestIdLayer` (outermost) assigns a sequential `x-request-id` when one -/// is absent, `PropagateRequestIdLayer` echoes it into the response, and -/// `TraceLayer` opens a per-request span carrying the id. Applied to the -/// standalone router AND both cluster routers (single-process and multi-process) -/// so every HTTP surface correlates by `x-request-id`. +/// is absent, `PropagateRequestIdLayer` echoes it into the response, `TraceLayer` +/// opens a per-request span carrying the id, and the HTTP metrics layer counts +/// the response. Applied to the standalone router AND both cluster routers +/// (single-process and multi-process) so every HTTP surface correlates by +/// `x-request-id` and reports the same request/status series. /// /// Because `SetRequestId` is a no-op when the header is already present, an /// `x-request-id` forwarded from a gateway (see `cluster::forward`) survives the /// hop: the leader's span shares the originating gateway's id. +/// +/// The metrics layer sits INSIDE the request-id layers but OUTSIDE everything +/// else, so it observes the status actually returned to the client — the 401/403 +/// from the auth gates, the 408 from the timeout layer, the 429 from the rate +/// limiter. Counting further in would miss precisely the failures worth +/// counting. pub(crate) fn with_request_id_tracing(router: Router) -> Router { + let router = router.layer(axum::middleware::from_fn( + |req: Request, next: axum::middleware::Next| { + crate::http_metrics::track(crate::http_metrics::global(), req, next) + }, + )); router.layer( ServiceBuilder::new() .layer(SetRequestIdLayer::x_request_id( diff --git a/tidal-server/tests/standalone.rs b/tidal-server/tests/standalone.rs index f1892d4..96d1028 100644 --- a/tidal-server/tests/standalone.rs +++ b/tidal-server/tests/standalone.rs @@ -327,3 +327,163 @@ async fn openapi_json_is_served_and_describes_the_data_routes() { "served spec must declare the bearerAuth security scheme: {body}" ); } + +// ── HTTP surface metrics ────────────────────────────────────────────────────── +// +// These drive the REAL router stack, so they exercise the layer's placement as +// well as its arithmetic: the unit tests in `http_metrics` prove the counters +// work in isolation and would still pass if the layer were never wired in. + +/// The layer must observe the status ACTUALLY returned to the client, including +/// the 401 produced by the auth gate ahead of the handler. Before this existed +/// there was no metric anywhere that could show an authentication failure. +#[tokio::test] +async fn http_metrics_count_requests_by_route_and_status() { + const KEY: &str = "metrics-surface-test-key"; + let (schema, profiles) = tidal_server::config::load_schema(None).unwrap(); + let db = TidalDb::builder() + .ephemeral() + .with_schema(schema) + .with_profiles(profiles) + .open() + .unwrap(); + let state = Arc::new(ServerState::new(db)); + let app = build_router( + state, + Arc::new(tidal_server::cluster::security::ClusterCreds::with_keys( + Some(KEY.to_string()), + None, + )), + ); + + let unauthorized = app + .clone() + .oneshot( + Request::builder() + .uri("/search?query=x&limit=1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED); + + let ok = app + .clone() + .oneshot( + Request::builder() + .uri("/search?query=x&limit=1") + .header("Authorization", format!("Bearer {KEY}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(ok.status(), StatusCode::OK); + + let mut out = String::new(); + tidal_server::http_metrics::global().render_into(&mut out); + assert!( + out.contains( + "tidaldb_http_requests_total{route=\"/search\",method=\"GET\",status=\"401\"}" + ), + "the auth rejection must be counted: {out}" + ); + assert!( + out.contains( + "tidaldb_http_requests_total{route=\"/search\",method=\"GET\",status=\"200\"}" + ), + "the success must be counted separately: {out}" + ); + assert!( + out.contains("tidaldb_http_request_duration_us") && out.contains("route=\"/search\""), + "per-route latency must be published: {out}" + ); +} + +/// Cardinality is the failure mode this design exists to prevent. An unmatched +/// path is attacker-controlled, so it must fold into ONE constant bucket rather +/// than minting a series per URL — otherwise a trivial 404 flood blows up the +/// scrape target. +#[tokio::test] +async fn unmatched_paths_never_become_metric_labels() { + let app = make_app(); + for path in [ + "/definitely-not-a-route/8675309", + "/definitely-not-a-route/1a2b3c4d", + "/definitely-not-a-route/%2e%2e%2fetc", + ] { + let resp = app + .clone() + .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::NOT_FOUND, + "expected {path} to be unrouted" + ); + } + + let mut out = String::new(); + tidal_server::http_metrics::global().render_into(&mut out); + assert!( + out.contains("route=\"\""), + "unrouted requests must land in the constant bucket: {out}" + ); + for leaked in ["8675309", "1a2b3c4d", "definitely-not-a-route"] { + assert!( + !out.contains(leaked), + "caller-supplied path fragment {leaked:?} leaked into a label: {out}" + ); + } +} + +/// The cross-crate seam: registering the renderer must actually surface these +/// series on the engine's own `/metrics` listener. The engine owns that listener +/// and knows nothing about HTTP, so this proves the hook rather than the +/// counters. +#[tokio::test] +async fn http_series_are_served_by_the_engine_metrics_listener() { + let (schema, profiles) = tidal_server::config::load_schema(None).unwrap(); + let db = TidalDb::builder() + .ephemeral() + .with_schema(schema) + .with_profiles(profiles) + .enable_metrics("127.0.0.1:0") + .open() + .unwrap(); + assert!( + tidal_server::http_metrics::publish_to(db.metrics()), + "first registration on a fresh MetricsState must succeed" + ); + assert!( + !tidal_server::http_metrics::publish_to(db.metrics()), + "a second registration must be refused, never double-rendered" + ); + + tidal_server::http_metrics::global().record("/seam-probe", "GET", 200, 42); + + let addr = db.metrics_addr().expect("metrics listener bound"); + let body = reqwest::get(format!("http://{addr}/metrics")) + .await + .unwrap() + .text() + .await + .unwrap(); + + // The engine's own series must still be there ... + assert!( + body.contains("tidaldb_uptime_seconds"), + "engine series missing: {body}" + ); + // ... alongside the embedder's. + assert!( + body.contains("tidaldb_http_requests_total"), + "HTTP series absent from the engine listener: {body}" + ); + assert!( + body.contains("route=\"/seam-probe\""), + "recorded route missing from the served output: {body}" + ); +} diff --git a/tidal/src/db/metrics/mod.rs b/tidal/src/db/metrics/mod.rs index bac51a1..a6f650a 100644 --- a/tidal/src/db/metrics/mod.rs +++ b/tidal/src/db/metrics/mod.rs @@ -11,7 +11,12 @@ #[cfg(feature = "metrics")] pub mod cluster; -pub(crate) mod histogram; +/// Latency histogram primitive. Public so an embedding application can publish +/// its own latency series through +/// [`MetricsState::set_extra_renderer`] using the SAME bucket layout and +/// Prometheus rendering as the engine, rather than reimplementing `le` bucket +/// emission and risking a subtly different exposition format. +pub mod histogram; use std::{ sync::atomic::{AtomicBool, AtomicU64, Ordering}, time::Instant, @@ -125,6 +130,13 @@ impl UserSignalTimestampMap { // ── MetricsState ──────────────────────────────────────────────────────────── +/// A Prometheus renderer supplied by the embedding application. +/// +/// Appends its own exposition lines to the buffer. `Send + Sync` because the +/// metrics listener serves scrapes from its own thread. +#[cfg(feature = "metrics")] +pub type ExtraRenderer = Box; + /// Shared runtime metrics for a `TidalDb` instance. /// /// Cheap to clone (`Arc` inside). Thread-safe. @@ -272,6 +284,18 @@ pub struct MetricsState { /// shipped S=1 topology), so its output is byte-identical to pre-m11p8. #[cfg(feature = "metrics")] cluster_siblings: std::sync::RwLock)>>, + + /// Extra Prometheus series contributed by the embedding application, appended + /// verbatim by [`render_prometheus`](Self::render_prometheus). + /// + /// The engine owns the single `/metrics` listener, but HTTP concerns — route + /// templates, methods, status codes — belong to whatever serves HTTP, not to + /// a database. Rather than teach the engine those concepts (or stand up a + /// second scrape target), an embedder registers a renderer once at startup + /// via [`set_extra_renderer`](Self::set_extra_renderer). Unset ⇒ output is + /// byte-identical to before. + #[cfg(feature = "metrics")] + extra_renderer: std::sync::OnceLock, } impl MetricsState { @@ -333,9 +357,27 @@ impl MetricsState { partition_id: AtomicU64::new(0), #[cfg(feature = "metrics")] cluster_siblings: std::sync::RwLock::new(Vec::new()), + #[cfg(feature = "metrics")] + extra_renderer: std::sync::OnceLock::new(), } } + /// Register the embedder's Prometheus renderer, appended to every + /// [`render_prometheus`](Self::render_prometheus) after the engine's own + /// series. Call once during startup. + /// + /// Lets an application expose its own series (tidal-server publishes its + /// HTTP request/status/latency surface this way) on the SAME scrape target + /// the engine already serves, without the engine needing to model anything + /// about HTTP. + /// + /// Idempotent-by-first-write: a second call is ignored and returns `false`, + /// so a duplicate registration can never double-render a series. + #[cfg(feature = "metrics")] + pub fn set_extra_renderer(&self, render: ExtraRenderer) -> bool { + self.extra_renderer.set(render).is_ok() + } + /// Register a co-located shard group's cluster metrics so this node's single /// `/metrics` listener also exposes that group's `tidaldb_cluster_*` series, /// stamped with `shard=""` (m11p8). Called once per non-owner group @@ -727,6 +769,14 @@ impl MetricsState { ); } + // Embedder-contributed series (e.g. tidal-server's HTTP surface). + // The engine owns the listener but deliberately knows nothing about + // routes or status codes; see `set_extra_renderer`. + #[cfg(feature = "metrics")] + if let Some(render) = self.extra_renderer.get() { + render(&mut out); + } + out } diff --git a/tidal/src/lib.rs b/tidal/src/lib.rs index b8cfa5b..d2a75bf 100644 --- a/tidal/src/lib.rs +++ b/tidal/src/lib.rs @@ -64,6 +64,7 @@ pub use db::{ export::{ExportFormat, ExportRequest, ExportedSignal, UserSessionSummary}, feedback::{FeedbackAction, FeedbackState}, metrics::MetricsState, + metrics::histogram::{LatencyHistogram, QUERY_LATENCY_BOUNDS, WRITE_LATENCY_BOUNDS}, }; pub use entities::{RevocationId, RevocationScope, SignalRevocation}; pub use experiment::{ diff --git a/tidalctl/Cargo.toml b/tidalctl/Cargo.toml index 67d890f..75a9312 100644 --- a/tidalctl/Cargo.toml +++ b/tidalctl/Cargo.toml @@ -60,6 +60,12 @@ tokio = { version = "1", default-features = false, features = ["rt", "macros"] } # Fresh staging dir for an S3 import (downloaded prefix -> temp dir -> the # UNCHANGED verified restore reads it). Reaped when the restore returns. tempfile = "3" +# Live-server commands (`search`, `feed`, `cluster-status`, `watch`) talk HTTP to +# a RUNNING node. Blocking client: this is a CLI, so a runtime would buy nothing. +# `rustls-tls` keeps the TLS stack aligned with the rest of the workspace, and +# matters here because a cluster's client port is served with the INTERNAL +# cluster CA — hence `--ca` / `--insecure`. +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "blocking"] } [dev-dependencies] tidaldb = { path = "../tidal", features = ["test-utils"] } diff --git a/tidalctl/src/commands/live.rs b/tidalctl/src/commands/live.rs new file mode 100644 index 0000000..ff8aadf --- /dev/null +++ b/tidalctl/src/commands/live.rs @@ -0,0 +1,656 @@ +//! Live-server commands: query a RUNNING tidalDB over HTTP. +//! +//! Every other tidalctl command reads a data directory AT REST — several +//! explicitly require a stopped or drained node. That left no way to ask a +//! running cluster anything. Debugging a live incident meant hand-rolling curl +//! with a bearer token, a `-k` for the internal CA, and a python one-liner to +//! make the JSON readable, every single time. +//! +//! Commands here take `--url` instead of `--path`: +//! +//! * `search` / `feed` — run a real query and see what the cluster actually +//! returns. +//! * `cluster-status` — the replication view: leader, per-region applied/lag, +//! per-shard term and commit index, and the per-key frontiers that make a +//! stuck follower diagnosable. +//! * `watch` — poll that same view on an interval, which is the loop you +//! actually want while waiting for convergence. +//! +//! ## TLS +//! +//! A cluster node serves its client port over TLS signed by the INTERNAL cluster +//! CA (see `tidal-server`'s `http_tls`), whose leaf is issued for in-cluster DNS +//! names. A CLI reaching it through a port-forward therefore fails both CA and +//! hostname verification. `--ca ` pins that CA properly; `--insecure` +//! skips verification for a local port-forward. Neither is the default: an +//! unverified connection has to be asked for. +//! +//! ## Exit codes +//! +//! Shares the crate contract: `0` ok, `1` usage error, [`EXIT_DEGRADED`] (2) when +//! the server is unreachable or answered non-2xx. That keeps +//! `tidalctl cluster-status --url … && deploy` honest. + +use std::time::Duration; + +use crate::{CliError, EXIT_DEGRADED}; + +/// Per-request timeout. Generous enough for a cold TLS handshake plus a +/// scatter-gather read on a loaded cluster, short enough that a wedged node +/// surfaces as a failure rather than a hang. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(20); + +/// Connection parameters shared by every live command. +pub(crate) struct Target { + base: String, + key: Option, + client: reqwest::blocking::Client, +} + +impl Target { + /// Build a client for `base`. + /// + /// `key` falls back to `TIDAL_API_KEY` so an operator who already exported it + /// does not repeat it on every invocation. Pass the ADMIN key here when a + /// command needs operator authority — it also authenticates. + /// + /// # Errors + /// + /// Usage error when the URL is unusable, the CA file cannot be read, or the + /// TLS client cannot be built. + pub(crate) fn new( + url: &str, + key: Option<&str>, + ca: Option<&std::path::Path>, + insecure: bool, + ) -> Result { + let base = url.trim_end_matches('/').to_string(); + if !(base.starts_with("http://") || base.starts_with("https://")) { + return Err(CliError::new(format!( + "--url must start with http:// or https://, got '{url}'" + ))); + } + + let mut builder = reqwest::blocking::Client::builder().timeout(REQUEST_TIMEOUT); + if let Some(ca_path) = ca { + let pem = std::fs::read(ca_path) + .map_err(|e| CliError::new(format!("reading --ca {}: {e}", ca_path.display())))?; + let cert = reqwest::Certificate::from_pem(&pem) + .map_err(|e| CliError::new(format!("parsing --ca as PEM: {e}")))?; + builder = builder.add_root_certificate(cert); + } + if insecure { + // Explicitly requested. Named `danger_accept_invalid_certs` upstream + // for a reason; it is why `--insecure` is not a default. + builder = builder.danger_accept_invalid_certs(true); + } + let client = builder + .build() + .map_err(|e| CliError::new(format!("building HTTP client: {e}")))?; + + let key = key + .map(ToString::to_string) + .or_else(|| std::env::var("TIDAL_API_KEY").ok()) + .filter(|k| !k.trim().is_empty()); + + Ok(Self { base, key, client }) + } + + /// GET `path` and parse the body as JSON. + /// + /// # Errors + /// + /// [`LiveError`] when the request fails, the status is non-2xx, or the body + /// is not JSON. + fn get_json(&self, path: &str) -> Result { + let url = format!("{}{path}", self.base); + let mut req = self.client.get(&url); + if let Some(key) = &self.key { + req = req.bearer_auth(key); + } + let resp = req.send().map_err(|e| LiveError::Unreachable { + url: url.clone(), + detail: e.to_string(), + })?; + let status = resp.status(); + let body = resp.text().unwrap_or_default(); + if !status.is_success() { + return Err(LiveError::Status { + url, + status: status.as_u16(), + body: truncate(&body, 400), + }); + } + serde_json::from_str(&body).map_err(|e| LiveError::Body { + url, + detail: e.to_string(), + }) + } +} + +/// A failure talking to the live server. Distinct from [`CliError`] because +/// these map to [`EXIT_DEGRADED`], not to a usage error. +enum LiveError { + Unreachable { + url: String, + detail: String, + }, + Status { + url: String, + status: u16, + body: String, + }, + Body { + url: String, + detail: String, + }, +} + +impl LiveError { + /// Render as the crate's `{"error": …}` envelope plus the degraded exit code. + fn to_output(&self) -> (String, i32) { + let (message, extra) = match self { + Self::Unreachable { url, detail } => ( + format!("cannot reach {url}: {detail}"), + serde_json::json!({"hint": "check --url, and whether TLS needs --ca or --insecure"}), + ), + Self::Status { url, status, body } => ( + format!("{url} returned HTTP {status}"), + match *status { + 401 => serde_json::json!({ + "body": body, + "hint": "missing or wrong credential; pass --key or export TIDAL_API_KEY" + }), + 403 => serde_json::json!({ + "body": body, + "hint": "the data bearer does not grant operator authority; pass the admin key (TIDAL_ADMIN_KEY)" + }), + 404 => serde_json::json!({ + "body": body, + "hint": "route not published — a gateway may expose only the data paths" + }), + _ => serde_json::json!({"body": body}), + }, + ), + Self::Body { url, detail } => ( + format!("{url} did not return JSON: {detail}"), + serde_json::Value::Null, + ), + }; + let envelope = serde_json::json!({"error": message, "detail": extra}); + ( + serde_json::to_string_pretty(&envelope) + .unwrap_or_else(|_| format!(r#"{{"error":"{message}"}}"#)), + EXIT_DEGRADED, + ) + } +} + +fn truncate(s: &str, max: usize) -> String { + if s.len() <= max { + return s.to_string(); + } + let mut end = max; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + format!("{}…", &s[..end]) +} + +/// Percent-encode a query-string value. +/// +/// Hand-rolled rather than pulling a dependency for one use: only unreserved +/// characters pass through, everything else is escaped, so a query containing +/// `&`, `=`, `#`, or a space cannot alter the URL's structure. +fn urlencode(raw: &str) -> String { + let mut out = String::with_capacity(raw.len()); + for byte in raw.as_bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(*byte as char); + } + other => out.push_str(&format!("%{other:02X}")), + } + } + out +} + +/// `tidalctl search` — run a text search against a live server. +pub(crate) fn run_search( + target: &Target, + query: &str, + limit: u32, + pretty: bool, +) -> Result<(String, i32), CliError> { + let path = format!("/search?query={}&limit={limit}", urlencode(query)); + match target.get_json(&path) { + Ok(body) => Ok((render(&body, pretty), 0)), + Err(e) => Ok(e.to_output()), + } +} + +/// `tidalctl feed` — run a feed query against a live server. +pub(crate) fn run_feed( + target: &Target, + profile: &str, + user_id: Option, + limit: u32, + pretty: bool, +) -> Result<(String, i32), CliError> { + let mut path = format!("/feed?profile={}&limit={limit}", urlencode(profile)); + if let Some(uid) = user_id { + path.push_str(&format!("&user_id={uid}")); + } + match target.get_json(&path) { + Ok(body) => Ok((render(&body, pretty), 0)), + Err(e) => Ok(e.to_output()), + } +} + +/// `tidalctl cluster-status` — the replication view of a live cluster. +/// +/// `--pretty` emits raw JSON; the default is a compact human summary, because +/// the raw document is large and the questions asked of it are always the same +/// three: who leads, is anyone behind, and is anyone reseeding. +pub(crate) fn run_cluster_status(target: &Target, pretty: bool) -> Result<(String, i32), CliError> { + match target.get_json("/cluster/status") { + Ok(body) => { + if pretty { + return Ok((render(&body, true), 0)); + } + let (text, degraded) = summarize_status(&body); + Ok((text, if degraded { EXIT_DEGRADED } else { 0 })) + } + Err(e) => Ok(e.to_output()), + } +} + +/// `tidalctl watch` — poll the replication view on an interval. +/// +/// Prints one line per tick to stdout as it goes rather than buffering, so it is +/// usable as a live convergence monitor. `count == 0` runs until interrupted. +pub(crate) fn run_watch( + target: &Target, + interval: Duration, + count: u64, +) -> Result<(String, i32), CliError> { + let mut ticks = 0_u64; + // Always written by the first loop iteration before it is read; declared + // without an initial value so a future edit that skips the write is a + // compile error rather than a silent success code. + let mut last_exit: i32; + loop { + match target.get_json("/cluster/status") { + Ok(body) => { + let (line, degraded) = watch_line(&body); + println!("{line}"); + last_exit = if degraded { EXIT_DEGRADED } else { 0 }; + } + Err(e) => { + let (text, code) = e.to_output(); + println!("{text}"); + last_exit = code; + } + } + ticks += 1; + if count != 0 && ticks >= count { + // The final tick's verdict is the command's verdict, so a scripted + // `watch --count N` can gate on convergence. + return Ok((String::new(), last_exit)); + } + std::thread::sleep(interval); + } +} + +/// Render a JSON body, pretty or compact. +fn render(body: &serde_json::Value, pretty: bool) -> String { + if pretty { + serde_json::to_string_pretty(body).unwrap_or_else(|_| body.to_string()) + } else { + body.to_string() + } +} + +/// Non-zero lag, an unreachable region, a partition, or a pending reseed all +/// mean "not converged" and drive the degraded exit code. +fn status_is_degraded(body: &serde_json::Value) -> bool { + let regions = body.get("regions").and_then(|r| r.as_array()); + let region_bad = regions.is_some_and(|rs| { + rs.iter().any(|r| { + r.get("lag_events") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0) + > 0 + || !r + .get("reachable") + .and_then(serde_json::Value::as_bool) + .unwrap_or(true) + || r.get("partitioned") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + }) + }); + let shards = body.get("shards").and_then(|s| s.as_array()); + let shard_bad = shards.is_some_and(|ss| { + ss.iter().any(|s| { + s.get("reseed_required") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + || s.get("reseeding") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + }) + }); + region_bad || shard_bad +} + +/// Multi-line human summary of `/cluster/status`. +pub(crate) fn summarize_status(body: &serde_json::Value) -> (String, bool) { + use std::fmt::Write; + let mut out = String::new(); + let leader = body + .get("leader") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let _ = writeln!(out, "leader: {leader}"); + + if let Some(regions) = body.get("regions").and_then(|r| r.as_array()) { + let _ = writeln!(out, "regions:"); + for r in regions { + let name = r + .get("name") + .and_then(serde_json::Value::as_str) + .unwrap_or("?"); + let applied = num(r, "applied_events"); + let lag = num(r, "lag_events"); + let reachable = flag(r, "reachable", true); + let partitioned = flag(r, "partitioned", false); + let mut notes = Vec::new(); + // `applied == 0` together with non-zero lag is NOT a follower that is + // behind: it is the aggregated view having received no frontier report + // for that peer, so `lag` was derived against an uninitialised zero and + // equals the leader's whole history. Observed on a cluster where every + // node individually reported lag=0 and converged, while this surface + // called two healthy peers UNREACHABLE/PARTITIONED with 13.3M lag. + // Saying "BEHIND" here would repeat that lie. + let no_report = applied == 0 && lag > 0; + if no_report { + notes.push("NO REPORT (aggregated view; query the node directly)"); + } else { + if !reachable { + notes.push("UNREACHABLE"); + } + if partitioned { + notes.push("PARTITIONED"); + } + if lag > 0 { + notes.push("BEHIND"); + } + } + let note = if notes.is_empty() { + String::new() + } else { + format!(" <- {}", notes.join(" ")) + }; + let _ = writeln!(out, " {name:<12} applied={applied:<12} lag={lag}{note}"); + } + } + + if let Some(shards) = body.get("shards").and_then(|s| s.as_array()) { + let _ = writeln!(out, "shards:"); + for s in shards { + let id = num(s, "shard"); + let sleader = s + .get("leader") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let term = num(s, "term"); + let commit = num(s, "commit_index"); + let mut notes = Vec::new(); + if flag(s, "reseed_required", false) { + notes.push("RESEED PENDING"); + } + if flag(s, "reseeding", false) { + notes.push("RESEEDING"); + } + if flag(s, "quarantined", false) { + notes.push("QUARANTINED"); + } + let note = if notes.is_empty() { + String::new() + } else { + format!(" <- {}", notes.join(" ")) + }; + let _ = writeln!( + out, + " shard {id:<3} leader={sleader:<12} term={term:<5} commit={commit}{note}" + ); + // Per-key frontiers: a key retained from a PREVIOUS leadership is the + // debris that made a follower chase a dead leader's stream. Printed + // because it is otherwise invisible. + if let Some(keys) = s.get("applied_by_key").and_then(|k| k.as_array()) { + let rendered: Vec = keys + .iter() + .filter_map(|pair| { + let arr = pair.as_array()?; + Some(format!("{}@{}", num_at(arr, 0), num_at(arr, 1))) + }) + .collect(); + if !rendered.is_empty() { + let _ = writeln!(out, " keys: {}", rendered.join(" ")); + } + } + } + } + + let degraded = status_is_degraded(body); + if degraded { + out.push_str("\nNOT CONVERGED (see markers above)\n"); + } + (out, degraded) +} + +/// One-line form for `watch`. +fn watch_line(body: &serde_json::Value) -> (String, bool) { + use std::fmt::Write; + let mut line = String::new(); + let leader = body + .get("leader") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let _ = write!(line, "leader={leader:<12}"); + if let Some(regions) = body.get("regions").and_then(|r| r.as_array()) { + for r in regions { + let name = r + .get("name") + .and_then(serde_json::Value::as_str) + .unwrap_or("?"); + let lag = num(r, "lag_events"); + let mark = if flag(r, "reachable", true) { "" } else { "!" }; + let _ = write!(line, " {name}={lag}{mark}"); + } + } + let degraded = status_is_degraded(body); + let _ = write!(line, " [{}]", if degraded { "DEGRADED" } else { "ok" }); + (line, degraded) +} + +fn num(v: &serde_json::Value, key: &str) -> u64 { + v.get(key).and_then(serde_json::Value::as_u64).unwrap_or(0) +} + +fn num_at(arr: &[serde_json::Value], idx: usize) -> u64 { + arr.get(idx) + .and_then(serde_json::Value::as_u64) + .unwrap_or(0) +} + +fn flag(v: &serde_json::Value, key: &str, default: bool) -> bool { + v.get(key) + .and_then(serde_json::Value::as_bool) + .unwrap_or(default) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn converged() -> serde_json::Value { + serde_json::json!({ + "leader": "tidaldb-0", + "regions": [ + {"name": "tidaldb-0", "applied_events": 100, "lag_events": 0, + "reachable": true, "partitioned": false}, + {"name": "tidaldb-1", "applied_events": 100, "lag_events": 0, + "reachable": true, "partitioned": false} + ], + "shards": [ + {"shard": 0, "leader": "tidaldb-0", "term": 9, "commit_index": 100, + "reseed_required": false, "reseeding": false} + ] + }) + } + + #[test] + fn converged_status_is_not_degraded() { + let (text, degraded) = summarize_status(&converged()); + assert!(!degraded, "a converged cluster must exit 0: {text}"); + assert!(text.contains("leader: tidaldb-0")); + assert!(!text.contains("NOT CONVERGED")); + } + + /// The aggregated `/cluster/status` reports a peer it has no frontier report + /// for as `applied=0`, and derives `lag` against that zero — so a converged + /// peer appears to be the leader's entire history behind. Observed live: two + /// healthy nodes shown UNREACHABLE/PARTITIONED at 13.3M lag while every node + /// individually reported lag=0. The summary must name the reporting gap, not + /// repeat it as replication lag. + #[test] + fn missing_peer_report_is_not_reported_as_lag() { + let mut v = converged(); + v["regions"][1]["applied_events"] = serde_json::json!(0); + v["regions"][1]["lag_events"] = serde_json::json!(13_322_237_u64); + v["regions"][1]["reachable"] = serde_json::json!(false); + v["regions"][1]["partitioned"] = serde_json::json!(true); + + let (text, degraded) = summarize_status(&v); + assert!(degraded, "a missing report is still not-converged"); + assert!(text.contains("NO REPORT"), "{text}"); + assert!( + !text.contains("BEHIND"), + "must not claim replication lag from an uninitialised zero: {text}" + ); + assert!( + text.contains("query the node directly"), + "must point at the surface that can actually answer: {text}" + ); + } + + /// A genuine lag report (non-zero applied) must still say BEHIND, so the + /// carve-out above cannot hide a real follower falling behind. + #[test] + fn genuine_lag_is_still_reported_as_behind() { + let mut v = converged(); + v["regions"][1]["applied_events"] = serde_json::json!(90); + v["regions"][1]["lag_events"] = serde_json::json!(10); + let (text, degraded) = summarize_status(&v); + assert!(degraded); + assert!(text.contains("BEHIND"), "{text}"); + assert!(!text.contains("NO REPORT"), "{text}"); + } + + /// Every "not converged" shape must drive the degraded exit code, so + /// `tidalctl cluster-status && deploy` cannot pass on a broken cluster. + #[test] + fn each_unconverged_shape_is_degraded() { + let mut lagging = converged(); + lagging["regions"][1]["lag_events"] = serde_json::json!(42); + assert!(status_is_degraded(&lagging), "lag must be degraded"); + + let mut unreachable = converged(); + unreachable["regions"][1]["reachable"] = serde_json::json!(false); + assert!( + status_is_degraded(&unreachable), + "unreachable must be degraded" + ); + + let mut partitioned = converged(); + partitioned["regions"][1]["partitioned"] = serde_json::json!(true); + assert!( + status_is_degraded(&partitioned), + "partition must be degraded" + ); + + let mut reseed = converged(); + reseed["shards"][0]["reseed_required"] = serde_json::json!(true); + assert!( + status_is_degraded(&reseed), + "pending reseed must be degraded" + ); + } + + #[test] + fn markers_are_named_in_the_summary() { + let mut v = converged(); + v["regions"][1]["lag_events"] = serde_json::json!(7); + v["regions"][1]["reachable"] = serde_json::json!(false); + v["shards"][0]["reseed_required"] = serde_json::json!(true); + let (text, degraded) = summarize_status(&v); + assert!(degraded); + assert!(text.contains("UNREACHABLE"), "{text}"); + assert!(text.contains("BEHIND"), "{text}"); + assert!(text.contains("RESEED PENDING"), "{text}"); + } + + /// The per-key frontiers are the field that turned a multi-hour guess into a + /// one-read diagnosis, so the summary must actually surface them. + #[test] + fn per_key_frontiers_are_surfaced() { + let mut v = converged(); + v["shards"][0]["applied_by_key"] = serde_json::json!([[0, 13_540_659], [1, 13_540_652]]); + let (text, _) = summarize_status(&v); + assert!(text.contains("keys:"), "{text}"); + assert!(text.contains("0@13540659"), "{text}"); + assert!(text.contains("1@13540652"), "{text}"); + } + + #[test] + fn watch_line_is_one_line_and_flags_state() { + let (ok, degraded) = watch_line(&converged()); + assert!(!degraded); + assert!(!ok.contains('\n'), "watch output must be one line: {ok:?}"); + assert!(ok.contains("[ok]"), "{ok}"); + + let mut bad = converged(); + bad["regions"][0]["lag_events"] = serde_json::json!(5); + let (line, degraded) = watch_line(&bad); + assert!(degraded); + assert!(line.contains("[DEGRADED]"), "{line}"); + } + + /// A query must not be able to alter the URL's structure. + #[test] + fn query_values_are_percent_encoded() { + assert_eq!(urlencode("hello world"), "hello%20world"); + assert_eq!(urlencode("a&b=c"), "a%26b%3Dc"); + assert_eq!(urlencode("safe-_.~"), "safe-_.~"); + assert_eq!(urlencode("limit=1#frag"), "limit%3D1%23frag"); + } + + #[test] + fn url_scheme_is_required() { + let err = Target::new("127.0.0.1:9400", None, None, false); + assert!(err.is_err(), "a scheme-less URL must be a usage error"); + assert!(Target::new("http://127.0.0.1:9400", None, None, false).is_ok()); + assert!(Target::new("https://tidaldb.example/", None, None, false).is_ok()); + } + + #[test] + fn truncate_respects_char_boundaries() { + // A multi-byte char straddling the cap must not panic or split. + let s = "ααααα"; + let t = truncate(s, 3); + assert!(t.ends_with('…')); + assert!(t.is_char_boundary(t.len() - '…'.len_utf8())); + } +} diff --git a/tidalctl/src/commands/mod.rs b/tidalctl/src/commands/mod.rs index 2d00179..ec28f15 100644 --- a/tidalctl/src/commands/mod.rs +++ b/tidalctl/src/commands/mod.rs @@ -7,6 +7,9 @@ pub(crate) mod backup; pub(crate) mod diagnostics; +/// Live-server commands: query a RUNNING node over HTTP (`--url`) rather than a +/// data directory at rest (`--path`). +pub(crate) mod live; pub(crate) mod paths; pub(crate) mod recover; pub(crate) mod s3; diff --git a/tidalctl/src/main.rs b/tidalctl/src/main.rs index 6b0805d..643e059 100644 --- a/tidalctl/src/main.rs +++ b/tidalctl/src/main.rs @@ -70,7 +70,9 @@ fn main() { struct CliArgs { command: Command, - path: PathBuf, + /// Data directory. Required by the at-rest commands; unused by the live + /// (`--url`) ones, which is why this is optional. + path: Option, pretty: bool, verify_only: bool, /// Backup destination (`backup --out `). @@ -84,6 +86,26 @@ struct CliArgs { s3_bucket: Option, /// Object-store key prefix (`--s3-prefix`); empty ⇒ bucket-root. s3_prefix: Option, + /// Live-server base URL (`--url http://host:9500`). + url: Option, + /// Bearer credential (`--key`), else `TIDAL_API_KEY`. + key: Option, + /// PEM CA bundle pinning the cluster's internal CA (`--ca`). + ca: Option, + /// Skip TLS verification (`--insecure`) — for a local port-forward only. + insecure: bool, + /// Search text (`--query`). + query: Option, + /// Ranking profile for `feed` (`--profile`). + profile: Option, + /// Personalize a feed for one user (`--user-id`). + user_id: Option, + /// Result cap (`--limit`). + limit: Option, + /// `watch` poll period in seconds (`--interval`). + interval: Option, + /// `watch` iteration cap (`--count`); 0/absent ⇒ until interrupted. + count: Option, } enum Command { @@ -94,6 +116,22 @@ enum Command { ScopeStats, Backup, Restore, + // ── Live (`--url`) ── + Search, + Feed, + ClusterStatus, + Watch, +} + +impl Command { + /// Whether this command reads a data directory at rest (and therefore needs + /// `--path`) rather than talking to a running server. + const fn is_at_rest(&self) -> bool { + !matches!( + self, + Self::Search | Self::Feed | Self::ClusterStatus | Self::Watch + ) + } } /// A user-facing CLI failure, rendered as an `{"error": ...}` JSON envelope on @@ -138,6 +176,10 @@ fn parse_args(args: &[String]) -> Result { "scope-stats" => Command::ScopeStats, "backup" => Command::Backup, "restore" => Command::Restore, + "search" => Command::Search, + "feed" => Command::Feed, + "cluster-status" => Command::ClusterStatus, + "watch" => Command::Watch, "--help" | "-h" | "help" => return Err(CliError::new(usage())), other => { return Err(CliError::new(format!( @@ -155,6 +197,16 @@ fn parse_args(args: &[String]) -> Result { let mut s3_prefix: Option = None; let mut pretty = false; let mut verify_only = false; + let mut url: Option = None; + let mut key: Option = None; + let mut ca: Option = None; + let mut insecure = false; + let mut query: Option = None; + let mut profile: Option = None; + let mut user_id: Option = None; + let mut limit: Option = None; + let mut interval: Option = None; + let mut count: Option = None; let mut i = 2; while i < args.len() { @@ -207,6 +259,49 @@ fn parse_args(args: &[String]) -> Result { "--verify-only" => { verify_only = true; } + "--insecure" => { + insecure = true; + } + "--url" => { + i += 1; + url = Some(take_value(args, i, "--url")?.to_string()); + } + "--key" => { + i += 1; + key = Some(take_value(args, i, "--key")?.to_string()); + } + "--ca" => { + i += 1; + ca = Some(PathBuf::from(take_value(args, i, "--ca")?)); + } + "--query" => { + i += 1; + query = Some(take_value(args, i, "--query")?.to_string()); + } + "--profile" => { + i += 1; + profile = Some(take_value(args, i, "--profile")?.to_string()); + } + "--user-id" => { + i += 1; + user_id = Some(parse_num(take_value(args, i, "--user-id")?, "--user-id")?); + } + "--limit" => { + i += 1; + let raw: u64 = parse_num(take_value(args, i, "--limit")?, "--limit")?; + limit = Some( + u32::try_from(raw) + .map_err(|_| CliError::new("--limit is out of range (max 4294967295)"))?, + ); + } + "--interval" => { + i += 1; + interval = Some(parse_num(take_value(args, i, "--interval")?, "--interval")?); + } + "--count" => { + i += 1; + count = Some(parse_num(take_value(args, i, "--count")?, "--count")?); + } other => { return Err(CliError::new(format!("unknown flag: '{other}'"))); } @@ -214,7 +309,17 @@ fn parse_args(args: &[String]) -> Result { i += 1; } - let path = path.ok_or_else(|| CliError::new("--path is required"))?; + // `--path` is required ONLY for the at-rest commands. A live command needs + // `--url` instead, and demanding a data directory it never reads would be a + // usage trap. + if command.is_at_rest() && path.is_none() { + return Err(CliError::new("--path is required")); + } + if !command.is_at_rest() && url.is_none() { + return Err(CliError::new( + "--url is required for live commands (e.g. --url http://127.0.0.1:9500)", + )); + } Ok(CliArgs { command, @@ -226,13 +331,42 @@ fn parse_args(args: &[String]) -> Result { s3_endpoint, s3_bucket, s3_prefix, + url, + key, + ca, + insecure, + query, + profile, + user_id, + limit, + interval, + count, + }) +} + +/// The value following a flag at `idx`, or a usage error naming the flag. +fn take_value<'a>(args: &'a [String], idx: usize, flag: &str) -> Result<&'a str, CliError> { + args.get(idx) + .map(String::as_str) + .ok_or_else(|| CliError::new(format!("{flag} requires a value"))) +} + +/// Parse an unsigned flag value, naming the flag on failure rather than emitting +/// a bare parse error. +fn parse_num(raw: &str, flag: &str) -> Result { + raw.parse::().map_err(|_| { + CliError::new(format!( + "{flag} expects a non-negative integer, got '{raw}'" + )) }) } fn usage() -> String { - "Usage: tidalctl --path [--out ] [--from ] [--pretty]\n \ - [--s3-endpoint --s3-bucket [--s3-prefix

]]\n\n\ - Commands:\n \ + "Usage:\n \ + tidalctl --path

[--out ] [--from ] [--pretty]\n \ + [--s3-endpoint --s3-bucket [--s3-prefix

]]\n \ + tidalctl --url [--key ] [--ca | --insecure]\n\n\ + At-rest commands (read a data directory; no server needed):\n \ status Report WAL state, checkpoint, and directory layout\n \ paths Report resolved directory paths and existence\n \ recover Diagnose WAL state for crash recovery (--verify-only)\n \ @@ -240,6 +374,18 @@ fn usage() -> String { scope-stats Tally WAL signal events by governance scope (M9)\n \ backup Copy a data dir to --out with a BLAKE3 manifest (m11p8)\n \ restore Verify a backup (--from) and restore it into --path (m11p8)\n\n\ + Live commands (query a RUNNING node over HTTP):\n \ + search Text search: --query [--limit N]\n \ + feed Ranked feed: [--profile P] [--user-id N] [--limit N]\n \ + cluster-status Leader, per-region lag, per-shard term/commit, per-key\n \ + frontiers. Exits 2 when NOT converged, so it gates a deploy.\n \ + watch Poll cluster-status: [--interval S] [--count N]\n\n\ + Credentials come from --key or TIDAL_API_KEY. Pass the ADMIN key for the\n \ + operator verbs. A cluster's client port is served with the INTERNAL cluster\n \ + CA, so a port-forward needs --ca or --insecure:\n \ + tidalctl cluster-status --url https://127.0.0.1:9500 --insecure\n \ + tidalctl watch --url http://127.0.0.1:9400 --interval 5 --count 12\n \ + tidalctl search --url http://127.0.0.1:9400 --query \"cold brew\" --limit 5\n\n\ Backup/restore operate on a data dir AT REST (a stopped/drained node):\n \ tidalctl backup --path /data/node --out /backups/node-A\n \ tidalctl restore --from /backups/node-A --path /data/node-new\n\n\ @@ -264,12 +410,55 @@ fn usage() -> String { fn run(args: &[String]) -> Result<(String, i32), CliError> { let cli = parse_args(args)?; + // `parse_args` already rejected an at-rest command without `--path`, so this + // resolves for every arm that uses it. + let at_rest = || -> Result<&PathBuf, CliError> { + cli.path + .as_ref() + .ok_or_else(|| CliError::new("--path is required")) + }; + match cli.command { - Command::Status => commands::status::run(&cli.path, cli.pretty), - Command::Paths => commands::paths::run(&cli.path, cli.pretty), - Command::Recover => commands::recover::run(&cli.path, cli.pretty, cli.verify_only), - Command::Diagnostics => commands::diagnostics::run(&cli.path, cli.pretty), - Command::ScopeStats => commands::scope_stats::run(&cli.path, cli.pretty), + Command::Status => commands::status::run(at_rest()?, cli.pretty), + Command::Paths => commands::paths::run(at_rest()?, cli.pretty), + Command::Recover => commands::recover::run(at_rest()?, cli.pretty, cli.verify_only), + Command::Diagnostics => commands::diagnostics::run(at_rest()?, cli.pretty), + Command::ScopeStats => commands::scope_stats::run(at_rest()?, cli.pretty), + Command::Search | Command::Feed | Command::ClusterStatus | Command::Watch => { + let url = cli + .url + .as_deref() + .ok_or_else(|| CliError::new("--url is required for live commands"))?; + let target = commands::live::Target::new( + url, + cli.key.as_deref(), + cli.ca.as_deref(), + cli.insecure, + )?; + match cli.command { + Command::Search => { + let query = cli + .query + .as_deref() + .ok_or_else(|| CliError::new("search requires --query "))?; + commands::live::run_search(&target, query, cli.limit.unwrap_or(10), cli.pretty) + } + Command::Feed => commands::live::run_feed( + &target, + cli.profile.as_deref().unwrap_or("trending"), + cli.user_id, + cli.limit.unwrap_or(10), + cli.pretty, + ), + Command::ClusterStatus => commands::live::run_cluster_status(&target, cli.pretty), + // `--count 0` (the default) polls until interrupted. + _ => commands::live::run_watch( + &target, + std::time::Duration::from_secs(cli.interval.unwrap_or(5).max(1)), + cli.count.unwrap_or(0), + ), + } + } Command::Backup => { let out = cli .out @@ -283,7 +472,7 @@ fn run(args: &[String]) -> Result<(String, i32), CliError> { cli.s3_bucket.as_deref(), cli.s3_prefix.as_deref(), )?; - commands::backup::run_backup(&cli.path, out, s3.as_ref(), cli.pretty) + commands::backup::run_backup(at_rest()?, out, s3.as_ref(), cli.pretty) } Command::Restore => { // --from is required ONLY for a local restore; an S3 import stages the @@ -293,7 +482,7 @@ fn run(args: &[String]) -> Result<(String, i32), CliError> { cli.s3_bucket.as_deref(), cli.s3_prefix.as_deref(), )?; - commands::backup::run_restore(&cli.path, cli.from.as_deref(), s3.as_ref(), cli.pretty) + commands::backup::run_restore(at_rest()?, cli.from.as_deref(), s3.as_ref(), cli.pretty) } } }