feat(server): user metrics
This commit is contained in:
RainbowBird
@@ -211,6 +211,16 @@ span name 目前允许保留业务可读格式，例如：
|
||||
|
||||
历史教训：`user.active_sessions` 最早是 UpDownCounter，登录 +1 / 登出 -1。但 Better Auth 的 session TTL 过期不会调 delete hook，counter 单实例就漂；多副本登录在 A、登出在 B 直接撕裂。改成 `ObservableGauge` 后由 [apps/server/src/app.ts](/apps/server/src/app.ts) 的 `registerActiveSessionsGauge` 通过 `SELECT COUNT(*) FROM session WHERE expires_at > NOW()` 在 scrape 时按需查 DB，带 10s 内存缓存避免 hammer。
|
||||
|
||||
三个 DB-backed user gauge 的语义区分（都是 cluster-wide，dashboard 用 `max()` / `avg()`）：
|
||||
|
||||
| Metric | 含义 | 来源 | 注册位置 |
|
||||
|---|---|---|---|
|
||||
| `user.active_sessions` | 当前未过期的 session **行数**（含 OIDC token 刷新产生的行，会膨胀） | `COUNT(*) FROM session WHERE expires_at > now()` | `registerActiveSessionsGauge` |
|
||||
| `user.distinct_active` | 当前持有 ≥1 个未过期 session 的**去重用户数**（"此刻在线"） | `COUNT(DISTINCT user_id) FROM session WHERE expires_at > now()` | `registerDistinctActiveUsersGauge` |
|
||||
| `user.active_rolling` | 滚动窗口去重活跃用户 DAU/WAU/MAU（"近 N 天回来过"），按 `window="24h"\|"7d"\|"30d"` 打 label | `COUNT(*) FILTER (WHERE last_seen_at > now()-window) FROM user` | `registerRollingActiveUsersGauge` |
|
||||
|
||||
`user.active_rolling` 用 `user.last_seen_at`（登录 + 每次 OIDC token 刷新约每小时 touch 一次，是 per-user 的「最后活跃」时间戳），不依赖 session 是否过期，所以能回答「本周回来过多少人」。一次 query 用三个 `FILTER` 把三个窗口算完，缓存 60s（窗口变化慢且要全表扫 `user`，TTL 比 session gauge 长）。
|
||||
|
||||
### Dashboard 查询模板
|
||||
|
||||
加新 panel 时按这个清单核对：
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -91,6 +91,17 @@ interface StatPanelOpts {
|
||||
decimals?: number
|
||||
noValue?: string
|
||||
graphMode?: 'area' | 'none'
|
||||
/**
|
||||
* Stat visual language:
|
||||
* - 'health' (default) — traffic-light colour driven by `steps`, no trend
|
||||
* delta. For numbers that are good or bad (req/s, 5xx, unbilled flux).
|
||||
* - 'count' — neutral fixed colour + period-over-period % delta. For pure
|
||||
* informational counts/totals with no good/bad threshold (active users,
|
||||
* DAU/WAU, revenue, tokens consumed).
|
||||
*/
|
||||
variant?: 'health' | 'count'
|
||||
/** Fixed colour for the 'count' variant. Ignored by 'health'. @default 'blue' */
|
||||
color?: string
|
||||
}
|
||||
|
||||
interface GaugePanelOpts {
|
||||
@@ -134,7 +145,23 @@ function defaultsBlock({ unit, steps, decimals, noValue, min, max }: DefaultsBlo
|
||||
}
|
||||
|
||||
function statPanel(id: number, title: string, description: string, queries: PanelQuery[], opts: StatPanelOpts = {}) {
|
||||
const { unit = 'short', steps = [{ color: 'green', value: 0 }], decimals, noValue, graphMode = 'area' } = opts
|
||||
const { unit = 'short', steps = [{ color: 'green', value: 0 }], decimals, noValue, graphMode = 'area', variant = 'health', color = 'blue' } = opts
|
||||
const isCount = variant === 'count'
|
||||
|
||||
// 'count' stats drop the traffic-light colouring (the value is neither good
|
||||
// nor bad) and instead surface a period-over-period % delta so the trend is
|
||||
// readable at a glance. 'health' keeps threshold colouring and no delta.
|
||||
const defaults = isCount
|
||||
? {
|
||||
color: { mode: 'fixed', fixedColor: color },
|
||||
fieldMinMax: false,
|
||||
thresholds: thresholds([{ color, value: 0 }]),
|
||||
unit,
|
||||
...(decimals != null && { decimals }),
|
||||
...(noValue != null && { noValue }),
|
||||
}
|
||||
: defaultsBlock({ unit, steps, decimals, noValue })
|
||||
|
||||
return {
|
||||
kind: 'Panel',
|
||||
spec: {
|
||||
@@ -147,16 +174,16 @@ function statPanel(id: number, title: string, description: string, queries: Pane
|
||||
group: 'stat',
|
||||
kind: 'VizConfig',
|
||||
spec: {
|
||||
fieldConfig: { defaults: defaultsBlock({ unit, steps, decimals, noValue }), overrides: [] },
|
||||
fieldConfig: { defaults, overrides: [] },
|
||||
options: {
|
||||
colorMode: 'value',
|
||||
colorMode: isCount ? 'none' : 'value',
|
||||
graphMode,
|
||||
justifyMode: 'auto',
|
||||
orientation: 'auto',
|
||||
percentChangeColorMode: 'standard',
|
||||
reduceOptions: { calcs: ['lastNotNull'], fields: '', values: false },
|
||||
showPercentChange: false,
|
||||
textMode: 'auto',
|
||||
showPercentChange: isCount,
|
||||
textMode: isCount ? 'value_and_name' : 'auto',
|
||||
wideLayout: true,
|
||||
},
|
||||
},
|
||||
@@ -300,6 +327,55 @@ function timeseriesPanel(id: number, title: string, description: string, queries
|
||||
}
|
||||
}
|
||||
|
||||
interface HeatmapPanelOpts {
|
||||
unit?: string
|
||||
}
|
||||
|
||||
// Status-code-over-time heatmap: each `sum by (label)` series becomes a Y-axis
|
||||
// row, colour encodes the rate at each time bucket. `calculate: false` means
|
||||
// the series are treated as pre-bucketed rows (one row per status code) rather
|
||||
// than re-binned by value. Reads the traffic mix at a glance — a sudden 5xx
|
||||
// row lighting up is obvious in a way a stacked line chart hides.
|
||||
function heatmapPanel(id: number, title: string, description: string, queries: PanelQuery[], opts: HeatmapPanelOpts = {}) {
|
||||
const { unit = 'short' } = opts
|
||||
return {
|
||||
kind: 'Panel',
|
||||
spec: {
|
||||
data: { kind: 'QueryGroup', spec: { queries, queryOptions: {}, transformations: [] } },
|
||||
description,
|
||||
id,
|
||||
links: [],
|
||||
title,
|
||||
vizConfig: {
|
||||
group: 'heatmap',
|
||||
kind: 'VizConfig',
|
||||
spec: {
|
||||
fieldConfig: {
|
||||
defaults: {
|
||||
custom: { hideFrom: { legend: false, tooltip: false, viz: false }, scaleDistribution: { type: 'linear' } },
|
||||
unit,
|
||||
},
|
||||
overrides: [],
|
||||
},
|
||||
options: {
|
||||
annotations: { clustering: -1, multiLane: false },
|
||||
calculate: false,
|
||||
cellGap: 1,
|
||||
color: { exponent: 0.5, fill: 'dark-orange', mode: 'scheme', reverse: false, scale: 'exponential', scheme: 'RdYlBu', steps: 64 },
|
||||
exemplars: { color: 'rgba(255,0,255,0.7)' },
|
||||
filterValues: { le: 1e-9 },
|
||||
legend: { show: false },
|
||||
rowsFrame: { layout: 'auto' },
|
||||
tooltip: { mode: 'single', showColorScale: false, yHistogram: false },
|
||||
yAxis: { axisPlacement: 'left', reverse: false },
|
||||
},
|
||||
},
|
||||
version: SCHEMA_VERSION,
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function logsPanel(id: number, title: string, description: string, expr: string) {
|
||||
return {
|
||||
kind: 'Panel',
|
||||
@@ -372,17 +448,17 @@ const elements: Record<string, unknown> = {}
|
||||
elements['panel-1'] = statPanel(
|
||||
1,
|
||||
'New Users 24h',
|
||||
'Rolling 24h `increase(user.registered)` — counts the Better Auth `databaseHooks.user.create.after` fires over the last 24 hours. Operational signup signal; for DAU / WAU / MAU query PostHog (`event = session_started`).',
|
||||
'Rolling 24h `increase(user.registered)` — counts the Better Auth `databaseHooks.user.create.after` fires over the last 24 hours. The signup half of the funnel; the returning-user half is DAU / WAU / MAU in the Users & Engagement row.',
|
||||
[query(`sum(increase(user_registered_total{${SERVICE_FILTER}}[24h]))`, 'new users')],
|
||||
{ unit: 'short', steps: [{ color: 'green', value: 0 }, { color: 'yellow', value: 1000 }] },
|
||||
{ unit: 'short', variant: 'count' },
|
||||
)
|
||||
|
||||
elements['panel-15'] = statPanel(
|
||||
15,
|
||||
'Active Sessions',
|
||||
'COUNT(*) over the Better Auth `session` table where `expires_at > now()`. Counts session **rows**, not users — divide by panel-1 to spot row inflation.',
|
||||
'COUNT(*) over the Better Auth `session` table where `expires_at > now()`, aggregated with `avg()` (cluster-wide gauge). Counts session **rows**, not users — compare against DAU to spot session-row inflation.',
|
||||
[query(`avg(user_active_sessions{${SERVICE_FILTER}})`, 'sessions')],
|
||||
{ unit: 'short', steps: [{ color: 'green', value: 0 }, { color: 'yellow', value: 5000 }] },
|
||||
{ unit: 'short', variant: 'count' },
|
||||
)
|
||||
|
||||
elements['panel-3'] = statPanel(
|
||||
@@ -412,15 +488,39 @@ elements['panel-5'] = statPanel(
|
||||
{ unit: 'reqps', decimals: 2 },
|
||||
)
|
||||
|
||||
elements['panel-6'] = gaugePanel(
|
||||
6,
|
||||
'Email Failure %',
|
||||
'Email failures ÷ total attempts over the last 5m. >5% means Resend / DNS / suppression-list problems blocking auth flows.',
|
||||
[query(
|
||||
`100 * sum(rate(airi_email_failures_total{${SERVICE_FILTER}}[5m])) / clamp_min(sum(rate(airi_email_send_total{${SERVICE_FILTER}}[5m])) + sum(rate(airi_email_failures_total{${SERVICE_FILTER}}[5m])), 1)`,
|
||||
'fail %',
|
||||
)],
|
||||
{ steps: [{ color: 'green', value: 0 }, { color: 'yellow', value: 1 }, { color: 'red', value: 5 }], max: 20, decimals: 1, noValue: '0' },
|
||||
// --- Users & Engagement: DAU/WAU/MAU + sessions + live WebSocket presence ---
|
||||
// DAU/WAU/MAU come from the `user.active_rolling` gauge (COUNT(*) over `user`
|
||||
// filtered by last_seen_at; one series per window). Cluster-wide gauge — every
|
||||
// replica reports the same value, so aggregate with max(), NOT sum().
|
||||
const ROLLING_USERS = [
|
||||
{ id: 80, window: '24h', title: 'DAU', label: 'Daily', span: 'last 24h' },
|
||||
{ id: 81, window: '7d', title: 'WAU', label: 'Weekly', span: 'last 7d' },
|
||||
{ id: 82, window: '30d', title: 'MAU', label: 'Monthly', span: 'last 30d' },
|
||||
] as const
|
||||
for (const { id, window, title, label, span } of ROLLING_USERS) {
|
||||
elements[`panel-${id}`] = statPanel(
|
||||
id,
|
||||
title,
|
||||
`${label} active users — distinct users with activity in the ${span}. Sourced from \`user.last_seen_at\` (touched on sign-in and every OIDC token refresh) via the \`user.active_rolling\` gauge. Cluster-wide gauge aggregated with \`max()\`.`,
|
||||
[query(`max(user_active_rolling{${SERVICE_FILTER}, window="${window}"})`, title)],
|
||||
{ unit: 'short', variant: 'count', noValue: '0' },
|
||||
)
|
||||
}
|
||||
|
||||
elements['panel-93'] = statPanel(
|
||||
93,
|
||||
'WS Online',
|
||||
'Current concurrent WebSocket connections across all replicas (`sum` — each replica holds its own connections). The live-presence counterpart to the rolling DAU/WAU windows.',
|
||||
[query(`sum(ws_connections_active{${SERVICE_FILTER}})`, 'online')],
|
||||
{ unit: 'short', variant: 'count', color: 'purple', noValue: '0' },
|
||||
)
|
||||
|
||||
elements['panel-92'] = timeseriesPanel(
|
||||
92,
|
||||
'WS Connections',
|
||||
'Concurrent WebSocket connections over time (`sum` across replicas). A cliff to zero with no matching deploy = mass disconnect (LB drop, network blackhole); a slow ramp without disconnects = connection leak.',
|
||||
[query(`sum(ws_connections_active{${SERVICE_FILTER}})`, 'connections')],
|
||||
{ unit: 'short', fillOpacity: 30 },
|
||||
)
|
||||
|
||||
// --- Row 2: HTTP — traffic ranking, error trend, latency trend -------------
|
||||
@@ -438,23 +538,15 @@ elements['panel-16'] = barGaugePanel(
|
||||
{ unit: 'short' },
|
||||
)
|
||||
|
||||
elements['panel-40'] = timeseriesPanel(
|
||||
elements['panel-40'] = heatmapPanel(
|
||||
40,
|
||||
'Error Rate %',
|
||||
'Error rate as a percentage of total non-OPTIONS HTTP traffic — 4xx (client-side: validation, auth, missing routes) and 5xx (server-side) over the same denominator.',
|
||||
[
|
||||
query(
|
||||
`100 * sum(rate(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS", http_response_status_code=~"4.."}[$__rate_interval])) / clamp_min(sum(rate(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS"}[$__rate_interval])), 1)`,
|
||||
'4xx %',
|
||||
'A',
|
||||
),
|
||||
query(
|
||||
`100 * sum(rate(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS", http_response_status_code=~"5.."}[$__rate_interval])) / clamp_min(sum(rate(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS"}[$__rate_interval])), 1)`,
|
||||
'5xx %',
|
||||
'B',
|
||||
),
|
||||
],
|
||||
{ unit: 'percent' },
|
||||
'HTTP status-code mix over time, one row per status code, colour = request rate in each time bucket. The 200 row dominates in steady state; a 5xx / 4xx row suddenly lighting up flags an incident at a glance. Non-OPTIONS traffic only.',
|
||||
[query(
|
||||
`sum by (http_response_status_code) (rate(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS"}[$__rate_interval]))`,
|
||||
'{{http_response_status_code}}',
|
||||
)],
|
||||
{ unit: 'short' },
|
||||
)
|
||||
|
||||
elements['panel-20'] = timeseriesPanel(
|
||||
@@ -468,6 +560,17 @@ elements['panel-20'] = timeseriesPanel(
|
||||
{ unit: 's' },
|
||||
)
|
||||
|
||||
elements['panel-94'] = timeseriesPanel(
|
||||
94,
|
||||
'Errors by Route',
|
||||
'Error responses per route, broken out by status code. Excludes success (2xx/3xx) and the expected-client-error codes 401/402/404 (auth-required / payment-required / not-found noise) so the curve isolates real failures: 4xx like 400/403/422/429 and all 5xx. The per-route companion to the aggregate Error Rate % stat.',
|
||||
[query(
|
||||
`sum by (http_route, http_response_status_code) (increase(http_server_request_duration_seconds_count{${SERVICE_FILTER}, http_request_method!="OPTIONS", http_response_status_code!~"2..|3..|401|402|404"}[$__rate_interval]))`,
|
||||
'{{http_response_status_code}} {{http_route}}',
|
||||
)],
|
||||
{ unit: 'short' },
|
||||
)
|
||||
|
||||
// --- Row 3: LLM Gateway — request mix, latency, billed usage ---------------
|
||||
elements['panel-11'] = timeseriesPanel(
|
||||
11,
|
||||
@@ -511,7 +614,7 @@ elements['panel-73'] = statPanel(
|
||||
query(`sum(increase(gen_ai_client_token_usage_input_total{${SERVICE_FILTER}}[$__range]))`, 'input', 'A'),
|
||||
query(`sum(increase(gen_ai_client_token_usage_output_total{${SERVICE_FILTER}}[$__range]))`, 'output', 'B'),
|
||||
],
|
||||
{ unit: 'short', noValue: '0', graphMode: 'none' },
|
||||
{ unit: 'short', variant: 'count', noValue: '0', graphMode: 'none' },
|
||||
)
|
||||
|
||||
elements['panel-71'] = timeseriesPanel(
|
||||
@@ -599,7 +702,7 @@ elements['panel-30'] = statPanel(
|
||||
`sum by (currency) (increase(airi_stripe_revenue_minor_unit_total{${SERVICE_FILTER}, currency!=""}[$__range])) / 100`,
|
||||
'{{currency}}',
|
||||
)],
|
||||
{ unit: 'short', decimals: 2, noValue: '—' },
|
||||
{ unit: 'short', variant: 'count', color: 'green', decimals: 2, noValue: '—' },
|
||||
)
|
||||
|
||||
elements['panel-31'] = gaugePanel(
|
||||
@@ -621,7 +724,7 @@ elements['panel-32'] = statPanel(
|
||||
`sum by (event_type) (increase(stripe_events_total{${SERVICE_FILTER}, event_type!=""}[$__range]))`,
|
||||
'{{event_type}}',
|
||||
)],
|
||||
{ unit: 'short', noValue: '—', graphMode: 'none' },
|
||||
{ unit: 'short', variant: 'count', noValue: '—', graphMode: 'none' },
|
||||
)
|
||||
|
||||
// --- Row 7: Infrastructure (collapsed) — process / DB health ---------------
|
||||
@@ -689,26 +792,40 @@ elements['panel-90'] = logsPanel(
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const rows = [
|
||||
// Row 1: six health stats/gauges, each 4 wide (4×6=24).
|
||||
// Row 1: Service Health — two rows of glance stats + the status-code heatmap
|
||||
// standing tall on the right, with the live WS-connections trend full-width
|
||||
// underneath. counts (New Users / Active Sessions / WS Online) read blue with
|
||||
// a trend delta; req-rate + 5xx stay traffic-light.
|
||||
row('Service Health', [
|
||||
item('panel-1', 0, 0, 4, 4),
|
||||
item('panel-15', 4, 0, 4, 4),
|
||||
item('panel-3', 8, 0, 4, 4),
|
||||
item('panel-4', 12, 0, 4, 4),
|
||||
item('panel-5', 16, 0, 4, 4),
|
||||
item('panel-6', 20, 0, 4, 4),
|
||||
item('panel-1', 0, 0, 6, 4),
|
||||
item('panel-3', 6, 0, 6, 4),
|
||||
item('panel-4', 12, 0, 6, 4),
|
||||
item('panel-40', 18, 0, 6, 8),
|
||||
item('panel-15', 0, 4, 6, 4),
|
||||
item('panel-5', 6, 4, 6, 4),
|
||||
item('panel-93', 12, 4, 6, 4),
|
||||
item('panel-92', 0, 8, 24, 5),
|
||||
]),
|
||||
// Row 2: HTTP — request ranking (bar), error trend, latency trend.
|
||||
// Row 2: User Engagement — rolling-window active users (DAU/WAU/MAU) from
|
||||
// user.last_seen_at. Kept its own row so it can grow (retention, cohorts)
|
||||
// without crowding the health glance above.
|
||||
row('User Engagement', [
|
||||
item('panel-80', 0, 0, 8, 4),
|
||||
item('panel-81', 8, 0, 8, 4),
|
||||
item('panel-82', 16, 0, 8, 4),
|
||||
]),
|
||||
// Row 3: HTTP — full-width error breakdown on top, then traffic ranking +
|
||||
// latency trend side by side.
|
||||
row('HTTP', [
|
||||
item('panel-16', 0, 0, 8, 8),
|
||||
item('panel-40', 8, 0, 8, 8),
|
||||
item('panel-20', 16, 0, 8, 8),
|
||||
item('panel-94', 0, 0, 24, 8),
|
||||
item('panel-16', 0, 8, 7, 11),
|
||||
item('panel-20', 7, 8, 17, 11),
|
||||
]),
|
||||
// Row 3: LLM gateway — request mix, latency, billed flux.
|
||||
// Row 4: LLM gateway — billed flux + latency side by side, request mix below.
|
||||
row('LLM Gateway', [
|
||||
item('panel-11', 0, 0, 8, 8),
|
||||
item('panel-21', 8, 0, 8, 8),
|
||||
item('panel-72', 16, 0, 8, 8),
|
||||
item('panel-72', 0, 0, 13, 8),
|
||||
item('panel-21', 13, 0, 11, 8),
|
||||
item('panel-11', 0, 8, 24, 8),
|
||||
]),
|
||||
// Row 4: token totals + throughput + the two revenue/quality alert stats.
|
||||
row('LLM Tokens & Quality', [
|
||||
@@ -809,14 +926,16 @@ const variables = [
|
||||
* AIRI Server Overview dashboard.
|
||||
*
|
||||
* Reading order:
|
||||
* 1. Service Health — six gauges/stats, "is everything OK right now?"
|
||||
* 2. HTTP — request ranking, error rate, latency by route
|
||||
* 3. LLM Gateway — request mix, latency (TTFB + end-to-end), billed flux
|
||||
* 4. LLM Tokens & Quality — token totals/throughput, revenue-leak alerts
|
||||
* 5. LLM Router Health — key/decrypt/fallback "wake someone up" signals
|
||||
* 6. Business — Stripe / Flux money flow
|
||||
* 7. Infrastructure (collapsed) — DB / runtime health for triage
|
||||
* 8. Logs — Loki for live debugging
|
||||
* 1. Service Health — signup/sessions/WS counts, req-rate, 5xx, status-code
|
||||
* heatmap, live WS trend: "is anything broken right now?"
|
||||
* 2. User Engagement — rolling DAU/WAU/MAU from user.last_seen_at
|
||||
* 3. HTTP — error breakdown by route, request ranking, latency by route
|
||||
* 4. LLM Gateway — billed flux, latency (TTFB + end-to-end), request mix
|
||||
* 5. LLM Tokens & Quality — token totals/throughput, revenue-leak alerts
|
||||
* 6. LLM Router Health — key/decrypt/fallback "wake someone up" signals
|
||||
* 7. Business — Stripe / Flux money flow
|
||||
* 8. Infrastructure (collapsed) — DB / runtime health for triage
|
||||
* 9. Logs — Loki for live debugging
|
||||
*
|
||||
* One metric, one panel: we deliberately do not duplicate a metric across
|
||||
* stat/trend/bar/pie forms. Counter conventions: rate() for "now" trends,
|
||||
|
||||
@@ -48,6 +48,7 @@ import { sessionMiddleware } from './middlewares/auth'
|
||||
import { emitOtelLog, initOtel } from './otel'
|
||||
import { registerActiveSessionsGauge } from './otel/gauges/active-sessions'
|
||||
import { registerDistinctActiveUsersGauge } from './otel/gauges/distinct-active-users'
|
||||
import { registerRollingActiveUsersGauge } from './otel/gauges/rolling-active-users'
|
||||
import { createAdminRouterConfigRoutes } from './routes/admin/config/router'
|
||||
import { createAdminFluxGrantsRoutes } from './routes/admin/flux-grants'
|
||||
import { createAdminUsersRoutes } from './routes/admin/users'
|
||||
@@ -703,6 +704,7 @@ export async function createApp() {
|
||||
if (resolved.otel) {
|
||||
registerActiveSessionsGauge(resolved.otel.auth.activeSessions, resolved.db, resolved.otel.observability.metricReadErrors)
|
||||
registerDistinctActiveUsersGauge(resolved.otel.auth.distinctActiveUsers, resolved.db, resolved.otel.observability.metricReadErrors)
|
||||
registerRollingActiveUsersGauge(resolved.otel.auth.rollingActiveUsers, resolved.db, resolved.otel.observability.metricReadErrors)
|
||||
}
|
||||
|
||||
const { app, injectWebSocket } = await buildApp({
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
import type { AuthMetrics, ObservabilityMetrics } from '..'
|
||||
import type { Database } from '../../libs/db'
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import { registerRollingActiveUsersGauge } from './rolling-active-users'
|
||||
|
||||
/**
|
||||
* Capture the callback registered via `gauge.addCallback` and a spyable
|
||||
* `observe` so tests can drive collection cycles by hand.
|
||||
*
|
||||
* @example
|
||||
* const { gauge, observe, run } = makeGauge()
|
||||
* register...(gauge, db, errs)
|
||||
* await run()
|
||||
* expect(observe).toHaveBeenCalledWith(5, { window: '24h' })
|
||||
*/
|
||||
function makeGauge() {
|
||||
let cb: ((result: { observe: (v: number, attrs: Record<string, string>) => void }) => void | Promise<void>) | null = null
|
||||
const observe = vi.fn()
|
||||
const gauge = {
|
||||
addCallback: vi.fn((fn: typeof cb) => { cb = fn }),
|
||||
} as unknown as AuthMetrics['rollingActiveUsers']
|
||||
return {
|
||||
gauge,
|
||||
observe,
|
||||
run: async () => {
|
||||
if (!cb)
|
||||
throw new Error('no callback registered')
|
||||
await cb({ observe })
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drizzle's `db.select({...}).from(table)` is awaited directly (thenable
|
||||
* query builder). Mock it as `select -> { from: () => Promise<rows> }`.
|
||||
*
|
||||
* @example
|
||||
* const db = makeDb([{ dau: '5', wau: '10', mau: '20' }])
|
||||
*/
|
||||
function makeDb(rows: unknown[], opts: { reject?: boolean } = {}) {
|
||||
const from = vi.fn(() => (opts.reject ? Promise.reject(new Error('db down')) : Promise.resolve(rows)))
|
||||
const select = vi.fn(() => ({ from }))
|
||||
return { db: { select } as unknown as Database, select, from }
|
||||
}
|
||||
|
||||
function makeReadErrors() {
|
||||
const add = vi.fn()
|
||||
return { metricReadErrors: { add } as unknown as ObservabilityMetrics['metricReadErrors'], add }
|
||||
}
|
||||
|
||||
describe('registerRollingActiveUsersGauge', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
vi.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('observes one point per window with the window attribute', async () => {
|
||||
// Postgres count(*) comes back as a numeric string — assert we coerce it.
|
||||
const { db } = makeDb([{ dau: '5', wau: '12', mau: '40' }])
|
||||
const { metricReadErrors } = makeReadErrors()
|
||||
const { gauge, observe, run } = makeGauge()
|
||||
|
||||
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
|
||||
await run()
|
||||
|
||||
expect(observe).toHaveBeenCalledTimes(3)
|
||||
expect(observe).toHaveBeenNthCalledWith(1, 5, { window: '24h' })
|
||||
expect(observe).toHaveBeenNthCalledWith(2, 12, { window: '7d' })
|
||||
expect(observe).toHaveBeenNthCalledWith(3, 40, { window: '30d' })
|
||||
})
|
||||
|
||||
it('serves cached values without re-querying inside the 60s TTL', async () => {
|
||||
const { db, select } = makeDb([{ dau: '3', wau: '7', mau: '9' }])
|
||||
const { metricReadErrors } = makeReadErrors()
|
||||
const { gauge, observe, run } = makeGauge()
|
||||
|
||||
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
|
||||
await run()
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
await run()
|
||||
|
||||
// Second collection within TTL hits cache: still one DB query, but six
|
||||
// observes (3 per cycle).
|
||||
expect(select).toHaveBeenCalledTimes(1)
|
||||
expect(observe).toHaveBeenCalledTimes(6)
|
||||
})
|
||||
|
||||
it('re-queries after the TTL expires', async () => {
|
||||
const { db, select } = makeDb([{ dau: '1', wau: '1', mau: '1' }])
|
||||
const { metricReadErrors } = makeReadErrors()
|
||||
const { gauge, run } = makeGauge()
|
||||
|
||||
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
|
||||
await run()
|
||||
await vi.advanceTimersByTimeAsync(61_000)
|
||||
await run()
|
||||
|
||||
expect(select).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('on DB error: increments read-errors and does not observe', async () => {
|
||||
// ROOT CAUSE:
|
||||
//
|
||||
// A silent gauge that observes a stale/zero value on DB failure masks the
|
||||
// outage forever — an absence-based alert can never fire.
|
||||
//
|
||||
// We fixed this by skipping result.observe(...) on error and bumping
|
||||
// airi.observability.read_errors{metric}, so Prometheus staleness exposes
|
||||
// the broken DB instead.
|
||||
const { db } = makeDb([], { reject: true })
|
||||
const { metricReadErrors, add } = makeReadErrors()
|
||||
const { gauge, observe, run } = makeGauge()
|
||||
|
||||
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
|
||||
await run()
|
||||
|
||||
expect(observe).not.toHaveBeenCalled()
|
||||
expect(add).toHaveBeenCalledWith(1, { metric: 'user.active_rolling' })
|
||||
})
|
||||
|
||||
it('coalesces concurrent collection cycles into one DB query', async () => {
|
||||
const { db, select } = makeDb([{ dau: '2', wau: '4', mau: '6' }])
|
||||
const { metricReadErrors } = makeReadErrors()
|
||||
const { gauge, run } = makeGauge()
|
||||
|
||||
registerRollingActiveUsersGauge(gauge, db, metricReadErrors)
|
||||
// Fire two callbacks before the first refresh resolves — the in-flight
|
||||
// lock must fold them into a single query.
|
||||
await Promise.all([run(), run()])
|
||||
|
||||
expect(select).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,132 @@
|
||||
import type { AuthMetrics, ObservabilityMetrics } from '..'
|
||||
import type { Database } from '../../libs/db'
|
||||
|
||||
import { useLogger } from '@guiiai/logg'
|
||||
import { sql } from 'drizzle-orm'
|
||||
|
||||
import { user as userTable } from '../../schemas/accounts'
|
||||
|
||||
/**
|
||||
* Trailing windows reported by the `user.active_rolling` gauge. The `label`
|
||||
* becomes the Prometheus `window` attribute (`user_active_rolling{window="24h"}`)
|
||||
* and `ms` is the lookback applied to `user.last_seen_at`.
|
||||
*/
|
||||
const WINDOWS = [
|
||||
{ label: '24h', ms: 24 * 60 * 60 * 1000 },
|
||||
{ label: '7d', ms: 7 * 24 * 60 * 60 * 1000 },
|
||||
{ label: '30d', ms: 30 * 24 * 60 * 60 * 1000 },
|
||||
] as const
|
||||
|
||||
type WindowLabel = (typeof WINDOWS)[number]['label']
|
||||
type RollingCounts = Record<WindowLabel, number>
|
||||
|
||||
/**
|
||||
* Wire the `user.active_rolling` ObservableGauge to a Postgres
|
||||
* `COUNT(*) FILTER (WHERE last_seen_at > cutoff)` over the `user` table, one
|
||||
* filter per trailing window (DAU / WAU / MAU).
|
||||
*
|
||||
* Use when:
|
||||
* - Assembling DI in `createApp()`, exactly once per process.
|
||||
*
|
||||
* Why this exists alongside `registerDistinctActiveUsersGauge`:
|
||||
* - `user.distinct_active` counts users with a currently non-expired session
|
||||
* ("signed in right now"). It cannot answer "how many users came back this
|
||||
* week" because expired sessions drop out.
|
||||
* - `user.active_rolling` reads `user.last_seen_at` — touched on sign-in and
|
||||
* on every OIDC access-token refresh (~hourly) — so it measures activity
|
||||
* over a trailing window regardless of session state. This is the standard
|
||||
* DAU / WAU / MAU engagement funnel.
|
||||
*
|
||||
* Expects:
|
||||
* - `gauge` is the ObservableGauge handle created in `initOtel`.
|
||||
* - `db` is the migrated Drizzle handle.
|
||||
* - `metricReadErrors` is the shared counter; failures inside the callback
|
||||
* increment it labelled with the originating metric name.
|
||||
*
|
||||
* Multi-replica note:
|
||||
* - Cluster-wide gauge — every replica reads the same DB and reports the same
|
||||
* value per window. Dashboards MUST aggregate with `max()`/`avg()`, NOT
|
||||
* `sum()`. See observability-conventions.md.
|
||||
*
|
||||
* Concurrency:
|
||||
* - Same in-flight promise lock as the sibling gauges, so concurrent OTel
|
||||
* collection cycles fold into one DB query. Cached for 60s — DAU/WAU/MAU
|
||||
* move slowly and the query scans the whole `user` table, so a longer TTL
|
||||
* than the per-row session gauges keeps DB load low.
|
||||
*
|
||||
* Failure mode:
|
||||
* - DB error → increment `airi.observability.read_errors{metric}` and skip
|
||||
* `result.observe(...)`. Prometheus staleness exposes the outage instead of
|
||||
* pinning stale values forever.
|
||||
*/
|
||||
export function registerRollingActiveUsersGauge(
|
||||
gauge: AuthMetrics['rollingActiveUsers'],
|
||||
db: Database,
|
||||
metricReadErrors: ObservabilityMetrics['metricReadErrors'],
|
||||
) {
|
||||
const log = useLogger('rolling-active-users-gauge').useGlobalConfig()
|
||||
const CACHE_TTL_MS = 60_000
|
||||
|
||||
let cachedAt = 0
|
||||
let cached: RollingCounts = { '24h': 0, '7d': 0, '30d': 0 }
|
||||
let refreshInFlight: Promise<boolean> | null = null
|
||||
|
||||
async function refresh(): Promise<boolean> {
|
||||
try {
|
||||
// Compute cutoffs from the app clock so all three windows are anchored
|
||||
// to the same instant within one query. `id` is the PK, so
|
||||
// `count(*)` == `count(distinct id)` — no DISTINCT needed.
|
||||
const now = Date.now()
|
||||
const since24h = new Date(now - WINDOWS[0].ms)
|
||||
const since7d = new Date(now - WINDOWS[1].ms)
|
||||
const since30d = new Date(now - WINDOWS[2].ms)
|
||||
const rows = await db
|
||||
.select({
|
||||
dau: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${since24h})`,
|
||||
wau: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${since7d})`,
|
||||
mau: sql<number>`count(*) filter (where ${userTable.lastSeenAt} > ${since30d})`,
|
||||
})
|
||||
.from(userTable)
|
||||
cached = {
|
||||
'24h': Number(rows[0]?.dau ?? 0),
|
||||
'7d': Number(rows[0]?.wau ?? 0),
|
||||
'30d': Number(rows[0]?.mau ?? 0),
|
||||
}
|
||||
cachedAt = Date.now()
|
||||
return true
|
||||
}
|
||||
catch (err) {
|
||||
log.withError(err).warn('Failed to read rolling active users for gauge')
|
||||
metricReadErrors.add(1, { metric: 'user.active_rolling' })
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
function observeAll(result: Parameters<Parameters<typeof gauge.addCallback>[0]>[0]) {
|
||||
for (const w of WINDOWS)
|
||||
result.observe(cached[w.label], { window: w.label })
|
||||
}
|
||||
|
||||
gauge.addCallback(async (result) => {
|
||||
const now = Date.now()
|
||||
|
||||
// Cache fresh — serve last good values without touching the DB.
|
||||
if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) {
|
||||
observeAll(result)
|
||||
return
|
||||
}
|
||||
|
||||
// Coalesce concurrent refreshes onto one in-flight promise.
|
||||
if (!refreshInFlight) {
|
||||
refreshInFlight = refresh().finally(() => {
|
||||
refreshInFlight = null
|
||||
})
|
||||
}
|
||||
const ok = await refreshInFlight
|
||||
|
||||
if (ok)
|
||||
observeAll(result)
|
||||
// else: deliberately skip observe — let Prometheus staleness expose the
|
||||
// DB outage instead of masking it with stale numbers.
|
||||
})
|
||||
}
|
||||
@@ -52,6 +52,7 @@ import {
|
||||
METRIC_STRIPE_EVENTS,
|
||||
METRIC_STRIPE_PAYMENT_FAILED,
|
||||
METRIC_STRIPE_SUBSCRIPTION_EVENT,
|
||||
METRIC_USER_ACTIVE_ROLLING,
|
||||
METRIC_USER_ACTIVE_SESSIONS,
|
||||
METRIC_USER_DISTINCT_ACTIVE,
|
||||
METRIC_USER_LOGIN,
|
||||
@@ -102,6 +103,26 @@ export interface AuthMetrics {
|
||||
* with `avg()`, not `sum()` — see observability-conventions.md.
|
||||
*/
|
||||
distinctActiveUsers: ObservableGauge
|
||||
/**
|
||||
* Pull-based gauge for rolling-window distinct active users (DAU / WAU /
|
||||
* MAU).
|
||||
*
|
||||
* Use when:
|
||||
* - Reporting "how many users were active in the last 24h / 7d / 30d" —
|
||||
* the standard product-engagement funnel, distinct from
|
||||
* {@link AuthMetrics.distinctActiveUsers} which only counts users with a
|
||||
* currently-live session.
|
||||
*
|
||||
* Expects:
|
||||
* - Backed by `COUNT(*) FILTER (WHERE last_seen_at > now() - window)` over
|
||||
* the `user` table. `last_seen_at` is touched on sign-in and on every
|
||||
* OIDC access-token refresh (~hourly), so it is a per-user last-activity
|
||||
* timestamp (see the `user.lastSeenAt` schema note).
|
||||
* - Observed once per window with a `window` attribute (`24h` / `7d` /
|
||||
* `30d`). Same cluster-wide truth as the other DB-backed gauges;
|
||||
* dashboards MUST aggregate with `max()`/`avg()`, not `sum()`.
|
||||
*/
|
||||
rollingActiveUsers: ObservableGauge
|
||||
}
|
||||
|
||||
export interface EngagementMetrics {
|
||||
@@ -320,6 +341,9 @@ export function initOtel(env: Env): OtelInstance | null {
|
||||
distinctActiveUsers: meter.createObservableGauge(METRIC_USER_DISTINCT_ACTIVE, {
|
||||
description: 'Distinct users with ≥1 non-expired session — true active-user count, immune to per-row session inflation (cluster-wide; dashboard must use avg(), not sum())',
|
||||
}),
|
||||
rollingActiveUsers: meter.createObservableGauge(METRIC_USER_ACTIVE_ROLLING, {
|
||||
description: 'Rolling-window distinct active users (DAU/WAU/MAU) from user.last_seen_at, labelled by window=24h|7d|30d (cluster-wide; dashboard must use max(), not sum())',
|
||||
}),
|
||||
}
|
||||
|
||||
// Engagement metrics
|
||||
|
||||
@@ -53,6 +53,12 @@ export const METRIC_USER_ACTIVE_SESSIONS = 'user.active_sessions'
|
||||
// creates a new row per sign-in / per OIDC token refresh and never GCs)
|
||||
// vs real user growth.
|
||||
export const METRIC_USER_DISTINCT_ACTIVE = 'user.distinct_active'
|
||||
// Rolling-window distinct active users (DAU / WAU / MAU), sourced from
|
||||
// `user.last_seen_at` (touched on sign-in and every OIDC token refresh).
|
||||
// Single gauge, observed once per window with a `window="24h"|"7d"|"30d"`
|
||||
// attribute. Unlike USER_DISTINCT_ACTIVE (live-session count) this measures
|
||||
// activity over a trailing time window, not "currently signed in".
|
||||
export const METRIC_USER_ACTIVE_ROLLING = 'user.active_rolling'
|
||||
|
||||
// Engagement (AIRI custom)
|
||||
export const METRIC_CHAT_MESSAGES = 'chat.messages'
|
||||
|
||||
Reference in New Issue
Block a user