Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions docs/http-metrics.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,11 @@ These three signals — request rate, error rate and p95 latency — are what an
specifically and cannot answer that question; it is unchanged and still means
what it always meant.

The same cardinality rule applies to every metric family: labels must come from
fixed buckets or server-defined values. Never put raw request paths, asset
codes, issuer addresses, query values or websocket subscription strings in a
Prometheus label.

## What is exported

### `http_requests_total`
Expand Down Expand Up @@ -52,6 +57,11 @@ So:
`methods x routes x 5` rather than `methods x routes x ~60`. If a concrete
need for an exact code appears, add it then — it is easy to add a label and
painful to remove one dashboards depend on.
- **Websocket pairs use fixed buckets.** `ws_clients_total`,
`ws_messages_sent_total` and `ws_messages_dropped_total` retain their `pair`
label, but pair-specific subscriptions are recorded as `other`; `*` remains
the aggregate across all subscriptions. Arbitrary pair names and issuer
addresses stay in fanout state only and cannot create Prometheus series.

## Bucket choice

Expand Down
44 changes: 40 additions & 4 deletions src/__tests__/fanout.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'

const { metricLabelValues } = vi.hoisted(() => ({
metricLabelValues: [] as Array<{ metric: string; value: string }>,
}))

vi.mock('ioredis', () => {
const mockRedis = {
publish: vi.fn().mockResolvedValue(1),
Expand Down Expand Up @@ -32,12 +36,15 @@ vi.mock('../events', () => ({
}))

vi.mock('prom-client', () => {
function MockMetric() {
const labels = vi.fn(() => new MockMetric())
function MockMetric(options: { name: string }) {
const metricName = options.name
this.inc = vi.fn()
this.dec = vi.fn()
this.set = vi.fn()
this.labels = labels
this.labels = vi.fn((value: string) => {
metricLabelValues.push({ metric: metricName, value })
return this
})
this.observe = vi.fn()
this.startTimer = vi.fn(() => vi.fn())
}
Expand All @@ -54,6 +61,7 @@ import { fanOutManager } from '../ws/fanout'
describe('FanOutManager', () => {
beforeEach(() => {
vi.clearAllMocks()
metricLabelValues.length = 0
})

afterEach(async () => {
Expand Down Expand Up @@ -96,6 +104,34 @@ describe('FanOutManager', () => {
expect(unsub2).toBeInstanceOf(Function)
})

it('uses bounded metric labels for arbitrary pair subscriptions and keeps the aggregate', () => {
const unsubscribers: Array<() => void> = []
for (let i = 0; i < 100; i++) {
unsubscribers.push(fanOutManager.register(`TOKEN${i}/ISSUER${i}`, {
id: `unknown-pair-${i}`,
send: vi.fn(),
close: vi.fn(),
}))
}

const clientLabels = metricLabelValues
.filter(({ metric }) => metric === 'ws_clients_total')
.map(({ value }) => value)
expect(new Set(clientLabels)).toEqual(new Set(['other', '*']))
expect(clientLabels.filter(value => value === 'other')).toHaveLength(100)
expect(clientLabels.filter(value => value === '*')).toHaveLength(100)

unsubscribers.forEach(unsubscribe => unsubscribe())
})

it('counts a wildcard subscriber once in the aggregate series', () => {
fanOutManager.register('*', { id: 'wildcard-client', send: vi.fn(), close: vi.fn() })

expect(metricLabelValues.filter(({ metric, value }) =>
metric === 'ws_clients_total' && value === '*'
)).toHaveLength(1)
})

it('destroy cleans up all state', async () => {
const mockSocket = {
id: 'client-cleanup',
Expand All @@ -119,4 +155,4 @@ describe('FanOutClient per-pair subscription', () => {

fanOutManager.register('ETH/USDC', mockSocket)
})
})
})
20 changes: 16 additions & 4 deletions src/ws/fanout.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,15 @@ export const ws_messages_dropped_total = new Counter({
const REDIS_CHANNEL = `lens:${activeNetwork}:price:updates`
const DEFAULT_COALESCE_INTERVAL_MS = 100 // max 10 updates/sec per client

/**
* Pair names can contain arbitrary asset codes and issuer identifiers supplied
* by websocket clients. Keep them out of Prometheus labels: `other` is the
* bounded bucket for pair-specific subscriptions and `*` remains the aggregate.
*/
function pairMetricLabel(pairKey: string): 'other' | null {
return pairKey === '*' ? null : 'other'
}

// ─── Client wrapper with backpressure & coalescing ────────────────────────────
export interface FanOutClient {
id: string
Expand Down Expand Up @@ -111,7 +120,8 @@ class FanOutManager {
}
clients.add(id)

ws_clients_total.labels(pairKey).inc()
const metricPair = pairMetricLabel(pairKey)
if (metricPair) ws_clients_total.labels(metricPair).inc()
ws_clients_total.labels('*').inc()

// Return unsubscribe function
Expand All @@ -133,7 +143,8 @@ class FanOutManager {
}
}

ws_clients_total.labels(client.pairKey).dec()
const metricPair = pairMetricLabel(client.pairKey)
if (metricPair) ws_clients_total.labels(metricPair).dec()
ws_clients_total.labels('*').dec()
}

Expand Down Expand Up @@ -214,7 +225,8 @@ class FanOutManager {
private sendToClient(client: InternalClient, payload: string): void {
try {
client.socket.send(payload)
ws_messages_sent_total.labels(client.pairKey).inc()
const metricPair = pairMetricLabel(client.pairKey)
if (metricPair) ws_messages_sent_total.labels(metricPair).inc()
ws_messages_sent_total.labels('*').inc()
} catch {
// Client likely disconnected
Expand All @@ -241,4 +253,4 @@ class FanOutManager {
}

export const fanOutManager = new FanOutManager()
export default FanOutManager
export default FanOutManager
Loading