diff --git a/examples/papi/chain-exporter.ts b/examples/papi/chain-exporter.ts new file mode 100644 index 00000000..c0d77aa7 --- /dev/null +++ b/examples/papi/chain-exporter.ts @@ -0,0 +1,353 @@ +// SPDX-License-Identifier: Apache-2.0 + +/** + * Domain-state Prometheus exporter for paritytech/web3-storage#267. + * + * The chain nodes already expose standard Substrate Prometheus metrics (block + * time, weight, txpool). What no metric covers is the pallet's own domain + * state: provider balances and stake, agreement counts, pending challenges, + * checkpoint accruals, and per-class block weight (the on_finalize sweep spike + * that Substrate's own Prometheus does not expose). This sidecar polls that + * state per finalized block and exports it as Prometheus gauges, plus counters + * accumulated from lifecycle events, so a stress run (see + * `challenge-saturation.ts`) is observable on a dashboard. It is the first + * consumer of the telemetry work in #214 and the scrape target the + * `monitoring/` stack (a follow-up) will chart. + * + * Per-provider metrics are labelled; their cardinality is bounded by the small + * `MaxPrimaryProviders` / `MaxMembers`. Aggregate counts (agreements, + * challenges) are exported as totals, not per-entry, so a saturation run does + * not blow up label cardinality. + * + * Values are float64 (Prometheus' only numeric type), so balances above 2^53 + * lose low-order precision. Fine for monitoring; do not treat as exact. + * + * Usage: + * pnpm --filter web3-storage-papi-demo run chain-exporter -- [options] + * + * --chain-rpc default ws://127.0.0.1:2222 + * --port metrics HTTP port, default 9700 + */ + +import { createServer } from "node:http"; +import { parseArgs } from "node:util"; + +import { + connect, + syncSs58Prefix, + waitForChainReady, + type ChainConnection, + type ParachainApi, +} from "@web3-storage/sdk"; + +function parseCli() { + const { values } = parseArgs({ + options: { + "chain-rpc": { type: "string", default: "ws://127.0.0.1:2222" }, + port: { type: "string", default: "9700" }, + help: { type: "boolean", default: false }, + }, + }); + if (values.help) { + console.log(USAGE); + process.exit(0); + } + const port = Number(values.port); + if (!Number.isInteger(port) || port < 1 || port > 65535) { + throw new Error(`--port must be 1..65535, got ${values.port}`); + } + return { chainRpc: values["chain-rpc"], port }; +} + +const USAGE = `chain-exporter: Prometheus exporter for web3-storage pallet domain state + + --chain-rpc default ws://127.0.0.1:2222 + --port metrics HTTP port, default 9700 + +Serves /metrics (Prometheus text) and /health.`; + +// ── Metric registry ────────────────────────────────────────────────────────── + +type Sample = { labels: Record; value: number }; + +interface Metric { + help: string; + type: "gauge" | "counter"; + samples: Sample[]; +} + +const PREFIX = "web3storage_"; + +/** Minimal Prometheus registry: gauges are replaced each scrape, counters persist. */ +class Registry { + private metrics = new Map(); + + private ensure(name: string, type: Metric["type"], help: string): Metric { + let m = this.metrics.get(name); + if (!m) { + m = { help, type, samples: [] }; + this.metrics.set(name, m); + } + return m; + } + + /** Replace a gauge's samples for this scrape. */ + setGauge(name: string, value: number, labels: Record = {}) { + const m = this.ensure(name, "gauge", GAUGE_HELP[name] ?? name); + m.samples.push({ labels, value }); + } + + clearGauge(name: string) { + const m = this.metrics.get(name); + if (m && m.type === "gauge") m.samples = []; + } + + incCounter(name: string, by = 1) { + const m = this.ensure(name, "counter", COUNTER_HELP[name] ?? name); + if (m.samples.length === 0) m.samples.push({ labels: {}, value: 0 }); + m.samples[0].value += by; + } + + render(): string { + const lines: string[] = []; + for (const [name, m] of this.metrics) { + const full = PREFIX + name; + lines.push(`# HELP ${full} ${m.help}`); + lines.push(`# TYPE ${full} ${m.type}`); + for (const s of m.samples) { + const labels = Object.entries(s.labels) + .map(([k, v]) => `${k}="${escapeLabel(v)}"`) + .join(","); + lines.push(`${full}${labels ? `{${labels}}` : ""} ${s.value}`); + } + } + return lines.join("\n") + "\n"; + } +} + +function escapeLabel(v: string): string { + return v.replace(/\\/g, "\\\\").replace(/\n/g, "\\n").replace(/"/g, '\\"'); +} + +const GAUGE_HELP: Record = { + finalized_block: "Latest finalized block number scraped", + providers_total: "Registered storage providers", + buckets_total: "Layer-0 buckets", + agreements_total: "Storage agreements (all)", + agreements_active: "Storage agreements not past expires_at", + agreements_expired: "Storage agreements past expires_at, not yet swept", + challenges_pending_total: "Unresolved challenges across all deadlines", + checkpoint_rewards_pending_total: "Sum of unclaimed checkpoint rewards (planck)", + checkpoint_pool_total: "Sum of checkpoint pool balances (planck)", + block_weight_ref_time: "Latest finalized block ref_time by dispatch class (picoseconds)", + block_weight_proof_size: "Latest finalized block proof_size (PoV) by dispatch class (bytes)", + provider_stake: "Provider bonded stake (planck)", + provider_committed_bytes: "Provider committed bytes", + provider_free_balance: "Provider free balance (planck)", + provider_reserved_balance: "Provider reserved balance (planck)", + provider_pending_challenges: "Unresolved challenges against a provider", + provider_challenges_received: "Lifetime challenges received (provider stats)", + provider_challenges_failed: "Lifetime challenges failed (provider stats)", + provider_deregistering: "1 if the provider has announced deregistration", + exporter_last_scrape_block: "Block number of the last successful scrape", +}; + +const COUNTER_HELP: Record = { + challenge_created_total: "ChallengeCreated events observed", + challenge_defended_total: "ChallengeDefended events observed", + challenge_slashed_total: "ChallengeSlashed events observed", + provider_checkpoint_submitted_total: "ProviderCheckpointSubmitted events observed", + checkpoint_miss_penalized_total: "CheckpointMissPenalized events observed", + provider_registered_total: "ProviderRegistered events observed", + exporter_scrape_errors_total: "Per-block scrape failures", +}; + +/** StorageProvider event type -> counter name. */ +const EVENT_COUNTERS: Record = { + ChallengeCreated: "challenge_created_total", + ChallengeDefended: "challenge_defended_total", + ChallengeSlashed: "challenge_slashed_total", + ProviderCheckpointSubmitted: "provider_checkpoint_submitted_total", + CheckpointMissPenalized: "checkpoint_miss_penalized_total", + ProviderRegistered: "provider_registered_total", +}; + +// ── Scraping ───────────────────────────────────────────────────────────────── + +const num = (b: bigint | number): number => (typeof b === "bigint" ? Number(b) : b); + +/** Snapshot the domain-state gauges at a finalized block hash. */ +async function scrapeGauges( + api: ParachainApi, + reg: Registry, + at: string, + blockNumber: number, +): Promise { + const [providers, agreements, challenges, buckets, rewards, pools, blockWeight] = + await Promise.all([ + api.query.StorageProvider.Providers.getEntries({ at }), + api.query.StorageProvider.StorageAgreements.getEntries({ at }), + api.query.StorageProvider.Challenges.getEntries({ at }), + api.query.StorageProvider.Buckets.getEntries({ at }), + api.query.StorageProvider.CheckpointRewards.getEntries({ at }), + api.query.StorageProvider.CheckpointPool.getEntries({ at }), + api.query.System.BlockWeight.getValue({ at }), + ]); + + for (const name of PER_PROVIDER_GAUGES) reg.clearGauge(name); + for (const { keyArgs, value } of providers) { + const provider = keyArgs[0]; + const [account, pending] = await Promise.all([ + api.query.System.Account.getValue(provider, { at }), + api.query.StorageProvider.PendingChallenges.getValue(provider, { at }), + ]); + const labels = { provider }; + reg.setGauge("provider_stake", num(value.stake), labels); + reg.setGauge("provider_committed_bytes", num(value.committed_bytes), labels); + reg.setGauge("provider_free_balance", num(account.data.free), labels); + reg.setGauge("provider_reserved_balance", num(account.data.reserved), labels); + reg.setGauge("provider_pending_challenges", num(pending), labels); + reg.setGauge("provider_challenges_received", value.stats.challenges_received, labels); + reg.setGauge("provider_challenges_failed", value.stats.challenges_failed, labels); + reg.setGauge("provider_deregistering", value.deregister_at != null ? 1 : 0, labels); + } + + let active = 0; + for (const { value } of agreements) { + if (value.expires_at > blockNumber) active++; + } + const rewardsTotal = rewards.reduce((sum, e) => sum + e.value, 0n); + const poolTotal = pools.reduce((sum, e) => sum + e.value, 0n); + + reg.clearGauge("finalized_block"); + reg.setGauge("finalized_block", blockNumber); + reg.clearGauge("providers_total"); + reg.setGauge("providers_total", providers.length); + reg.clearGauge("buckets_total"); + reg.setGauge("buckets_total", buckets.length); + reg.clearGauge("agreements_total"); + reg.setGauge("agreements_total", agreements.length); + reg.clearGauge("agreements_active"); + reg.setGauge("agreements_active", active); + reg.clearGauge("agreements_expired"); + reg.setGauge("agreements_expired", agreements.length - active); + reg.clearGauge("challenges_pending_total"); + reg.setGauge("challenges_pending_total", challenges.length); + reg.clearGauge("checkpoint_rewards_pending_total"); + reg.setGauge("checkpoint_rewards_pending_total", num(rewardsTotal)); + reg.clearGauge("checkpoint_pool_total"); + reg.setGauge("checkpoint_pool_total", num(poolTotal)); + + // Per-class weight of the latest finalized block. Substrate's own Prometheus + // does not expose this; it is the signal that spikes at a challenge-deadline + // block (the on_finalize slash sweep is charged to the mandatory class). See + // challenge-saturation.ts. + reg.clearGauge("block_weight_ref_time"); + reg.clearGauge("block_weight_proof_size"); + for (const cls of ["normal", "operational", "mandatory"] as const) { + reg.setGauge("block_weight_ref_time", num(blockWeight[cls].ref_time), { class: cls }); + reg.setGauge("block_weight_proof_size", num(blockWeight[cls].proof_size), { class: cls }); + } +} + +const PER_PROVIDER_GAUGES = [ + "provider_stake", + "provider_committed_bytes", + "provider_free_balance", + "provider_reserved_balance", + "provider_pending_challenges", + "provider_challenges_received", + "provider_challenges_failed", + "provider_deregistering", +]; + +/** Count lifecycle events in one block into the persistent counters. */ +async function scrapeEvents(api: ParachainApi, reg: Registry, at: string): Promise { + const records = (await api.query.System.Events.getValue({ at })) as unknown as Array<{ + event: { type: string; value: { type: string } }; + }>; + for (const r of records) { + if (r.event?.type !== "StorageProvider") continue; + const counter = EVENT_COUNTERS[r.event.value?.type]; + if (counter) reg.incCounter(counter); + } +} + +// ── Main ───────────────────────────────────────────────────────────────────── + +async function main() { + const opts = parseCli(); + const reg = new Registry(); + // Register counters at 0 up front so they exist before their first event; + // `rate()` over a counter that only appears mid-run reads as a gap. + for (const name of Object.values(EVENT_COUNTERS)) reg.incCounter(name, 0); + reg.incCounter("exporter_scrape_errors_total", 0); + + const conn = connect(opts.chainRpc); + const { papi, api } = conn; + await waitForChainReady(api); + await syncSs58Prefix(api); + + const server = createServer((req, res) => { + if (req.url === "/metrics") { + res.writeHead(200, { "content-type": "text/plain; version=0.0.4" }); + res.end(reg.render()); + } else if (req.url === "/health") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ status: "ok" })); + } else { + res.writeHead(404).end("not found\n"); + } + }); + server.listen(opts.port, () => { + console.log(`chain-exporter on :${opts.port}/metrics, chain ${opts.chainRpc}`); + }); + + // Gauges track the latest finalized block; counters catch up over every block + // since the last one so a finalization jump does not drop events. + let lastEventBlock = -1; + // Skip a block if the previous scrape is still running: overlapping runs would + // corrupt the cleared-then-set gauge samples. The event catch-up loop reads + // from lastEventBlock+1, so a skipped block's events are counted next time. + let scraping = false; + const sub = papi.finalizedBlock$.subscribe({ + next: async (block) => { + if (scraping) return; + scraping = true; + try { + await scrapeGauges(api, reg, block.hash, block.number); + if (lastEventBlock < 0) lastEventBlock = block.number - 1; + for (let n = lastEventBlock + 1; n <= block.number; n++) { + const hash = + n === block.number + ? block.hash + : await papi._request("chain_getBlockHash", [n]); + if (hash) await scrapeEvents(api, reg, hash); + } + lastEventBlock = block.number; + reg.clearGauge("exporter_last_scrape_block"); + reg.setGauge("exporter_last_scrape_block", block.number); + } catch (err) { + reg.incCounter("exporter_scrape_errors_total"); + console.error(`scrape failed at #${block.number}: ${(err as Error)?.message ?? err}`); + } finally { + scraping = false; + } + }, + error: (err) => console.error(`finalizedBlock$ error: ${err}`), + }); + + const shutdown = () => { + sub.unsubscribe(); + server.close(); + (conn as ChainConnection).papi.destroy(); + process.exit(0); + }; + process.on("SIGINT", shutdown); + process.on("SIGTERM", shutdown); +} + +main().catch((err) => { + console.error(`❌ ${err?.message ?? err}`); + process.exit(1); +}); diff --git a/examples/papi/package.json b/examples/papi/package.json index 3b816761..d311a7a6 100644 --- a/examples/papi/package.json +++ b/examples/papi/package.json @@ -10,6 +10,7 @@ "demo:checkpoint-missed": "node --import tsx checkpoint-missed.ts", "demo:sc-flow": "node --import tsx sc-flow.ts", "e2e": "node --import tsx e2e/runner.bin.ts", + "chain-exporter": "node --import tsx chain-exporter.ts", "typecheck": "tsc --noEmit" }, "dependencies": {