324 lines
14 KiB
TypeScript
324 lines
14 KiB
TypeScript
import { Injectable, OnModuleDestroy } from '@nestjs/common';
|
|
import { monitorEventLoopDelay } from 'node:perf_hooks';
|
|
|
|
const HTTP_DURATION_BUCKETS = [0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const;
|
|
const CMPP_INBOUND_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const;
|
|
const SEND_WORKER_DURATION_BUCKETS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 3, 10] as const;
|
|
|
|
export type CmppInboundStage =
|
|
| 'application_lookup'
|
|
| 'inbox_persist'
|
|
| 'worker_claim'
|
|
| 'reference_preload'
|
|
| 'daily_quota'
|
|
| 'long_message_fragment'
|
|
| 'submission_precheck'
|
|
| 'template_match'
|
|
| 'task_persist'
|
|
| 'api_request_persist'
|
|
| 'content_detection'
|
|
| 'message_persist'
|
|
| 'risk_frequency'
|
|
| 'billing'
|
|
| 'queue_publish'
|
|
| 'complete_submit'
|
|
| 'total';
|
|
|
|
export type CmppInboundStageResult = 'success' | 'error';
|
|
|
|
export type SendWorkerStage =
|
|
| 'message_load'
|
|
| 'phone_routing'
|
|
| 'route_lookup'
|
|
| 'signature_candidates'
|
|
| 'signature_final_check'
|
|
| 'rate_limit'
|
|
| 'submit_transaction'
|
|
| 'gateway_bullmq_publish'
|
|
| 'gateway_stream_publish'
|
|
| 'task_progress'
|
|
| 'total';
|
|
|
|
export type SendWorkerStageResult = 'success' | 'error' | 'skipped';
|
|
export type SendWorkerQueueState = 'waiting' | 'active' | 'completed' | 'failed' | 'delayed' | 'prioritized';
|
|
|
|
type HttpMetric = {
|
|
count: number;
|
|
durationSum: number;
|
|
buckets: number[];
|
|
};
|
|
|
|
function escapeLabel(value: string) {
|
|
return value.replace(/\\/g, '\\\\').replace(/\n/g, '\\n').replace(/"/g, '\\"');
|
|
}
|
|
|
|
function metricLine(name: string, value: number, labels?: Record<string, string>) {
|
|
const suffix = labels
|
|
? `{${Object.entries(labels)
|
|
.map(([key, item]) => `${key}="${escapeLabel(item)}"`)
|
|
.join(',')}}`
|
|
: '';
|
|
return `${name}${suffix} ${Number.isFinite(value) ? value : 0}`;
|
|
}
|
|
|
|
@Injectable()
|
|
export class MetricsService implements OnModuleDestroy {
|
|
private readonly startedAt = process.hrtime.bigint();
|
|
private readonly eventLoopDelay = monitorEventLoopDelay({ resolution: 20 });
|
|
private readonly http = new Map<string, HttpMetric>();
|
|
private readonly cmppInbound = new Map<string, HttpMetric>();
|
|
private readonly sendWorkerStages = new Map<string, HttpMetric>();
|
|
private readonly sendWorkerQueueJobs = new Map<SendWorkerQueueState, number>();
|
|
private readonly sendWorkerResults = new Map<string, number>();
|
|
private sendWorkerConfiguredSlots = 0;
|
|
private sendWorkerInFlightSlots = 0;
|
|
private readonly sendWorkerDatabasePool = new Map<'max' | 'total' | 'idle' | 'waiting', number>();
|
|
private inFlight = 0;
|
|
private inboundWorkflowPending = 0;
|
|
private inboundWorkflowProcessing = 0;
|
|
private inboundWorkflowOldestPendingAgeSeconds = 0;
|
|
private inboundWorkflowConfiguredSlots = 0;
|
|
private inboundWorkflowInFlightSlots = 0;
|
|
private readonly inboundWorkflowResults = new Map<string, number>();
|
|
private readonly authProtectionResults = new Map<string, number>();
|
|
|
|
constructor() {
|
|
this.eventLoopDelay.enable();
|
|
}
|
|
|
|
beginRequest() {
|
|
this.inFlight += 1;
|
|
return process.hrtime.bigint();
|
|
}
|
|
|
|
finishRequest(startedAt: bigint, method: string, route: string, statusCode: number) {
|
|
this.inFlight = Math.max(0, this.inFlight - 1);
|
|
// Only route templates enter labels. Raw URLs, IDs, phone numbers and query strings would create unbounded time series.
|
|
const normalizedRoute = route.startsWith('/') ? route : `/${route}`;
|
|
const labels = [method.toUpperCase(), normalizedRoute, String(statusCode)];
|
|
const key = labels.join('\u0000');
|
|
const metric = this.http.get(key) ?? { count: 0, durationSum: 0, buckets: HTTP_DURATION_BUCKETS.map(() => 0) };
|
|
const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
|
|
metric.count += 1;
|
|
metric.durationSum += durationSeconds;
|
|
HTTP_DURATION_BUCKETS.forEach((bucket, index) => {
|
|
if (durationSeconds <= bucket) metric.buckets[index] += 1;
|
|
});
|
|
this.http.set(key, metric);
|
|
}
|
|
|
|
beginCmppInboundStage() {
|
|
return process.hrtime.bigint();
|
|
}
|
|
|
|
finishCmppInboundStage(startedAt: bigint, stage: CmppInboundStage, result: CmppInboundStageResult) {
|
|
const key = `${stage}\u0000${result}`;
|
|
const metric = this.cmppInbound.get(key) ?? {
|
|
count: 0,
|
|
durationSum: 0,
|
|
buckets: CMPP_INBOUND_DURATION_BUCKETS.map(() => 0),
|
|
};
|
|
const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
|
|
metric.count += 1;
|
|
metric.durationSum += durationSeconds;
|
|
CMPP_INBOUND_DURATION_BUCKETS.forEach((bucket, index) => {
|
|
if (durationSeconds <= bucket) metric.buckets[index] += 1;
|
|
});
|
|
this.cmppInbound.set(key, metric);
|
|
}
|
|
|
|
beginSendWorkerStage() {
|
|
return process.hrtime.bigint();
|
|
}
|
|
|
|
finishSendWorkerStage(startedAt: bigint, stage: SendWorkerStage, result: SendWorkerStageResult) {
|
|
const key = `${stage}\u0000${result}`;
|
|
const metric = this.sendWorkerStages.get(key) ?? {
|
|
count: 0,
|
|
durationSum: 0,
|
|
buckets: SEND_WORKER_DURATION_BUCKETS.map(() => 0),
|
|
};
|
|
const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
|
|
metric.count += 1;
|
|
metric.durationSum += durationSeconds;
|
|
SEND_WORKER_DURATION_BUCKETS.forEach((bucket, index) => {
|
|
if (durationSeconds <= bucket) metric.buckets[index] += 1;
|
|
});
|
|
this.sendWorkerStages.set(key, metric);
|
|
}
|
|
|
|
setSendWorkerSlots(configured: number, inFlight: number) {
|
|
this.sendWorkerConfiguredSlots = Math.max(0, configured);
|
|
this.sendWorkerInFlightSlots = Math.max(0, inFlight);
|
|
}
|
|
|
|
setSendWorkerQueueJobs(state: SendWorkerQueueState, count: number) {
|
|
this.sendWorkerQueueJobs.set(state, Math.max(0, count));
|
|
}
|
|
|
|
recordSendWorkerResult(result: 'completed' | 'failed' | 'skipped') {
|
|
this.sendWorkerResults.set(result, (this.sendWorkerResults.get(result) ?? 0) + 1);
|
|
}
|
|
|
|
setSendWorkerDatabasePool(state: 'max' | 'total' | 'idle' | 'waiting', count: number) {
|
|
this.sendWorkerDatabasePool.set(state, Math.max(0, count));
|
|
}
|
|
|
|
setInboundWorkflowState(pending: number, processing: number, oldestPendingAgeSeconds: number) {
|
|
this.inboundWorkflowPending = Math.max(0, pending);
|
|
this.inboundWorkflowProcessing = Math.max(0, processing);
|
|
this.inboundWorkflowOldestPendingAgeSeconds = Math.max(0, oldestPendingAgeSeconds);
|
|
}
|
|
|
|
setInboundWorkflowSlots(configured: number, inFlight: number) {
|
|
this.inboundWorkflowConfiguredSlots = Math.max(0, configured);
|
|
this.inboundWorkflowInFlightSlots = Math.max(0, inFlight);
|
|
}
|
|
|
|
recordInboundWorkflowResult(result: 'completed' | 'retry') {
|
|
this.inboundWorkflowResults.set(result, (this.inboundWorkflowResults.get(result) ?? 0) + 1);
|
|
}
|
|
|
|
recordAuthProtectionResult(
|
|
event: 'captcha_allowed' | 'captcha_rejected' | 'login_locked',
|
|
scope: 'none' | 'account' | 'ip' | 'pair' = 'none',
|
|
) {
|
|
const key = `${event}\u0000${scope}`;
|
|
this.authProtectionResults.set(key, (this.authProtectionResults.get(key) ?? 0) + 1);
|
|
}
|
|
|
|
render() {
|
|
const memory = process.memoryUsage();
|
|
const uptime = Number(process.hrtime.bigint() - this.startedAt) / 1_000_000_000;
|
|
const lines = [
|
|
'# HELP cmpp_api_process_uptime_seconds API process uptime.',
|
|
'# TYPE cmpp_api_process_uptime_seconds gauge',
|
|
metricLine('cmpp_api_process_uptime_seconds', uptime),
|
|
'# HELP cmpp_api_process_resident_memory_bytes API resident memory.',
|
|
'# TYPE cmpp_api_process_resident_memory_bytes gauge',
|
|
metricLine('cmpp_api_process_resident_memory_bytes', memory.rss),
|
|
'# HELP cmpp_api_nodejs_heap_used_bytes Node.js heap currently used.',
|
|
'# TYPE cmpp_api_nodejs_heap_used_bytes gauge',
|
|
metricLine('cmpp_api_nodejs_heap_used_bytes', memory.heapUsed),
|
|
'# HELP cmpp_api_nodejs_heap_total_bytes Node.js allocated heap.',
|
|
'# TYPE cmpp_api_nodejs_heap_total_bytes gauge',
|
|
metricLine('cmpp_api_nodejs_heap_total_bytes', memory.heapTotal),
|
|
'# HELP cmpp_api_nodejs_event_loop_lag_p99_seconds Event loop delay p99 since the previous scrape.',
|
|
'# TYPE cmpp_api_nodejs_event_loop_lag_p99_seconds gauge',
|
|
metricLine(
|
|
'cmpp_api_nodejs_event_loop_lag_p99_seconds',
|
|
this.eventLoopDelay.count ? this.eventLoopDelay.percentile(99) / 1_000_000_000 : 0,
|
|
),
|
|
'# HELP cmpp_api_http_requests_in_flight Current API requests in flight.',
|
|
'# TYPE cmpp_api_http_requests_in_flight gauge',
|
|
metricLine('cmpp_api_http_requests_in_flight', this.inFlight),
|
|
'# HELP cmpp_api_http_requests_total API requests grouped by bounded route templates.',
|
|
'# TYPE cmpp_api_http_requests_total counter',
|
|
'# HELP cmpp_api_http_request_duration_seconds API request duration.',
|
|
'# TYPE cmpp_api_http_request_duration_seconds histogram',
|
|
'# HELP cmpp_api_cmpp_inbound_stage_duration_seconds CMPP inbound processing duration by bounded stage and result.',
|
|
'# TYPE cmpp_api_cmpp_inbound_stage_duration_seconds histogram',
|
|
'# HELP cmpp_worker_send_stage_duration_seconds Send worker processing duration by bounded stage and result.',
|
|
'# TYPE cmpp_worker_send_stage_duration_seconds histogram',
|
|
'# HELP cmpp_worker_send_queue_jobs BullMQ send jobs by queue state.',
|
|
'# TYPE cmpp_worker_send_queue_jobs gauge',
|
|
'# HELP cmpp_worker_send_slots Send worker concurrency slots by state.',
|
|
'# TYPE cmpp_worker_send_slots gauge',
|
|
metricLine('cmpp_worker_send_slots', this.sendWorkerConfiguredSlots, { state: 'configured' }),
|
|
metricLine('cmpp_worker_send_slots', this.sendWorkerInFlightSlots, { state: 'in_flight' }),
|
|
'# HELP cmpp_worker_send_jobs_total Send worker processing outcomes.',
|
|
'# TYPE cmpp_worker_send_jobs_total counter',
|
|
'# HELP cmpp_worker_database_pool_connections Worker PostgreSQL client pool slots by state.',
|
|
'# TYPE cmpp_worker_database_pool_connections gauge',
|
|
'# HELP cmpp_worker_inbound_workflow_items Current durable CMPP inbound workflow items by state.',
|
|
'# TYPE cmpp_worker_inbound_workflow_items gauge',
|
|
metricLine('cmpp_worker_inbound_workflow_items', this.inboundWorkflowPending, { state: 'pending' }),
|
|
metricLine('cmpp_worker_inbound_workflow_items', this.inboundWorkflowProcessing, { state: 'processing' }),
|
|
'# HELP cmpp_worker_inbound_workflow_slots Durable workflow worker slots by state.',
|
|
'# TYPE cmpp_worker_inbound_workflow_slots gauge',
|
|
metricLine('cmpp_worker_inbound_workflow_slots', this.inboundWorkflowConfiguredSlots, { state: 'configured' }),
|
|
metricLine('cmpp_worker_inbound_workflow_slots', this.inboundWorkflowInFlightSlots, { state: 'in_flight' }),
|
|
'# HELP cmpp_worker_inbound_workflow_oldest_pending_age_seconds Age of the oldest pending durable workflow.',
|
|
'# TYPE cmpp_worker_inbound_workflow_oldest_pending_age_seconds gauge',
|
|
metricLine(
|
|
'cmpp_worker_inbound_workflow_oldest_pending_age_seconds',
|
|
this.inboundWorkflowOldestPendingAgeSeconds,
|
|
),
|
|
'# HELP cmpp_worker_inbound_workflow_results_total Durable workflow processing outcomes.',
|
|
'# TYPE cmpp_worker_inbound_workflow_results_total counter',
|
|
'# HELP cmpp_api_auth_protection_events_total Authentication protection outcomes by bounded event and scope.',
|
|
'# TYPE cmpp_api_auth_protection_events_total counter',
|
|
];
|
|
for (const [key, metric] of this.http) {
|
|
const [method, route, status] = key.split('\u0000');
|
|
const labels = { method, route, status };
|
|
lines.push(metricLine('cmpp_api_http_requests_total', metric.count, labels));
|
|
HTTP_DURATION_BUCKETS.forEach((bucket, index) => {
|
|
lines.push(
|
|
metricLine('cmpp_api_http_request_duration_seconds_bucket', metric.buckets[index], {
|
|
...labels,
|
|
le: String(bucket),
|
|
}),
|
|
);
|
|
});
|
|
lines.push(metricLine('cmpp_api_http_request_duration_seconds_bucket', metric.count, { ...labels, le: '+Inf' }));
|
|
lines.push(metricLine('cmpp_api_http_request_duration_seconds_sum', metric.durationSum, labels));
|
|
lines.push(metricLine('cmpp_api_http_request_duration_seconds_count', metric.count, labels));
|
|
}
|
|
for (const [key, metric] of this.cmppInbound) {
|
|
const [stage, result] = key.split('\u0000');
|
|
const labels = { stage, result };
|
|
CMPP_INBOUND_DURATION_BUCKETS.forEach((bucket, index) => {
|
|
lines.push(
|
|
metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_bucket', metric.buckets[index], {
|
|
...labels,
|
|
le: String(bucket),
|
|
}),
|
|
);
|
|
});
|
|
lines.push(
|
|
metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_bucket', metric.count, { ...labels, le: '+Inf' }),
|
|
);
|
|
lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_sum', metric.durationSum, labels));
|
|
lines.push(metricLine('cmpp_api_cmpp_inbound_stage_duration_seconds_count', metric.count, labels));
|
|
}
|
|
for (const [key, metric] of this.sendWorkerStages) {
|
|
const [stage, result] = key.split('\u0000');
|
|
const labels = { stage, result };
|
|
SEND_WORKER_DURATION_BUCKETS.forEach((bucket, index) => {
|
|
lines.push(
|
|
metricLine('cmpp_worker_send_stage_duration_seconds_bucket', metric.buckets[index], {
|
|
...labels,
|
|
le: String(bucket),
|
|
}),
|
|
);
|
|
});
|
|
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_bucket', metric.count, { ...labels, le: '+Inf' }));
|
|
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_sum', metric.durationSum, labels));
|
|
lines.push(metricLine('cmpp_worker_send_stage_duration_seconds_count', metric.count, labels));
|
|
}
|
|
for (const [state, count] of this.sendWorkerQueueJobs) {
|
|
lines.push(metricLine('cmpp_worker_send_queue_jobs', count, { state }));
|
|
}
|
|
for (const [result, count] of this.sendWorkerResults) {
|
|
lines.push(metricLine('cmpp_worker_send_jobs_total', count, { result }));
|
|
}
|
|
for (const [state, count] of this.sendWorkerDatabasePool) {
|
|
lines.push(metricLine('cmpp_worker_database_pool_connections', count, { state }));
|
|
}
|
|
for (const [result, count] of this.inboundWorkflowResults) {
|
|
lines.push(metricLine('cmpp_worker_inbound_workflow_results_total', count, { result }));
|
|
}
|
|
for (const [key, count] of this.authProtectionResults) {
|
|
const [event, scope] = key.split('\u0000');
|
|
lines.push(metricLine('cmpp_api_auth_protection_events_total', count, { event, scope }));
|
|
}
|
|
this.eventLoopDelay.reset();
|
|
return `${lines.join('\n')}\n`;
|
|
}
|
|
|
|
onModuleDestroy() {
|
|
this.eventLoopDelay.disable();
|
|
}
|
|
}
|