add prometheus metrics for query types and worker threads

New metrics:
- sqd_hotblocks_cache_queries_total: queries served from cache
- sqd_hotblocks_below_queries_total: queries below cache (spawn worker)
- sqd_hotblocks_active_workers: currently active worker threads

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: mo4islona

evm/evm-data-service/src/main.ts

@@ -82,6 +82,7 @@ runProgram(async () => {
     }
 
     let mainWorker = new WorkerClient(dataSourceOptions)
+    let service: Awaited<ReturnType<typeof runDataService>> | undefined
     let dataSource: DataSource<Block> = {
         getHead() {
             return mainWorker.getHead()
@@ -91,9 +92,11 @@ runProgram(async () => {
         },
         async *getFinalizedStream(req: StreamRequest): BlockStream<Block> {
             let worker = new WorkerClient(dataSourceOptions)
+            service?.metrics.incActiveWorkers()
             try {
                 yield* worker.getFinalizedStream(req)
             } finally {
+                service?.metrics.decActiveWorkers()
                 worker.close()
             }
         },
@@ -102,7 +105,7 @@ runProgram(async () => {
         }
     }
 
-    let service = await runDataService({
+    service = await runDataService({
         source: dataSource,
         blockCacheSize: args.blockCacheSize,
         port: args.port

util/util-internal-data-service/src/data-service.ts

@@ -54,8 +54,10 @@ export class DataService {
 
     async query(from: number, parentHash?: string): Promise<DataResponse | InvalidBaseBlock> {
         if (from <= this.chain.firstBlock().parentNumber) {
+            this.metrics.incBelowQueries()
             return this.belowQuery(from, parentHash)
         } else {
+            this.metrics.incCacheQueries()
             let res = this.chain.query(from, parentHash)
             if (res instanceof InvalidBaseBlock) return res
             if (res.tail) return res

util/util-internal-data-service/src/index.ts

@@ -2,6 +2,7 @@ import {BlockBatch, BlockStream, DataSource, StreamRequest} from '@subsquid/util
 import {ListeningServer} from '@subsquid/util-internal-http-server'
 import {DataService} from './data-service'
 import {createHttpApp} from './http-app'
+import {Metrics} from './metrics'
 import {Block, BlockHeader, BlockRef} from './types'
 
 
@@ -12,6 +13,7 @@ export {
     BlockRef,
     BlockStream,
     DataSource,
+    Metrics,
     StreamRequest
 }
 
@@ -23,7 +25,7 @@ export interface DataServiceOptions {
 }
 
 
-export async function runDataService(args: DataServiceOptions): Promise<ListeningServer & {started: Promise<void>}> {
+export async function runDataService(args: DataServiceOptions): Promise<ListeningServer & {started: Promise<void>, metrics: Metrics}> {
     let service = new DataService(args.source, args.blockCacheSize ?? 1000)
     let app = createHttpApp(service)
 
@@ -34,6 +36,7 @@ export async function runDataService(args: DataServiceOptions): Promise<Listenin
 
     return {
         started: service.started(),
+        metrics: service.metrics,
         port: server.port,
         close(): Promise<void> {
             service.stop()

util/util-internal-data-service/src/metrics.ts

@@ -11,6 +11,9 @@ export class Metrics {
     private hotBlocksStoredBlocksGauge: Gauge
     private blockLagHistogram: Histogram
     private blockProcessingTimeHistogram: Histogram
+    private cacheQueriesCounter: Counter
+    private belowQueriesCounter: Counter
+    private activeWorkersGauge: Gauge
 
     constructor() {
         this.hotBlocksLastBlockGauge = new Gauge({
@@ -57,6 +60,24 @@ export class Metrics {
             registers: [this.registry]
         });
 
+        this.cacheQueriesCounter = new Counter({
+            name: 'sqd_hotblocks_cache_queries_total',
+            help: 'Total number of queries served from cache',
+            registers: [this.registry],
+        })
+
+        this.belowQueriesCounter = new Counter({
+            name: 'sqd_hotblocks_below_queries_total',
+            help: 'Total number of below-cache queries (require worker thread)',
+            registers: [this.registry],
+        })
+
+        this.activeWorkersGauge = new Gauge({
+            name: 'sqd_hotblocks_active_workers',
+            help: 'Number of currently active worker threads',
+            registers: [this.registry],
+        })
+
         collectDefaultMetrics({register: this.registry})
     }
 
@@ -95,6 +116,22 @@ export class Metrics {
         const duration = Date.now() - startTime;
         this.blockProcessingTimeHistogram.observe(duration);
     }
+
+    incCacheQueries() {
+        this.cacheQueriesCounter.inc()
+    }
+
+    incBelowQueries() {
+        this.belowQueriesCounter.inc()
+    }
+
+    incActiveWorkers() {
+        this.activeWorkersGauge.inc()
+    }
+
+    decActiveWorkers() {
+        this.activeWorkersGauge.dec()
+    }
 }
 
 class BlockTimestampCache {