feat: implement dump metrics

common/changes/@subsquid/substrate-dump/feat-dump-metrics_2023-09-25-10-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/substrate-dump",
+      "comment": "feat: implement prometheus metrics",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/substrate-dump"
+}

common/changes/@subsquid/util-internal-archive-layout/feat-dump-metrics_2023-09-25-10-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@subsquid/util-internal-archive-layout",
+      "comment": "feat: implement onSuccessWrite callback",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@subsquid/util-internal-archive-layout"
+}

substrate/substrate-dump/package.json

@@ -27,8 +27,10 @@
     "@subsquid/util-internal-counters": "^1.3.0",
     "@subsquid/util-internal-fs": "^0.1.0",
     "@subsquid/util-internal-hex": "^1.2.0",
+    "@subsquid/util-internal-prometheus-server": "^1.2.0",
     "@subsquid/util-internal-range": "^0.0.0",
-    "commander": "^11.0.0"
+    "commander": "^11.0.0",
+    "prom-client": "^14.2.0"
   },
   "devDependencies": {
     "@types/node": "^18.16.17",

substrate/substrate-dump/src/dumper.ts

@@ -14,6 +14,7 @@ import {printTimeInterval, Progress} from '@subsquid/util-internal-counters'
 import {createFs, Fs} from '@subsquid/util-internal-fs'
 import {assertRange, printRange, Range, rangeEnd} from '@subsquid/util-internal-range'
 import {MetadataWriter} from './metadata'
+import { PrometheusServer } from './prometheus'
 
 
 export interface DumperOptions {
@@ -26,6 +27,7 @@ export interface DumperOptions {
     lastBlock?: number
     withTrace?: boolean | string
     chunkSize: number
+    metrics?: number
 }
 
 
@@ -82,6 +84,11 @@ export class Dumper {
         })
     }
 
+    @def
+    prometheus() {
+        return new PrometheusServer(this.options.metrics ?? 0, this.rpc())
+    }
+
     ingest(range: Range): AsyncIterable<BlockBatch> {
         let request: DataRequest = {
             runtimeVersion: true,
@@ -107,6 +114,7 @@ export class Dumper {
 
         let height = new Throttler(() => this.src().getFinalizedHeight(), 300_000)
         let chainHeight = await height.get()
+        this.prometheus().setChainHeight(chainHeight)
 
         let progress = new Progress({
             initialValue: this.range().from,
@@ -193,11 +201,17 @@ export class Dumper {
                 }
             }
         } else {
-            let archive = new ArchiveLayout(this.fs())
+            const archive = new ArchiveLayout(this.fs())
+            const prometheus = this.prometheus()
+            if (this.options.metrics != null) {
+                const promServer = await prometheus.serve()
+                this.log().info(`prometheus metrics are available on port ${promServer.port}`)
+            }
             await archive.appendRawBlocks({
                 blocks: (nextBlock, prevHash) => this.saveMetadata(this.process(nextBlock, prevHash)),
                 range: this.range(),
-                chunkSize: this.options.chunkSize * 1024 * 1024
+                chunkSize: this.options.chunkSize * 1024 * 1024,
+                onSuccessWrite: ({ blockRange: { to } }) => { prometheus.setLastWrittenBlock(to.height) }
             })
         }
     }

substrate/substrate-dump/src/main.ts

@@ -22,6 +22,7 @@ runProgram(() => {
     program.option('--last-block <number>', 'Height of the last block to dump', nat)
     program.option('--with-trace [targets]', 'Fetch block trace')
     program.option('--chunk-size <MB>', 'Data chunk size in megabytes', positiveInt, 32)
+    program.option('--metrics <port>', 'Enable prometheus metrics server', nat)
 
     let args = program.parse().opts() as DumperOptions
 

substrate/substrate-dump/src/prometheus.ts

@@ -0,0 +1,59 @@
+import { RpcClient } from '@subsquid/rpc-client';
+import {createPrometheusServer, ListeningServer} from '@subsquid/util-internal-prometheus-server'
+import promClient, { collectDefaultMetrics, Gauge, Registry } from 'prom-client';
+
+
+export class PrometheusServer {
+    private registry = new Registry()
+    private port?: number | string
+    private chainHeightGauge: Gauge;
+    private lastWrittenBlockGauge: Gauge;
+    private rpcRequestsGauge: Gauge;
+
+    constructor(port: number, rpc: RpcClient) {
+        this.port = port;
+        this.chainHeightGauge = new Gauge({
+            name: 'sqd_dump_chain_height',
+            help: 'Last block available in the chain',
+            registers: [this.registry]
+        });
+
+        this.lastWrittenBlockGauge = new Gauge({
+            name: 'sqd_dump_last_written_block',
+            help: 'Last saved block',
+            registers: [this.registry]
+        });
+
+        this.rpcRequestsGauge = new Gauge({
+            name: 'sqd_dump_rpc_requests_count',
+            help: 'Number of rpc requests of different kinds',
+            labelNames: ['kind', 'url'],
+            registers: [this.registry],
+            collect() {
+                const metrics = rpc.getMetrics();
+                this.set({
+                    kind: 'successful',
+                    url: rpc.url
+                }, metrics.requestsServed);
+                this.set({
+                    kind: 'failed',
+                    url: rpc.url
+                }, metrics.connectionErrors);
+            }
+        });
+
+        collectDefaultMetrics({register: this.registry})
+    }
+
+    setChainHeight(height: number) {
+        this.chainHeightGauge.set(height);
+    }
+
+    setLastWrittenBlock(block: number) {
+        this.lastWrittenBlockGauge.set(block);
+    }
+
+    serve(): Promise<ListeningServer> {
+        return createPrometheusServer(this.registry, this.port)
+    }
+}