Export metrics for watcher config and upstream and external chain heads (#497)

* Add a cli to export chain head block numbers

* Use ETH RPC endpoint and allow env overrides

* Use ethers provider

* Export upstream chain head block number in watcher metrics

* Remove unnecessary exports

* Upgrade package versions

* Fix defaults usage

* Export watcher config in metrics

* Add metric for watcher sync mode

* Remove cache flag from watcher config metrics

* Update watcher config field names

lerna.json

@@ -2,7 +2,7 @@
   "packages": [
     "packages/*"
   ],
-  "version": "0.2.78",
+  "version": "0.2.79",
   "npmClient": "yarn",
   "useWorkspaces": true,
   "command": {

packages/cache/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cache",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "description": "Generic object cache",
   "main": "dist/index.js",
   "scripts": {

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cli",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "scripts": {
@@ -10,17 +10,18 @@
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
     "chat": "DEBUG='vulcanize:*, laconic:*' node dist/chat.js",
     "compare-gql": "DEBUG='vulcanize:*' node dist/compare-gql.js",
-    "proxy": "DEBUG='laconic:*' node dist/proxy.js"
+    "proxy": "DEBUG='laconic:*' node dist/proxy.js",
+    "export-metrics:chain-heads": "DEBUG='laconic:*' node dist/chain-head-exporter.js"
   },
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.78",
+    "@cerc-io/cache": "^0.2.79",
-    "@cerc-io/ipld-eth-client": "^0.2.78",
+    "@cerc-io/ipld-eth-client": "^0.2.79",
     "@cerc-io/libp2p": "^0.42.2-laconic-0.1.4",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.78",
+    "@cerc-io/peer": "^0.2.79",
-    "@cerc-io/rpc-eth-client": "^0.2.78",
+    "@cerc-io/rpc-eth-client": "^0.2.79",
-    "@cerc-io/util": "^0.2.78",
+    "@cerc-io/util": "^0.2.79",
     "@ethersproject/providers": "^5.4.4",
     "@graphql-tools/utils": "^9.1.1",
     "@ipld/dag-cbor": "^8.0.0",

packages/cli/src/chain-head-exporter.ts

@@ -0,0 +1,87 @@
+//
+// Copyright 2023 Vulcanize, Inc.
+//
+
+import express from 'express';
+import * as promClient from 'prom-client';
+import debug from 'debug';
+import { ethers } from 'ethers';
+import JsonRpcProvider = ethers.providers.JsonRpcProvider;
+
+import { fetchLatestBlockNumber } from '@cerc-io/util';
+
+const log = debug('laconic:chain-head-exporter');
+
+// Env overrides:
+// ETH_RPC_ENDPOINT - Ethereum RPC API endpoint
+// ETH_RPC_API_KEY  - Ethereum RPC API endpoint key
+// FIL_RPC_ENDPOINT - Filecoin RPC API endpoint
+// PORT             - Metrics server listening port
+
+// Defaults
+const DEFAULT_ETH_RPC_ENDPOINT = 'https://mainnet.infura.io/v3';
+const DEFAULT_FIL_RPC_ENDPOINT = 'https://api.node.glif.io/rpc/v1';
+const DEFAULT_PORT = 5000;
+
+async function main (): Promise<void> {
+  const app = express();
+  const metricsRegister = new promClient.Registry();
+
+  const ethRpcApiKey = process.env.ETH_RPC_API_KEY;
+  if (!ethRpcApiKey) {
+    log('WARNING: ETH_RPC_API_KEY not set');
+  }
+
+  const ethRpcBaseUrl = process.env.ETH_RPC_ENDPOINT || DEFAULT_ETH_RPC_ENDPOINT;
+  const ethUrlSuffix = ethRpcApiKey ? `/${ethRpcApiKey}` : '';
+  const ethRpcUrl = `${ethRpcBaseUrl}${ethUrlSuffix}`;
+  let ethProvider: JsonRpcProvider;
+  try {
+    ethProvider = new JsonRpcProvider(ethRpcUrl);
+  } catch (err) {
+    log(`Error creating ETH RPC provider from URL ${ethRpcBaseUrl}`, err);
+  }
+
+  const filRpcUrl = process.env.FILECOIN_RPC_ENDPOINT || DEFAULT_FIL_RPC_ENDPOINT;
+  let filProvider: JsonRpcProvider;
+  try {
+    filProvider = new JsonRpcProvider(filRpcUrl);
+  } catch (err) {
+    log(`Error creating FIL RPC provider from URL ${filRpcUrl}`, err);
+  }
+
+  // eslint-disable-next-line no-new
+  new promClient.Gauge({
+    name: 'latest_block_number',
+    help: 'Latest block number / height from various block chains',
+    registers: [metricsRegister],
+    labelNames: ['chain'] as const,
+    async collect () {
+      const [
+        latestEthBlockNumber,
+        latestFilBlockNumber
+      ] = await Promise.all([
+        fetchLatestBlockNumber(ethProvider),
+        fetchLatestBlockNumber(filProvider)
+      ]);
+
+      this.set({ chain: 'ethereum' }, latestEthBlockNumber);
+      this.set({ chain: 'filecoin' }, latestFilBlockNumber);
+    }
+  });
+
+  app.get('/metrics', async (req, res) => {
+    res.set('Content-Type', metricsRegister.contentType);
+    const metrics = await metricsRegister.metrics();
+    res.send(metrics);
+  });
+
+  const port = Number(process.env.PORT) || DEFAULT_PORT;
+  app.listen(port, () => {
+    log(`Server running on port ${port}`);
+  });
+}
+
+main().catch(err => {
+  log(err);
+});

packages/cli/src/index.ts

@@ -17,4 +17,3 @@ export * from './fill';
 export * from './create-state-gql';
 export * from './peer';
 export * from './utils';
-export * from './proxy';

packages/codegen/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/codegen",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "description": "Code generator",
   "private": true,
   "main": "index.js",
@@ -20,7 +20,7 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/util": "^0.2.78",
+    "@cerc-io/util": "^0.2.79",
     "@graphql-tools/load-files": "^6.5.2",
     "@npmcli/package-json": "^5.0.0",
     "@poanet/solidity-flattener": "https://github.com/vulcanize/solidity-flattener.git",

packages/codegen/src/templates/package-template.handlebars

@@ -41,12 +41,12 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/cli": "^0.2.78",
+    "@cerc-io/cli": "^0.2.79",
-    "@cerc-io/ipld-eth-client": "^0.2.78",
+    "@cerc-io/ipld-eth-client": "^0.2.79",
-    "@cerc-io/solidity-mapper": "^0.2.78",
+    "@cerc-io/solidity-mapper": "^0.2.79",
-    "@cerc-io/util": "^0.2.78",
+    "@cerc-io/util": "^0.2.79",
     {{#if (subgraphPath)}}
-    "@cerc-io/graph-node": "^0.2.78",
+    "@cerc-io/graph-node": "^0.2.79",
     {{/if}}
     "@ethersproject/providers": "^5.4.4",
     "debug": "^4.3.1",

packages/graph-node/package.json

@@ -1,10 +1,10 @@
 {
   "name": "@cerc-io/graph-node",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {
-    "@cerc-io/solidity-mapper": "^0.2.78",
+    "@cerc-io/solidity-mapper": "^0.2.79",
     "@ethersproject/providers": "^5.4.4",
     "@graphprotocol/graph-ts": "^0.22.0",
     "@nomiclabs/hardhat-ethers": "^2.0.2",
@@ -51,9 +51,9 @@
   "dependencies": {
     "@apollo/client": "^3.3.19",
     "@cerc-io/assemblyscript": "0.19.10-watcher-ts-0.1.2",
-    "@cerc-io/cache": "^0.2.78",
+    "@cerc-io/cache": "^0.2.79",
-    "@cerc-io/ipld-eth-client": "^0.2.78",
+    "@cerc-io/ipld-eth-client": "^0.2.79",
-    "@cerc-io/util": "^0.2.78",
+    "@cerc-io/util": "^0.2.79",
     "@types/json-diff": "^0.5.2",
     "@types/yargs": "^17.0.0",
     "bn.js": "^4.11.9",

packages/ipld-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/ipld-eth-client",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "description": "IPLD ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -20,8 +20,8 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.78",
+    "@cerc-io/cache": "^0.2.79",
-    "@cerc-io/util": "^0.2.78",
+    "@cerc-io/util": "^0.2.79",
     "cross-fetch": "^3.1.4",
     "debug": "^4.3.1",
     "ethers": "^5.4.4",

packages/peer/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/peer",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "description": "libp2p module",
   "main": "dist/index.js",
   "exports": "./dist/index.js",

packages/rpc-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/rpc-eth-client",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "description": "RPC ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -19,9 +19,9 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/cache": "^0.2.78",
+    "@cerc-io/cache": "^0.2.79",
-    "@cerc-io/ipld-eth-client": "^0.2.78",
+    "@cerc-io/ipld-eth-client": "^0.2.79",
-    "@cerc-io/util": "^0.2.78",
+    "@cerc-io/util": "^0.2.79",
     "chai": "^4.3.4",
     "ethers": "^5.4.4",
     "left-pad": "^1.3.0",

packages/solidity-mapper/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/solidity-mapper",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {

packages/test/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/test",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "private": true,

packages/tracing-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/tracing-client",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "description": "ETH VM tracing client",
   "main": "dist/index.js",
   "scripts": {

packages/util/package.json

@@ -1,13 +1,13 @@
 {
   "name": "@cerc-io/util",
-  "version": "0.2.78",
+  "version": "0.2.79",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "dependencies": {
     "@apollo/utils.keyvaluecache": "^1.0.1",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.78",
+    "@cerc-io/peer": "^0.2.79",
-    "@cerc-io/solidity-mapper": "^0.2.78",
+    "@cerc-io/solidity-mapper": "^0.2.79",
     "@cerc-io/ts-channel": "1.0.3-ts-nitro-0.1.1",
     "@ethersproject/properties": "^5.7.0",
     "@ethersproject/providers": "^5.4.4",
@@ -52,7 +52,7 @@
     "yargs": "^17.0.1"
   },
   "devDependencies": {
-    "@cerc-io/cache": "^0.2.78",
+    "@cerc-io/cache": "^0.2.79",
     "@nomiclabs/hardhat-waffle": "^2.0.1",
     "@types/bunyan": "^1.8.8",
     "@types/express": "^4.17.14",

packages/util/src/events.ts

@@ -102,7 +102,7 @@ export class EventWatcher {
     await this._jobQueue.waitForEmptyQueue(QUEUE_EVENT_PROCESSING);
 
     // Get latest block in chain and sync status from DB
-    // Also get historical-processing queu size
+    // Also get historical-processing queue size
     const [{ block: latestBlock }, syncStatus, historicalProcessingQueueSize] = await Promise.all([
       this._ethClient.getBlockByHash(),
       this._indexer.getSyncStatus(),

packages/util/src/job-runner.ts

@@ -31,7 +31,7 @@ import {
   fetchBlocksAtHeight,
   fetchAndSaveFilteredLogsAndBlocks
 } from './common';
-import { lastBlockNumEvents, lastBlockProcessDuration, lastProcessedBlockNumber } from './metrics';
+import { isSyncingHistoricalBlocks, lastBlockNumEvents, lastBlockProcessDuration, lastProcessedBlockNumber } from './metrics';
 
 const log = debug('vulcanize:job-runner');
 
@@ -116,6 +116,9 @@ export class JobRunner {
 
     switch (kind) {
       case JOB_KIND_INDEX: {
+        // Update metrics
+        isSyncingHistoricalBlocks.set(0);
+
         const { data: { cid, blockHash, blockNumber, parentHash, timestamp } } = job;
 
         // Check if blockHash present in job.
@@ -166,6 +169,9 @@ export class JobRunner {
   }
 
   async processHistoricalBlocks (job: PgBoss.JobWithDoneCallback<HistoricalJobData, HistoricalJobResponseData>): Promise<void> {
+    // Update metrics
+    isSyncingHistoricalBlocks.set(1);
+
     const { data: { blockNumber: startBlock, processingEndBlockNumber } } = job;
 
     if (this._historicalProcessingCompletedUpto) {
@@ -640,10 +646,6 @@ export class JobRunner {
       );
       console.timeEnd(`time:job-runner#_processEvents-events-${block.blockNumber}`);
 
-      // Update metrics
-      lastProcessedBlockNumber.set(block.blockNumber);
-      lastBlockNumEvents.set(block.numEvents);
-
       this._blockAndEventsMap.delete(block.blockHash);
 
       // Check if new contract was added and filterLogsByAddresses is set to true
@@ -671,9 +673,12 @@ export class JobRunner {
         await this.jobQueue.deleteJobs(QUEUE_EVENT_PROCESSING);
       }
 
+      // Update metrics
       if (this._endBlockProcessTimer) {
         this._endBlockProcessTimer();
       }
+      lastProcessedBlockNumber.set(block.blockNumber);
+      lastBlockNumEvents.set(block.numEvents);
 
       this._endBlockProcessTimer = lastBlockProcessDuration.startTimer();
       await this._indexer.updateSyncStatusProcessedBlock(block.blockHash, block.blockNumber);

packages/util/src/metrics.ts

@@ -7,6 +7,8 @@ import express, { Application } from 'express';
 import { createConnection } from 'typeorm';
 import debug from 'debug';
 import assert from 'assert';
+import { ethers } from 'ethers';
+import JsonRpcProvider = ethers.providers.JsonRpcProvider;
 
 import { Config } from './config';
 import { IndexerInterface } from './types';
@@ -15,6 +17,15 @@ const DB_SIZE_QUERY = 'SELECT pg_database_size(current_database())';
 
 const log = debug('vulcanize:metrics');
 
+export async function fetchLatestBlockNumber (provider: JsonRpcProvider): Promise<number> {
+  try {
+    return await provider.getBlockNumber();
+  } catch (err) {
+    log('Error fetching latest block number', err);
+    return -1;
+  }
+}
+
 // Create custom metrics
 export const jobCount = new client.Gauge({
   name: 'pgboss_jobs_total',
@@ -78,6 +89,12 @@ export const eventProcessingEthCallDuration = new client.Histogram({
   help: 'Duration of eth_calls made in event processing'
 });
 
+export const isSyncingHistoricalBlocks = new client.Gauge({
+  name: 'is_syncing_historical_blocks',
+  help: 'Whether the watcher is syncing in historical mode'
+});
+isSyncingHistoricalBlocks.set(Number(undefined));
+
 // Export metrics on a server
 const app: Application = express();
 
@@ -107,8 +124,12 @@ export const startMetricsServer = async (config: Config, indexer: IndexerInterfa
     }
   });
 
+  await registerWatcherConfigMetrics(config);
+
   await registerDBSizeMetrics(config);
 
+  await registerUpstreamChainHeadMetrics(config);
+
   // Collect default metrics
   client.collectDefaultMetrics();
 
@@ -157,3 +178,44 @@ const registerDBSizeMetrics = async ({ database, jobQueue }: Config): Promise<vo
     }
   });
 };
+
+const registerUpstreamChainHeadMetrics = async ({ upstream }: Config): Promise<void> => {
+  const ethRpcProvider = new JsonRpcProvider(upstream.ethServer.rpcProviderEndpoint);
+
+  // eslint-disable-next-line no-new
+  new client.Gauge({
+    name: 'latest_upstream_block_number',
+    help: 'Latest upstream block number',
+    async collect () {
+      const latestBlockNumber = await fetchLatestBlockNumber(ethRpcProvider);
+      this.set(latestBlockNumber);
+    }
+  });
+};
+
+const registerWatcherConfigMetrics = async ({ server, upstream, jobQueue }: Config): Promise<void> => {
+  const watcherConfigMetric = new client.Gauge({
+    name: 'watcher_config_info',
+    help: 'Watcher configuration info (static)',
+    labelNames: ['category', 'field']
+  });
+
+  watcherConfigMetric.set({ category: 'server', field: 'is_active' }, Number(server.kind === 'active'));
+  watcherConfigMetric.set({ category: 'server', field: 'is_subgraph_watcher' }, Number(server.subgraphPath?.length > 0));
+  watcherConfigMetric.set({ category: 'server', field: 'max_events_block_range' }, Number(server.maxEventsBlockRange));
+  watcherConfigMetric.set({ category: 'server', field: 'clear_entities_cache_interval' }, Number(server.clearEntitiesCacheInterval));
+  watcherConfigMetric.set({ category: 'server', field: 'max_simultaneous_requests' }, Number(server.maxSimultaneousRequests));
+  watcherConfigMetric.set({ category: 'server', field: 'max_request_queue_limit' }, Number(server.maxRequestQueueLimit));
+
+  watcherConfigMetric.set({ category: 'upstream', field: 'is_using_rpc_client' }, Number(upstream.ethServer.rpcClient));
+  watcherConfigMetric.set({ category: 'upstream', field: 'is_fevm' }, Number(upstream.ethServer.isFEVM));
+  watcherConfigMetric.set({ category: 'server', field: 'rpc_supports_block_hash' }, Number(server.rpcSupportsBlockHashParam));
+  watcherConfigMetric.set({ category: 'upstream', field: 'filter_logs_by_addresses' }, Number(upstream.ethServer.filterLogsByAddresses));
+  watcherConfigMetric.set({ category: 'upstream', field: 'filter_logs_by_topics' }, Number(upstream.ethServer.filterLogsByTopics));
+
+  watcherConfigMetric.set({ category: 'jobqueue', field: 'num_events_in_batch' }, Number(jobQueue.eventsInBatch));
+  watcherConfigMetric.set({ category: 'jobqueue', field: 'block_delay_seconds' }, (Number(jobQueue.blockDelayInMilliSecs) || 0) / 1000);
+  watcherConfigMetric.set({ category: 'jobqueue', field: 'use_block_ranges' }, Number(jobQueue.useBlockRanges));
+  watcherConfigMetric.set({ category: 'jobqueue', field: 'historical_logs_block_range' }, Number(jobQueue.historicalLogsBlockRange));
+  watcherConfigMetric.set({ category: 'jobqueue', field: 'historical_max_fetch_ahead' }, Number(jobQueue.historicalMaxFetchAhead));
+};