diff --git a/README.md b/README.md index 4481bdd9..a336255b 100644 --- a/README.md +++ b/README.md @@ -343,6 +343,7 @@ curl http://localhost:3000/status { "ok": true, "lastIndexedLedger": 5842100, + "last_indexed_ledger": 5842100, "latestLedger": 5842102, "lagLedgers": 2, "startedAt": "2025-10-01T10:00:00.000Z", @@ -351,6 +352,43 @@ curl http://localhost:3000/status } ``` +`last_indexed_ledger` is a snake_case alias of `lastIndexedLedger` — the same +name the Prometheus gauge is exported under. Both always carry the same value. + +*** + +### `GET /metrics` + +Indexer and process metrics in Prometheus text exposition format, for scraping, +dashboards, and alerting. + +```bash +curl http://localhost:3000/metrics +``` + +``` +# HELP ledgers_indexed_total Ledgers advanced through by the indexer, per network +# TYPE ledgers_indexed_total counter +ledgers_indexed_total{network="mainnet"} 48213 +# HELP last_indexed_ledger Highest ledger sequence committed by the indexer, per network +# TYPE last_indexed_ledger gauge +last_indexed_ledger{network="mainnet"} 5842100 +``` + +| Metric | Type | Labels | What it says | +| ------ | ---- | ------ | ------------ | +| `ledgers_indexed_total` | counter | `network` | Ledgers the indexer has advanced through. A flat `rate()` on a network whose loop should be running means it has stalled. | +| `transfers_stored_total` | counter | `network`, `type` | Rows persisted, split `fungible` / `nft` — one parse path can break while the other keeps working. | +| `rpc_errors_total` | counter | `outcome` | Failed RPC attempts. `retry` counts attempts `withRetry` absorbed, `exhausted` counts calls that gave up — a degrading endpoint shows up in `retry` long before it fails a call. | +| `last_indexed_ledger` | gauge | `network` | Highest committed ledger. Against the chain tip, this is lag. | +| `db_query_duration_seconds` | histogram | `operation` | Duration of instrumented DB operations, failures included. | + +Standard `process_*` and `nodejs_*` metrics are exported alongside these. + +The endpoint reads in-process counters only — no DB, no RPC — so it keeps +answering while the subsystems it reports on are down, and it is exempt from the +API rate limit so scrapes do not go dark under load. + *** ### `GET /transfers/incoming/:address` @@ -531,6 +569,8 @@ curl -H "Accept: application/vnd.api+json" http://localhost:3000/summary/GABC123 | `GET /healthz` | `health` | | `GET /readyz` | `readiness` | +`GET /metrics` is not a JSON:API resource — it serves Prometheus text format. + *** ## Event Types Indexed diff --git a/openapi.json b/openapi.json index 0834cf9e..fd39d382 100644 --- a/openapi.json +++ b/openapi.json @@ -257,6 +257,24 @@ } } }, + "/metrics": { + "get": { + "summary": "Prometheus metrics", + "description": "Indexer and process metrics in Prometheus text exposition format. Served from in-process counters only — no database or RPC call — so it keeps answering while the subsystems it reports on are down.", + "responses": { + "200": { + "description": "Prometheus text exposition format", + "content": { + "text/plain": { + "schema": { + "type": "string" + } + } + } + } + } + } + }, "/status": { "get": { "summary": "Indexer status", diff --git a/package-lock.json b/package-lock.json index bb88b78c..0ace7a77 100644 --- a/package-lock.json +++ b/package-lock.json @@ -21,6 +21,7 @@ "graphql": "^16.11.0", "ioredis": "^5.11.1", "parquetjs-lite": "^0.8.7", + "prom-client": "^15.1.3", "ws": "^8.20.0", "zod": "^4.4.3" }, @@ -2334,6 +2335,15 @@ "url": "https://paulmillr.com/funding/" } }, + "node_modules/@opentelemetry/api": { + "version": "1.9.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.1.tgz", + "integrity": "sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==", + "license": "Apache-2.0", + "engines": { + "node": ">=8.0.0" + } + }, "node_modules/@paralleldrive/cuid2": { "version": "2.3.1", "resolved": "https://registry.npmjs.org/@paralleldrive/cuid2/-/cuid2-2.3.1.tgz", @@ -4196,6 +4206,12 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/bintrees": { + "version": "1.0.2", + "resolved": "https://registry.npmjs.org/bintrees/-/bintrees-1.0.2.tgz", + "integrity": "sha512-VOMgTMwjAaUG580SXn3LacVgjurrbMme7ZZNYGSSV7mmtY6QQRh0Eg3pwIcntQ77DErK1L0NxkbetjcoXzVwKw==", + "license": "MIT" + }, "node_modules/body-parser": { "version": "1.20.4", "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.20.4.tgz", @@ -5513,7 +5529,6 @@ "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", - "dev": true, "hasInstallScript": true, "license": "MIT", "optional": true, @@ -7887,6 +7902,20 @@ "fsevents": "2.3.3" } }, + "node_modules/prom-client": { + "version": "15.1.3", + "resolved": "https://registry.npmjs.org/prom-client/-/prom-client-15.1.3.tgz", + "integrity": "sha512-6ZiOBfCywsD4k1BN9IX0uZhF+tJkV8q8llP64G5Hajs4JOeVLPCwpPVcpXy3BwYiUGgyJzsJJQeOIv7+hDSq8g==", + "deprecated": "prom-client has been replaced by @prometheus-io/client", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/api": "^1.4.0", + "tdigest": "^0.1.1" + }, + "engines": { + "node": "^16 || ^18 || >=20" + } + }, "node_modules/proxy-addr": { "version": "2.0.7", "resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.7.tgz", @@ -8733,6 +8762,15 @@ "url": "https://opencollective.com/synckit" } }, + "node_modules/tdigest": { + "version": "0.1.3", + "resolved": "https://registry.npmjs.org/tdigest/-/tdigest-0.1.3.tgz", + "integrity": "sha512-zbRt+lT+/H4fRItHshczHErVCQnitJk8MfMT24MqFJf3YL7SJJPqGIGeuOdvxXxM/AHFzKBl7WoyaYwqO9s3Kw==", + "license": "MIT", + "dependencies": { + "bintrees": "1.0.2" + } + }, "node_modules/test-exclude": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/test-exclude/-/test-exclude-6.0.0.tgz", diff --git a/package.json b/package.json index 08f698a6..f3f37450 100644 --- a/package.json +++ b/package.json @@ -75,6 +75,7 @@ "graphql": "^16.11.0", "ioredis": "^5.11.1", "parquetjs-lite": "^0.8.7", + "prom-client": "^15.1.3", "ws": "^8.20.0", "zod": "^4.4.3" }, diff --git a/src/__tests__/metrics.test.ts b/src/__tests__/metrics.test.ts new file mode 100644 index 00000000..dae66d61 --- /dev/null +++ b/src/__tests__/metrics.test.ts @@ -0,0 +1,142 @@ +import request from "supertest"; +import { createApp } from "../api"; +import { + ledgersIndexedTotal, + transfersStoredTotal, + lastIndexedLedger, + observeDbQuery, + recordRpcError, + registry, + _resetMetrics, +} from "../metrics"; + +jest.mock("../db", () => ({ + getLastIndexedLedger: jest.fn().mockResolvedValue(1000), + getLastIndexedState: jest.fn(), + queryTransfers: jest.fn(), + queryAllTransfers: jest.fn(), + queryByTxHash: jest.fn(), + querySummary: jest.fn(), + queryNftTransfers: jest.fn(), + getNftOwner: jest.fn(), + getNftMetadata: jest.fn(), + prisma: { $queryRaw: jest.fn().mockResolvedValue([{ 1: 1 }]) }, +})); + +jest.mock("../rpc", () => ({ + getLatestLedger: jest.fn().mockResolvedValue(1050), +})); + +jest.mock("../indexer", () => ({ + getAllIndexerStats: jest.fn().mockReturnValue({}), + runningNetworks: jest.fn().mockReturnValue([]), + getIndexerStats: jest + .fn() + .mockReturnValue({ startedAt: "2024-01-01T00:00:00.000Z", uptimeSeconds: 100, totalIndexed: 50 }), +})); + +describe("Prometheus metrics (#39)", () => { + const app = createApp(); + + beforeEach(() => { + _resetMetrics(); + }); + + describe("GET /metrics", () => { + it("serves Prometheus text exposition format", async () => { + const res = await request(app).get("/metrics"); + + expect(res.status).toBe(200); + expect(res.headers["content-type"]).toContain("text/plain"); + // Every series in the exposition format is preceded by its HELP and TYPE + // lines; a body without them is not something Prometheus will scrape. + expect(res.text).toMatch(/^# HELP /m); + expect(res.text).toMatch(/^# TYPE /m); + }); + + it("exports every custom metric, declared with the right type", async () => { + const res = await request(app).get("/metrics"); + + expect(res.text).toContain("# TYPE ledgers_indexed_total counter"); + expect(res.text).toContain("# TYPE transfers_stored_total counter"); + expect(res.text).toContain("# TYPE rpc_errors_total counter"); + expect(res.text).toContain("# TYPE last_indexed_ledger gauge"); + expect(res.text).toContain("# TYPE db_query_duration_seconds histogram"); + }); + + it("reports recorded samples with their labels", async () => { + ledgersIndexedTotal.inc({ network: "testnet" }, 12); + transfersStoredTotal.inc({ network: "testnet", type: "fungible" }, 5); + transfersStoredTotal.inc({ network: "testnet", type: "nft" }, 2); + lastIndexedLedger.set({ network: "testnet" }, 987_654); + recordRpcError("retry"); + + const res = await request(app).get("/metrics"); + + expect(res.text).toContain('ledgers_indexed_total{network="testnet"} 12'); + expect(res.text).toContain('transfers_stored_total{network="testnet",type="fungible"} 5'); + expect(res.text).toContain('transfers_stored_total{network="testnet",type="nft"} 2'); + expect(res.text).toContain('last_indexed_ledger{network="testnet"} 987654'); + expect(res.text).toContain('rpc_errors_total{outcome="retry"} 1'); + }); + + it("does not depend on the database or RPC being up", async () => { + // The scrape must still answer when the things it reports on are broken — + // otherwise monitoring goes dark exactly when it is needed. + const { prisma } = jest.requireMock("../db"); + const { getLatestLedger } = jest.requireMock("../rpc"); + prisma.$queryRaw.mockRejectedValueOnce(new Error("db down")); + getLatestLedger.mockRejectedValueOnce(new Error("rpc down")); + + const res = await request(app).get("/metrics"); + + expect(res.status).toBe(200); + expect(res.text).toContain("# TYPE ledgers_indexed_total counter"); + }); + }); + + describe("observeDbQuery", () => { + it("times a successful query under its operation label", async () => { + const value = await observeDbQuery("someQuery", async () => "result"); + + expect(value).toBe("result"); + const text = await registry.metrics(); + expect(text).toContain('db_query_duration_seconds_count{operation="someQuery"} 1'); + }); + + it("still times a query that throws, and rethrows it", async () => { + // A query that takes eight seconds and then fails is the one worth + // seeing; dropping it would make the histogram describe only good runs. + await expect( + observeDbQuery("failingQuery", async () => { + throw new Error("boom"); + }) + ).rejects.toThrow("boom"); + + const text = await registry.metrics(); + expect(text).toContain('db_query_duration_seconds_count{operation="failingQuery"} 1'); + }); + }); + + describe("recordRpcError", () => { + it("separates a retried attempt from an exhausted one", async () => { + recordRpcError("retry"); + recordRpcError("retry"); + recordRpcError("exhausted"); + + const res = await request(app).get("/metrics"); + expect(res.text).toContain('rpc_errors_total{outcome="retry"} 2'); + expect(res.text).toContain('rpc_errors_total{outcome="exhausted"} 1'); + }); + }); + + describe("GET /status", () => { + it("reports last_indexed_ledger alongside the camelCase field", async () => { + const res = await request(app).get("/status"); + + expect(res.status).toBe(200); + expect(res.body.last_indexed_ledger).toBe(1000); + expect(res.body.lastIndexedLedger).toBe(1000); + }); + }); +}); diff --git a/src/api.ts b/src/api.ts index 601ce228..6425d252 100644 --- a/src/api.ts +++ b/src/api.ts @@ -24,6 +24,7 @@ import { transferQuerySchema, } from "./openapi/schemas"; import { parseOr400 } from "./openapi/validation"; +import { renderMetrics, metricsContentType } from "./metrics"; // ─── RPC Health Check Cache ─────────────────────────────────────────────── let cachedRpcHealth: { healthy: boolean; timestamp: number } | null = null; @@ -58,7 +59,9 @@ const limiter = rateLimit({ standardHeaders: true, legacyHeaders: false, message: { error: "Too many requests, please try again later." }, - skip: () => process.env.NODE_ENV === "test", + // /metrics is exempt: a scrape endpoint that starts 429ing goes blind exactly + // when load is high enough to matter, which is when you need the graphs. + skip: (req) => process.env.NODE_ENV === "test" || req.path === "/metrics", }); // ─── Response cache (opt-in via CACHE_ENABLED) ───────────────────────────── @@ -128,7 +131,9 @@ export function createApp(): express.Application { // Attaches X-Data-Stale and X-As-Of-Ledger headers plus `stale` / `as_of_ledger` // body parameters when serving DB data during an RPC outage. app.use(async (req: Request, res: Response, next: NextFunction) => { - if (req.method !== "GET" || req.path === "/healthz") { + // /metrics is served from in-process counters, so the RPC health probe + // below would add a network round-trip to every scrape and report nothing. + if (req.method !== "GET" || req.path === "/healthz" || req.path === "/metrics") { return next(); } @@ -224,6 +229,23 @@ export function createApp(): express.Application { res.json({ ok: true, uptime: process.uptime() }); }); + // ─── GET /metrics — Prometheus scrape endpoint ────────────────────────── + /** + * Indexer and process metrics in Prometheus text exposition format. + * + * Reads only in-process counters — no DB, no RPC — so it stays answerable + * while the things it is reporting on are down, which is the point. + */ + app.get("/metrics", async (_req: Request, res: Response, next: NextFunction) => { + try { + const body = await renderMetrics(); + res.setHeader("Content-Type", metricsContentType); + res.send(body); + } catch (err) { + next(err); + } + }); + // ─── GET /readyz — K8s/Render readiness probe ─────────────────────────── /** * Returns 200 when database connection is alive. @@ -343,6 +365,9 @@ export function createApp(): express.Application { ok: true, status: "healthy", lastIndexedLedger, + // snake_case alias alongside the camelCase field, for consumers that + // read the same name the metric is exported under. Both always agree. + last_indexed_ledger: lastIndexedLedger, latestLedger, lagLedgers: latestLedger - (lastIndexedLedger ?? latestLedger), ...stats, @@ -359,6 +384,7 @@ export function createApp(): express.Application { stale: true, as_of_ledger: lastIndexedLedger ?? undefined, lastIndexedLedger, + last_indexed_ledger: lastIndexedLedger, latestLedger: null, lagLedgers: null, ...stats, diff --git a/src/db.ts b/src/db.ts index 4ca80066..c723a02c 100644 --- a/src/db.ts +++ b/src/db.ts @@ -20,6 +20,7 @@ export function toDisplayAmount(amount: string): string { } import { withReadReplicas } from "./db/router"; +import { observeDbQuery } from "./metrics"; // ─── Singleton Prisma client ─────────────────────────────────────────────── // Re-use one connection pool across the process. @@ -180,10 +181,12 @@ export async function upsertTransfers( const net = resolveNetwork(network); // Prisma's createMany with skipDuplicates is the most efficient bulk path. - const result = await prisma.tokenTransfer.createMany({ - data: records.map((r) => ({ ...r, network: net })), - skipDuplicates: true, - }); + const result = await observeDbQuery("upsertTransfers", () => + prisma.tokenTransfer.createMany({ + data: records.map((r) => ({ ...r, network: net })), + skipDuplicates: true, + }) + ); return result.count; } @@ -194,9 +197,11 @@ export async function upsertTransfers( * Returns null if no state row exists yet for this network. */ export async function getLastIndexedLedger(network?: Network): Promise { - const state = await prisma.indexerState.findUnique({ - where: { network: resolveNetwork(network) }, - }); + const state = await observeDbQuery("getLastIndexedLedger", () => + prisma.indexerState.findUnique({ + where: { network: resolveNetwork(network) }, + }) + ); return state?.lastIndexedLedger ?? null; } @@ -217,11 +222,13 @@ export async function getLastIndexedState( */ export async function setLastIndexedLedger(ledger: number, network?: Network): Promise { const net = resolveNetwork(network); - await prisma.indexerState.upsert({ - where: { network: net }, - create: { network: net, lastIndexedLedger: ledger }, - update: { lastIndexedLedger: ledger }, - }); + await observeDbQuery("setLastIndexedLedger", () => + prisma.indexerState.upsert({ + where: { network: net }, + create: { network: net, lastIndexedLedger: ledger }, + update: { lastIndexedLedger: ledger }, + }) + ); } // ─── Backfill cursor helpers ─────────────────────────────────────────────── @@ -373,16 +380,18 @@ export async function queryTransfers(params: TransferQueryParams) { const cap = Math.min(limit, 200); const cursorId = decodeCursor(cursor); - const [total, transfers] = await prisma.$transaction([ - prisma.tokenTransfer.count({ where }), - prisma.tokenTransfer.findMany({ - where, - orderBy: [{ ledger: "desc" }, { id: "desc" }], - take: cap + 1, - ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : { skip: offset }), - ...(prismaSelect ? { select: prismaSelect } : {}), - }), - ]); + const [total, transfers] = await observeDbQuery("queryTransfers", () => + prisma.$transaction([ + prisma.tokenTransfer.count({ where }), + prisma.tokenTransfer.findMany({ + where, + orderBy: [{ ledger: "desc" }, { id: "desc" }], + take: cap + 1, + ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : { skip: offset }), + ...(prismaSelect ? { select: prismaSelect } : {}), + }), + ]) + ); const page = buildListPage(transfers as Array<{ id: number }>, cap); @@ -458,10 +467,12 @@ export async function upsertNftTransfers( ): Promise { if (records.length === 0) return 0; const net = resolveNetwork(network); - const result = await prisma.nftTransfer.createMany({ - data: records.map((r) => ({ ...r, network: net })), - skipDuplicates: true, - }); + const result = await observeDbQuery("upsertNftTransfers", () => + prisma.nftTransfer.createMany({ + data: records.map((r) => ({ ...r, network: net })), + skipDuplicates: true, + }) + ); return result.count; } diff --git a/src/indexer.ts b/src/indexer.ts index 10535ca9..3f002ffc 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -18,6 +18,7 @@ import { pollParallel } from "./indexer/parallel"; import { isNftTransferEvent, parseNftEvents, fetchNftMetadata } from "./ingester/nft"; import { createSourceSwitcherWithConfig, type SourceSwitcher } from "./indexer/sources"; import { currentNetwork, enabledNetworks, resolveNetwork, type Network } from "./network"; +import { ledgersIndexedTotal, transfersStoredTotal, lastIndexedLedger } from "./metrics"; // ─── NFT Contract IDs ───────────────────────────────────────────────────────── /** @@ -188,6 +189,20 @@ export function _resetIndexerLoops(): void { loops.clear(); } +/** + * Record one cursor advance for `network`. + * + * The counter takes the delta, not the absolute sequence: a resumed process + * starts from wherever the DB left it, and feeding that in as an increment + * would report a few million ledgers indexed in one second every restart. + * A non-advancing or backwards cursor contributes nothing. + */ +function recordLedgerProgress(network: Network, fromLedger: number, highestLedger: number): void { + const advanced = highestLedger - fromLedger; + if (advanced > 0) ledgersIndexedTotal.inc({ network }, advanced); + lastIndexedLedger.set({ network }, highestLedger); +} + // ─── Core poll step ─────────────────────────────────────────────────────────── /** * Fetch one batch of events starting from `fromLedger`, parse and persist them. @@ -209,6 +224,7 @@ async function pollOnce( if (events.length === 0) { await setLastIndexedLedger(highestLedger, net); + recordLedgerProgress(net, fromLedger, highestLedger); return highestLedger; } @@ -226,6 +242,7 @@ async function pollOnce( ); const inserted = await upsertTransfers(records, net); loop.totalIndexed += inserted; + transfersStoredTotal.inc({ network: net, type: "fungible" }, inserted); // Update materialized account summaries alongside transfer inserts if (inserted > 0) { @@ -255,6 +272,7 @@ async function pollOnce( const nftRecords = nftParsed.map((p) => p.record); const nftInserted = await upsertNftTransfers(nftRecords, net); loop.totalIndexed += nftInserted; + transfersStoredTotal.inc({ network: net, type: "nft" }, nftInserted); // Lazy-load metadata for unique (contractId, tokenId) pairs not yet cached if (nftParsed.length > 0) { @@ -274,6 +292,7 @@ async function pollOnce( } await setLastIndexedLedger(highestLedger, net); + recordLedgerProgress(net, fromLedger, highestLedger); console.log( `[indexer/${net}] Processed ${events.length} events → ${inserted} fungible + ${nftInserted} NFT records saved (ledger ${highestLedger})` @@ -359,6 +378,8 @@ export async function startIndexer(network?: Network): Promise { net, ); loop.totalIndexed += totalInserted; + transfersStoredTotal.inc({ network: net, type: "fungible" }, totalInserted); + recordLedgerProgress(net, currentLedger, highestLedger); currentLedger = highestLedger; } else { currentLedger = await pollOnce(loop, currentLedger, target); diff --git a/src/metrics.ts b/src/metrics.ts new file mode 100644 index 00000000..15ea399b --- /dev/null +++ b/src/metrics.ts @@ -0,0 +1,124 @@ +/** + * Prometheus metrics for the indexer and the API. + * + * Wraith runs as a persistent background service: when the indexer stalls or + * RPC starts failing, nothing outside the logs says so. These metrics are the + * machine-readable version of that signal — `GET /metrics` serves them in + * Prometheus text format for scraping, dashboards, and alerting. + * + * Everything registers into a module-local {@link registry} rather than + * prom-client's global default. A global register is process-wide state shared + * with any dependency that also uses prom-client, and it cannot be cleared + * between tests without clobbering theirs. + */ +import { Counter, Gauge, Histogram, Registry, collectDefaultMetrics } from "prom-client"; + +export const registry = new Registry(); + +// Standard process/Node metrics (process_cpu_seconds_total, heap sizes, event +// loop lag, …). Cheap, and they answer "is the process itself unhealthy?" +// before any of the counters below can. +collectDefaultMetrics({ register: registry }); + +/** + * Ledgers the indexer has advanced through, per network. + * + * The rate of this is the alert that matters: a flat + * `rate(ledgers_indexed_total[5m])` on a network whose loop is supposed to be + * running means the indexer has stalled, whether or not the process is alive. + */ +export const ledgersIndexedTotal = new Counter({ + name: "ledgers_indexed_total", + help: "Ledgers advanced through by the indexer, per network", + labelNames: ["network"] as const, + registers: [registry], +}); + +/** + * Rows written by the indexer, split by record type. + * + * `type` separates fungible transfers from NFT ones: they come off different + * parse paths and one can break while the other keeps working. + */ +export const transfersStoredTotal = new Counter({ + name: "transfers_stored_total", + help: "Transfer records persisted by the indexer, per network and record type", + labelNames: ["network", "type"] as const, + registers: [registry], +}); + +/** + * Failed RPC attempts, counted per attempt rather than per call. + * + * `withRetry` hides transient failures from callers by design, so counting only + * calls that exhausted their retries would report zero right up until the + * moment the indexer falls over. Counting attempts surfaces a degrading + * endpoint while it is still succeeding. + */ +export const rpcErrorsTotal = new Counter({ + name: "rpc_errors_total", + help: "Failed RPC attempts (each retry counts separately)", + labelNames: ["outcome"] as const, + registers: [registry], +}); + +/** + * The highest ledger the indexer has committed, per network. + * + * A gauge, not a counter: it is a position, and comparing it against the chain + * tip is how lag is measured. + */ +export const lastIndexedLedger = new Gauge({ + name: "last_indexed_ledger", + help: "Highest ledger sequence committed by the indexer, per network", + labelNames: ["network"] as const, + registers: [registry], +}); + +/** + * Wall-clock duration of instrumented database operations. + * + * Buckets run from 5ms to 10s: below 5ms nothing here is worth alerting on, + * and past 10s the request has already failed for whatever is waiting on it. + */ +export const dbQueryDurationSeconds = new Histogram({ + name: "db_query_duration_seconds", + help: "Duration of database operations, by operation name", + labelNames: ["operation"] as const, + buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], + registers: [registry], +}); + +/** + * Time `fn` into {@link dbQueryDurationSeconds} under `operation`. + * + * Failures are timed too — a query that takes eight seconds and then throws is + * exactly the one worth seeing, and dropping it would make the histogram + * describe only the healthy path. + */ +export async function observeDbQuery(operation: string, fn: () => Promise): Promise { + const end = dbQueryDurationSeconds.startTimer({ operation }); + try { + return await fn(); + } finally { + end(); + } +} + +/** Record one failed RPC attempt. `outcome` distinguishes a retry from a give-up. */ +export function recordRpcError(outcome: "retry" | "exhausted"): void { + rpcErrorsTotal.inc({ outcome }); +} + +/** Serialize the registry in Prometheus text exposition format. */ +export function renderMetrics(): Promise { + return registry.metrics(); +} + +/** The content type Prometheus expects on a scrape response. */ +export const metricsContentType = registry.contentType; + +/** Test-only: clears every recorded sample without unregistering the metrics. */ +export function _resetMetrics(): void { + registry.resetMetrics(); +} diff --git a/src/openapi/build.ts b/src/openapi/build.ts index 92247693..faca3034 100644 --- a/src/openapi/build.ts +++ b/src/openapi/build.ts @@ -73,6 +73,22 @@ registry.registerPath({ }, }); +registry.registerPath({ + method: "get", + path: "/metrics", + summary: "Prometheus metrics", + description: + "Indexer and process metrics in Prometheus text exposition format. Served from " + + "in-process counters only — no database or RPC call — so it keeps answering while " + + "the subsystems it reports on are down.", + responses: { + 200: { + description: "Prometheus text exposition format", + content: { "text/plain": { schema: { type: "string" as const } } }, + }, + }, +}); + registry.registerPath({ method: "get", path: "/status", diff --git a/src/rpc.ts b/src/rpc.ts index 16ee4a39..aec981f0 100644 --- a/src/rpc.ts +++ b/src/rpc.ts @@ -1,5 +1,6 @@ import { rpc as RPC, xdr } from "@stellar/stellar-sdk"; import { resolveNetwork, currentNetwork, type Network } from "./network"; +import { recordRpcError } from "./metrics"; // ─── Network config ─────────────────────────────────────────────────────────── const TESTNET_RPC_URL = "https://soroban-testnet.stellar.org"; @@ -162,7 +163,11 @@ export async function withRetry( return await fn(); } catch (err) { attempt++; - if (attempt >= maxAttempts) throw err; + if (attempt >= maxAttempts) { + recordRpcError("exhausted"); + throw err; + } + recordRpcError("retry"); const delay = baseDelayMs * 2 ** (attempt - 1); console.warn( `[rpc] Attempt ${attempt} failed — retrying in ${delay}ms…`,