feat: implement dump metrics

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
+    metricsPort?: number
 }
 
 
@@ -82,6 +84,11 @@ export class Dumper {
         })
     }
 
+    @def
+    prometheus() {
+        return new PrometheusServer(this.options.metricsPort ?? 3000);
+    }
+
     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().setLastBlock(this.options.lastBlock ? this.options.lastBlock : 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.metricsPort) {
+                await prometheus.serve();
+                this.log().info(`prometheus metrics are available on port {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: (block) => { prometheus.setLastSavedBlock(block); }
             })
         }
     }

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 <port>', 'Port to serve metrics on', nat)
 
     let args = program.parse().opts() as DumperOptions
 

substrate/substrate-dump/src/prometheus.ts

@@ -0,0 +1,40 @@
+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 lastBlockGauge: Gauge;
+    private lastSavedBlockGauge: Gauge;
+
+    constructor(port: number) {
+        this.port = port;
+        this.lastBlockGauge = new Gauge({
+            name: 'sqd_last_block_total',
+            help: 'Last block available in the chain',
+            registers: [this.registry]
+        });
+
+        this.lastSavedBlockGauge = new Gauge({
+            name: 'sqd_last_saved_block_total',
+            help: 'Last saved block',
+            registers: [this.registry]
+        });
+
+        collectDefaultMetrics({register: this.registry})
+    }
+
+    setLastBlock(block: number) {
+        this.lastBlockGauge.set(block);
+    }
+
+    setLastSavedBlock(block: number) {
+        this.lastSavedBlockGauge.set(block);
+    }
+
+
+    serve(): Promise<ListeningServer> {
+        return createPrometheusServer(this.registry, this.port)
+    }
+}

util/util-internal-archive-layout/src/layout.ts

@@ -172,7 +172,8 @@ export class ArchiveLayout {
         args: {
             blocks: (nextBlock: number, prevHash?: string) => AsyncIterable<HashAndHeight[]>
             range?: Range
-            chunkSize?: number
+            chunkSize?: number,
+            onSuccessWrite?: (lastBlock: number) => void
         }
     ): Promise<void> {
         return this.append(
@@ -190,7 +191,9 @@ export class ArchiveLayout {
                         assertNotNull(lastBlock)
                     ).transactDir('.', async fs => {
                         let content = await out.end()
-                        return fs.write('blocks.jsonl.gz', content)
+                        const writeResult = fs.write('blocks.jsonl.gz', content)
+                        writeResult.then(() => args.onSuccessWrite ? args.onSuccessWrite(lastBlock?.height || 0) : null)
+                        return writeResult
                     })
                     firstBlock = undefined
                     lastBlock = undefined