diff --git a/.DS_Store b/.DS_Store index 7bd6c82..4bdc589 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/README.md b/README.md index 9a6e44f..a44847c 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,12 @@ Every HTTP API route is mounted under `/v1`, including health and admin routes. For example: `GET /v1/health`, `GET /v1/ngos`, `GET /v1/streams`, and `GET /v1/indexer/status`. +## Prometheus metrics + +`GET /metrics` returns Prometheus text format (`text/plain; version=0.0.4`). +It exposes HTTP request totals and duration, indexer checkpoint and lag, and webhook +delivery totals. Configure Prometheus to scrape it on the API host and port. + ## NGO Verification Model An NGO's verified status consists of two distinct steps kept deliberately separate: diff --git a/src/.DS_Store b/src/.DS_Store new file mode 100644 index 0000000..32a5fd6 Binary files /dev/null and b/src/.DS_Store differ diff --git a/src/indexer/worker.ts b/src/indexer/worker.ts index 5f53fc8..b5a2981 100644 --- a/src/indexer/worker.ts +++ b/src/indexer/worker.ts @@ -1,3 +1,4 @@ +import { recordIndexerPosition } from '../metrics.js'; import { getLatestLedgerSequence, rpcServer } from '../stellar/rpc.js'; import { getCheckpointState, saveCheckpoint } from './checkpoint.js'; import { WATCHED_CONTRACT_IDS } from './contracts.js'; @@ -94,6 +95,7 @@ export async function pollOnce(handleEvent: EventHandler): Promise { // otherwise start from "now" to avoid replaying all history. lastProcessedLedger = START_LEDGER ?? (await getLatestLedgerSequence()); await saveCheckpoint(lastProcessedLedger); + recordIndexerPosition(lastProcessedLedger, lastProcessedLedger); return; } } @@ -108,6 +110,7 @@ export async function pollOnce(handleEvent: EventHandler): Promise { // Skip ahead to the current ledger; the alternative is an indexer that // never recovers. Events in the gap are lost, so say so loudly. const latestLedger = await getLatestLedgerSequence(); + recordIndexerPosition(lastProcessedLedger, latestLedger); if (lastProcessedLedger >= latestLedger && lastProcessedEventId === undefined) { return; diff --git a/src/metrics.ts b/src/metrics.ts new file mode 100644 index 0000000..1b38eb8 --- /dev/null +++ b/src/metrics.ts @@ -0,0 +1,80 @@ +type HttpKey = `${string}\u0000${string}\u0000${number}`; + +const httpRequests = new Map(); +const httpDuration = new Map(); +const notificationDeliveries = new Map<'success' | 'failure', number>(); + +let indexerCheckpointLedger = 0; +let indexerLatestLedger = 0; + +function labels(method: string, route: string, statusCode: number): string { + return `method="${method}",route="${route.replaceAll('"', '\\"')}",status_code="${statusCode}"`; +} + +export function recordHttpRequest( + method: string, + route: string, + statusCode: number, + durationSeconds: number, +): void { + const key: HttpKey = `${method}\u0000${route}\u0000${statusCode}`; + httpRequests.set(key, (httpRequests.get(key) ?? 0) + 1); + const current = httpDuration.get(key) ?? { count: 0, sum: 0 }; + httpDuration.set(key, { count: current.count + 1, sum: current.sum + durationSeconds }); +} + +export function recordIndexerPosition(checkpointLedger: number, latestLedger: number): void { + indexerCheckpointLedger = checkpointLedger; + indexerLatestLedger = latestLedger; +} + +export function recordNotificationDelivery(outcome: 'success' | 'failure'): void { + notificationDeliveries.set(outcome, (notificationDeliveries.get(outcome) ?? 0) + 1); +} + +export function renderMetrics(): string { + const lines = [ + '# HELP streamgive_http_requests_total Total HTTP requests handled.', + '# TYPE streamgive_http_requests_total counter', + ]; + + for (const [key, count] of httpRequests) { + const [method, route, statusCode] = key.split('\u0000'); + lines.push( + `streamgive_http_requests_total{${labels(method, route, Number(statusCode))}} ${count}`, + ); + } + + lines.push( + '# HELP streamgive_http_request_duration_seconds HTTP request duration in seconds.', + '# TYPE streamgive_http_request_duration_seconds summary', + ); + for (const [key, value] of httpDuration) { + const [method, route, statusCode] = key.split('\u0000'); + const metricLabels = labels(method, route, Number(statusCode)); + lines.push(`streamgive_http_request_duration_seconds_sum{${metricLabels}} ${value.sum}`); + lines.push(`streamgive_http_request_duration_seconds_count{${metricLabels}} ${value.count}`); + } + + lines.push( + '# HELP streamgive_indexer_checkpoint_ledger Last ledger checkpointed by the indexer.', + '# TYPE streamgive_indexer_checkpoint_ledger gauge', + `streamgive_indexer_checkpoint_ledger ${indexerCheckpointLedger}`, + '# HELP streamgive_indexer_latest_ledger Latest ledger observed from Stellar RPC.', + '# TYPE streamgive_indexer_latest_ledger gauge', + `streamgive_indexer_latest_ledger ${indexerLatestLedger}`, + '# HELP streamgive_indexer_lag_ledgers Difference between the latest and checkpoint ledgers.', + '# TYPE streamgive_indexer_lag_ledgers gauge', + `streamgive_indexer_lag_ledgers ${Math.max(0, indexerLatestLedger - indexerCheckpointLedger)}`, + '# HELP streamgive_notification_deliveries_total Webhook notification delivery attempts.', + '# TYPE streamgive_notification_deliveries_total counter', + ); + + for (const outcome of ['success', 'failure'] as const) { + lines.push( + `streamgive_notification_deliveries_total{outcome="${outcome}"} ${notificationDeliveries.get(outcome) ?? 0}`, + ); + } + + return `${lines.join('\n')}\n`; +} diff --git a/src/notifications/service.ts b/src/notifications/service.ts index 087d989..5cb6434 100644 --- a/src/notifications/service.ts +++ b/src/notifications/service.ts @@ -1,6 +1,7 @@ import { createHmac } from 'node:crypto'; import { prisma } from '../db.js'; +import { recordNotificationDelivery } from '../metrics.js'; import type { NotificationEvent } from './types.js'; const deliveredEventIds = new Set(); @@ -64,15 +65,14 @@ async function notifyWebhook(event: NotificationEvent): Promise { success = true; } else { error = `status ${res.status}`; - console.error( - `webhook notification failed with status ${res.status} for ${webhookUrl}`, - ); + console.error(`webhook notification failed with status ${res.status} for ${webhookUrl}`); } } catch (err) { error = err instanceof Error ? err.message : String(err); console.error('webhook notification failed', err); } + recordNotificationDelivery(success ? 'success' : 'failure'); await prisma.notificationLog.create({ data: { eventType: event.type, channel: 'webhook', success, error }, }); diff --git a/src/server.ts b/src/server.ts index 485d715..f323cc1 100644 --- a/src/server.ts +++ b/src/server.ts @@ -7,6 +7,8 @@ import Fastify from 'fastify'; import type { FastifyError } from 'fastify'; import { prisma } from './db.js'; +import { recordHttpRequest, renderMetrics } from './metrics.js'; +import { getCheckpoint } from './indexer/checkpoint.js'; import { donorRoutes } from './routes/donors.js'; import { impactRoutes } from './routes/impact.js'; import { indexerStatusRoutes } from './routes/indexerStatus.js'; @@ -72,6 +74,7 @@ export interface BuildServerOptions { } export function buildServer(options?: BuildServerOptions) { + const requestStartedAt = new WeakMap(); const trustProxySetting = options?.trustProxy !== undefined ? options.trustProxy @@ -165,6 +168,26 @@ export function buildServer(options?: BuildServerOptions) { }, }); + app.addHook('onRequest', async (request) => { + requestStartedAt.set(request, process.hrtime.bigint()); + }); + + app.addHook('onResponse', async (request, reply) => { + const startedAt = requestStartedAt.get(request); + if (startedAt === undefined) return; + const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; + recordHttpRequest( + request.method, + request.routeOptions.url ?? 'unmatched', + reply.statusCode, + durationSeconds, + ); + }); + + app.get('/metrics', async (_request, reply) => { + return reply.type('text/plain; version=0.0.4; charset=utf-8').send(renderMetrics()); + }); + app.register(swagger, { openapi: { info: { diff --git a/test/.DS_Store b/test/.DS_Store new file mode 100644 index 0000000..a371591 Binary files /dev/null and b/test/.DS_Store differ diff --git a/test/metrics.test.ts b/test/metrics.test.ts new file mode 100644 index 0000000..c6be4c3 --- /dev/null +++ b/test/metrics.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest'; + +vi.mock('../src/routes/ngoApplications.js', () => ({ + ngoApplicationRoutes: async () => {}, +})); + +import { buildServer } from '../src/server.js'; + +describe('GET /metrics', () => { + it('returns Prometheus text with HTTP and indexer metrics', async () => { + const app = buildServer(); + + await app.inject({ method: 'GET', url: '/not-a-route' }); + const response = await app.inject({ method: 'GET', url: '/metrics' }); + + expect(response.statusCode).toBe(200); + expect(response.headers['content-type']).toContain('text/plain'); + expect(response.body).toContain('# TYPE streamgive_http_requests_total counter'); + expect(response.body).toContain( + 'streamgive_http_requests_total{method="GET",route="unmatched",status_code="404"}', + ); + expect(response.body).toContain('# TYPE streamgive_http_request_duration_seconds summary'); + expect(response.body).toContain('# TYPE streamgive_indexer_lag_ledgers gauge'); + expect(response.body).toContain('streamgive_indexer_checkpoint_ledger 0'); + expect(response.body).toContain('# TYPE streamgive_notification_deliveries_total counter'); + + await app.close(); + }); +});