diff --git a/scripts/grafana-dashboard.json b/scripts/grafana-dashboard.json index 021cf83..550ad2f 100644 --- a/scripts/grafana-dashboard.json +++ b/scripts/grafana-dashboard.json @@ -471,6 +471,104 @@ ], "title": "Avg Batch Size", "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { + "axisBorderShow": false, + "axisCenteredZero": false, + "axisColorMode": "text", + "axisLabel": "", + "axisPlacement": "auto", + "barAlignment": 0, + "drawStyle": "bars", + "fillOpacity": 80, + "gradientMode": "none", + "hideFrom": { "legend": false, "tooltip": false, "viz": false }, + "insertNulls": false, + "lineInterpolation": "linear", + "lineWidth": 1, + "pointSize": 5, + "scaleDistribution": { "type": "linear" }, + "showPoints": "never", + "spanNulls": false, + "stacking": { "group": "A", "mode": "normal" }, + "thresholdsStyle": { "mode": "off" } + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null } + ] + }, + "unit": "short" + } + }, + "gridPos": { "h": 10, "w": 12, "x": 0, "y": 30 }, + "id": 9, + "options": { + "legend": { "calcs": ["sum"], "displayMode": "table", "placement": "bottom", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "pluginVersion": "11.0.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "increase(keeper_charge_results_total[$__rate_interval])", + "legendFormat": "{{result}}", + "range": true, + "refId": "A" + } + ], + "title": "Charge Results Breakdown", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "orange", "value": 1 }, + { "color": "red", "value": 5 } + ] + }, + "unit": "short" + } + }, + "gridPos": { "h": 10, "w": 12, "x": 12, "y": 30 }, + "id": 10, + "options": { + "colorMode": "background", + "graphMode": "area", + "justifyMode": "auto", + "orientation": "auto", + "percentChangeColorMode": "standard", + "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false }, + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true + }, + "pluginVersion": "11.0.0", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "editorMode": "code", + "expr": "rate(keeper_rpc_failovers_total[$__rate_interval])", + "refId": "A" + } + ], + "title": "RPC Failover Rate", + "type": "stat" } ], "refresh": "10s", diff --git a/scripts/keeper.ts b/scripts/keeper.ts index ee9c861..a8253e7 100644 --- a/scripts/keeper.ts +++ b/scripts/keeper.ts @@ -53,6 +53,13 @@ import { MultiEndpointServer } from "./rpc-client.js"; import { buildOptimizedBatches } from "./batch-optimizer"; import { Server, assembleTransaction } from "@stellar/stellar-sdk/rpc"; import { buildOptimizedBatches } from "./batch-optimizer.js"; +import { + startMetricsServer, + recordBatchCharge, + recordChargeResults, + incrementCycles, + setActiveSubscribers, +} from "./metrics-server"; import { Address, Contract, @@ -656,14 +663,26 @@ async function processPageDryRun( if (users.length === 0) return result; - const { results, amounts } = await simulateBatchCharge(users); + const startMs = Date.now(); + try { + const { results, amounts } = await simulateBatchCharge(users); - // results is index-aligned with users (same order, one entry per input). - let amountIdx = 0; - for (let i = 0; i < results.length; i++) { - const variant = results[i]; - const user = i < users.length ? users[i] : "unknown"; + // results is index-aligned with users (same order, one entry per input). + let amountIdx = 0; + for (let i = 0; i < results.length; i++) { + const variant = results[i]; + const user = i < users.length ? users[i] : "unknown"; + if (variant === "Charged") { + const amt = amountIdx < amounts.length ? amounts[amountIdx] : 0n; + result.wouldCharge++; + result.totalVolume += amt; + result.candidates.push({ user, result: variant, amountStroops: amt.toString() }); + amountIdx++; + } else { + result.skipCounts[variant] = (result.skipCounts[variant] || 0) + 1; + result.candidates.push({ user, result: variant, amountStroops: "0" }); + } if (variant === "Charged") { const amt = amountIdx < amounts.length ? amounts[amountIdx] : 0n; result.wouldCharge++; @@ -678,6 +697,26 @@ async function processPageDryRun( result.skipCounts[variant] = (result.skipCounts[variant] || 0) + 1; result.candidates.push({ user, result: variant, amountStroops: "0" }); } + + const durationMs = Date.now() - startMs; + recordBatchCharge({ + status: "success", + charged: result.wouldCharge, + skipped: users.length - result.wouldCharge, + durationMs, + }); + recordChargeResults({ Charged: result.wouldCharge, ...result.skipCounts }); + } catch (err) { + const durationMs = Date.now() - startMs; + recordBatchCharge({ + status: "failed", + charged: 0, + skipped: 0, + durationMs, + rpcError: true, + }); + const errorStr = err instanceof Error ? err.message : String(err); + result.errors.push(`Page ${pageOffset}: ${errorStr}`); } return result; @@ -697,6 +736,8 @@ async function processPageLive( if (users.length === 0) return result; + const startMs = Date.now(); + try { const { results, amounts, txHash } = await submitBatchCharge(users); result.txHash = txHash; @@ -722,7 +763,26 @@ async function processPageLive( result.candidates.push({ user, result: variant, amountStroops: "0" }); } } + + const durationMs = Date.now() - startMs; + recordBatchCharge({ + status: "success", + charged: result.charged, + skipped: users.length - result.charged, + durationMs, + }); + recordChargeResults({ Charged: result.charged, ...result.skipCounts }); + } catch (err) { + const durationMs = Date.now() - startMs; + recordBatchCharge({ + status: "failed", + charged: 0, + skipped: 0, + durationMs, + rpcError: true, + }); + const errorStr = err instanceof Error ? err.message : String(err); result.errors.push(`Page ${pageOffset}: ${errorStr}`); let ledgerSeq: number | undefined; @@ -825,6 +885,8 @@ async function runCycle(): Promise { // ── Post-cycle reporting ─────────────────────────────────────────────────── + setActiveSubscribers(report.totalChecked); + if (!isDryRun) { writeLatestLive(report); } else { @@ -1149,17 +1211,20 @@ async function main(): Promise { log(false, "Keeper started in LIVE mode"); } + startMetricsServer(); BATCH_SIZE = await resolveBatchSize(); log(DRY_RUN, `Effective legacy page size (BATCH_SIZE): ${BATCH_SIZE}`); if (once) { const report = await runCycle(); + incrementCycles(); process.exit(report.errors.length > 0 && report.totalCharged === 0 ? 1 : 0); } // Loop mode while (true) { const report = await runCycle(); + incrementCycles(); const nextRun = new Date(Date.now() + INTERVAL_SECONDS * 1000); log( DRY_RUN, diff --git a/scripts/metrics-server.ts b/scripts/metrics-server.ts index 60d92ea..062b314 100644 --- a/scripts/metrics-server.ts +++ b/scripts/metrics-server.ts @@ -115,6 +115,21 @@ const indexerDedupEvictionsTotal = new Counter({ registers: [registry], }); +/** Total RPC failovers across multiple endpoints. */ +const rpcFailoversTotal = new Counter({ + name: "keeper_rpc_failovers_total", + help: "Total number of RPC failovers triggered", + registers: [registry], +}); + +/** Granular outcomes for each subscriber checked. */ +const chargeResultsTotal = new Counter({ + name: "keeper_charge_results_total", + help: "Total number of charge outcomes labeled by specific contract result", + labelNames: ["result"] as const, + registers: [registry], +}); + // ── Public API for the keeper run loop ─────────────────────────────────────── /** @@ -149,11 +164,23 @@ export function recordBatchCharge(params: { } } +/** Record specific granular charge results. */ +export function recordChargeResults(results: Record): void { + for (const [result, count] of Object.entries(results)) { + chargeResultsTotal.inc({ result }, count); + } +} + /** Increment the RPC error counter. */ export function incrementRpcErrors(): void { rpcErrorsTotal.inc(1); } +/** Increment the RPC failovers counter. */ +export function incrementRpcFailovers(): void { + rpcFailoversTotal.inc(1); +} + /** Set the current active subscriber count gauge. */ export function setActiveSubscribers(count: number): void { activeSubscribers.set(count); diff --git a/scripts/rpc-client.ts b/scripts/rpc-client.ts index 45e8f31..da0042c 100644 --- a/scripts/rpc-client.ts +++ b/scripts/rpc-client.ts @@ -8,6 +8,7 @@ import { Networks, } from "@stellar/stellar-sdk"; import { Server, Durability, Api } from "@stellar/stellar-sdk/rpc"; +import { incrementRpcFailovers } from "./metrics-server"; /** * Interface representing state of a single RPC endpoint. @@ -167,6 +168,7 @@ export class MultiEndpointServer { console.warn( `[RPC Failover] Endpoint ${failedUrl} failed: ${err?.message || err}. Retrying with ${nextUrl}...`, ); + incrementRpcFailovers(); } }