diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts index fe14f8d8..c2da8ccf 100644 --- a/src/ingest/venues/amm.ts +++ b/src/ingest/venues/amm.ts @@ -1,4 +1,4 @@ -import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp } from '../../metrics' +import { amm_snapshots_total, trades_ingested_total, last_trade_timestamp, markIngestSuccess } from '../../metrics' import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' import { upsertPricePoints, getIndexerState, setIndexerCursor, prisma } from '../../db' import { resolvePageLedgers } from '../../ingesters/toid' @@ -106,6 +106,7 @@ export async function snapshotPool( ], network ) + markIngestSuccess('AMM', network) const trackerKey = pair.pairKey const { previousPrice } = setLastPrice('AMM', network, trackerKey, spotPrice) @@ -192,6 +193,7 @@ export async function ingestPoolTrades( ) await upsertPricePoints(points, network) + markIngestSuccess('AMM', network) trades_ingested_total.inc({ pair: pair.pairKey, network }, points.length) last_trade_timestamp.set( diff --git a/src/ingest/venues/runner.ts b/src/ingest/venues/runner.ts index 20cd3b83..9dad46b3 100644 --- a/src/ingest/venues/runner.ts +++ b/src/ingest/venues/runner.ts @@ -6,6 +6,7 @@ import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' +import { ingest_cycle_duration_seconds, ingest_errors_total, markIngestSuccess } from '../../metrics' const log = createLogger('ingester:runner') @@ -52,6 +53,7 @@ export async function writeVenuePricePoints( if (points.length === 0) return await upsertPricePoints(points as any, network) + markIngestSuccess(venue, network) const latest = points[points.length - 1] const { previousPrice, currentPrice } = setLastPrice(venue, network, pair.pairKey, latest.price) @@ -104,13 +106,26 @@ export async function startVenueAdapter( let consecutiveErrors = 0 while (true) { + const cycleStartedAt = Date.now() try { const pairs = getActivePairs() for (const pair of pairs) { await adapter.ingestPair(pair, network) } + ingest_cycle_duration_seconds.observe( + { venue: adapter.name, network }, + (Date.now() - cycleStartedAt) / 1000, + ) consecutiveErrors = 0 } catch (err) { + const error = err as Error + const category = /429|rate.?limit/i.test(error.message) ? 'rate_limit' : + /timeout|timed out|abort/i.test(error.message) ? 'timeout' : 'other' + ingest_errors_total.inc({ venue: adapter.name, network, category }) + ingest_cycle_duration_seconds.observe( + { venue: adapter.name, network }, + (Date.now() - cycleStartedAt) / 1000, + ) consecutiveErrors++ const backoffMs = Math.min(10_000 * Math.pow(2, consecutiveErrors - 1), 60_000) log.error( diff --git a/src/ingest/venues/sdex.ts b/src/ingest/venues/sdex.ts index e5167fcb..1767077c 100644 --- a/src/ingest/venues/sdex.ts +++ b/src/ingest/venues/sdex.ts @@ -1,5 +1,5 @@ import { Asset } from '@stellar/stellar-sdk' -import { trades_ingested_total, last_trade_timestamp } from '../../metrics' +import { trades_ingested_total, last_trade_timestamp, markIngestSuccess } from '../../metrics' import { config, activeNetwork, type NetworkName } from '../../config' import { getHorizonServer } from '../../network/clients' import { upsertPricePoints, getIndexerState, setIndexerCursor } from '../../db' @@ -135,6 +135,7 @@ export async function writeSdexPricePoints( const { previousPrice, currentPrice } = setLastPrice('SDEX', network, trackerKey, points[points.length - 1].price) await upsertPricePoints(points, network) + markIngestSuccess('SDEX', network) trades_ingested_total.inc({ pair: pair.pairKey, network }, points.length) last_trade_timestamp.set( diff --git a/src/ingest/venues/soroswap.ts b/src/ingest/venues/soroswap.ts index 217f3520..b8f44e41 100644 --- a/src/ingest/venues/soroswap.ts +++ b/src/ingest/venues/soroswap.ts @@ -22,6 +22,7 @@ import { import { activeNetwork, getNetworkConfig, type NetworkName } from '../../config' import { getRpcServer } from '../../network/clients' import { upsertPricePoints } from '../../db' +import { markIngestSuccess } from '../../metrics' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' import { createLogger } from '../../logger' @@ -241,6 +242,7 @@ export async function ingestPool( ], network ) + markIngestSuccess('Soroswap', network) const trackerKey = pair.pairKey const { previousPrice } = setLastPrice('Soroswap', network, trackerKey, spotPrice) diff --git a/src/metrics.ts b/src/metrics.ts index ac8ac646..611e8a04 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -85,6 +85,38 @@ export const last_trade_timestamp = new Gauge({ registers: [register] }) +// Ingestion health is labelled only by the finite adapter registry and the +// two supported networks. Do not add pair or pool identifiers here: they make +// the series count grow with user configuration and on-chain pool creation. +export const ingest_last_success_timestamp = new Gauge({ + name: 'ingest_last_success_timestamp', + help: 'Unix timestamp of the last successful price write by venue and network', + labelNames: ['venue', 'network'], + registers: [register], +}) + +export const ingest_cycle_duration_seconds = new Histogram({ + name: 'ingest_cycle_duration_seconds', + help: 'Duration of a venue ingestion cycle in seconds', + labelNames: ['venue', 'network'], + // Venue polls are mostly sub-second, but Horizon and Soroswap requests can + // take several seconds. These buckets expose normal cycles and slow calls. + buckets: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60], + registers: [register], +}) + +export const ingest_errors_total = new Counter({ + name: 'ingest_errors_total', + help: 'Ingestion cycle errors by venue, network and bounded error category', + labelNames: ['venue', 'network', 'category'], + registers: [register], +}) + +/** Record a successful persisted price write without high-cardinality labels. */ +export function markIngestSuccess(venue: string, network: string): void { + ingest_last_success_timestamp.set({ venue, network }, Math.floor(Date.now() / 1000)) +} + export const http_requests_total = new Counter({ name: 'http_requests_total', help: 'Total number of HTTP requests served, by method, route template and status class',