feat(metrics): Add queue depth metrics

src/arweave/composite-client.ts

@@ -781,4 +781,8 @@ export class ArweaveCompositeClient
       results,
     };
   }
+
+  queueDepth(): number {
+    return this.trustedNodeRequestQueue.length();
+  }
 }

src/metrics.ts

@@ -16,6 +16,8 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 import * as promClient from 'prom-client';
+import { Gauge } from 'prom-client';
+
 /* eslint-disable */
 // @ts-ignore
 import PrometheusMetrics from 'opossum-prometheus';
@@ -197,3 +199,24 @@ export const getDataStreamSuccessesTotal = new promClient.Counter({
   help: 'Count of data stream successes',
   labelNames: ['class'],
 });
+
+// Queue length metrics
+
+const queues: { [key: string]: { length: () => number } } = {};
+export function registerQueueLengthGauge(
+  name: string,
+  queue: { length: () => number },
+) {
+  queues[name] = queue;
+}
+
+export const queueLengthGauge = new Gauge({
+  name: 'queue_length',
+  help: 'Current length of queues',
+  labelNames: ['queue_name'],
+  collect() {
+    Object.entries(queues).forEach(([queueName, queue]) => {
+      this.set({ queue_name: queueName }, queue.length());
+    });
+  },
+});

src/system.ts

@@ -136,6 +136,9 @@ export const arweaveClient = new ArweaveCompositeClient({
     failureRate: config.SIMULATED_REQUEST_FAILURE_RATE,
   }),
 });
+metrics.registerQueueLengthGauge('arweaveClientRequests', {
+  length: () => arweaveClient.queueDepth(),
+});
 
 export const db = new StandaloneSqliteDatabase({
   log,
@@ -255,6 +258,9 @@ export const txFetcher = new TransactionFetcher({
   chainSource: arweaveClient,
   eventEmitter,
 });
+metrics.registerQueueLengthGauge('txFetcher', {
+  length: () => txFetcher.queueDepth(),
+});
 
 // Async fetch block TXs that failed sync fetch
 eventEmitter.on(events.BLOCK_TX_FETCH_FAILED, ({ id: txId }) => {
@@ -266,6 +272,9 @@ const txImporter = new TransactionImporter({
   chainIndex,
   eventEmitter,
 });
+metrics.registerQueueLengthGauge('txImporter', {
+  length: () => txImporter.queueDepth(),
+});
 
 // Queue fetched TXs to
 eventEmitter.addListener(events.TX_FETCHED, (tx: PartialJsonTransaction) => {
@@ -283,6 +292,9 @@ const txOffsetImporter = new TransactionOffsetImporter({
   chainSource: arweaveClient,
   chainOffsetIndex,
 });
+metrics.registerQueueLengthGauge('txOffsetImporter', {
+  length: () => txOffsetImporter.queueDepth(),
+});
 
 export const txOffsetRepairWorker = new TransactionOffsetRepairWorker({
   log,
@@ -364,6 +376,9 @@ const dataContentAttributeImporter = new DataContentAttributeImporter({
   log,
   contiguousDataIndex: contiguousDataIndex,
 });
+metrics.registerQueueLengthGauge('dataContentAttributeImporter', {
+  length: () => dataContentAttributeImporter.queueDepth(),
+});
 
 export const contiguousDataSource = new ReadThroughDataCache({
   log,
@@ -381,13 +396,19 @@ export const dataItemIndexer = new DataItemIndexer({
   eventEmitter,
   indexWriter: dataItemIndexWriter,
 });
+metrics.registerQueueLengthGauge('dataItemIndexer', {
+  length: () => dataItemIndexer.queueDepth(),
+});
 
 const ans104DataIndexer = new Ans104DataIndexer({
   log,
   eventEmitter,
   indexWriter: nestedDataIndexWriter,
   contiguousDataIndex,
 });
+metrics.registerQueueLengthGauge('ans104DataIndexer', {
+  length: () => ans104DataIndexer.queueDepth(),
+});
 
 const shouldUnbundleDataItems = () =>
   ans104DataIndexer.queueDepth() < config.MAX_DATA_ITEM_QUEUE_SIZE &&
@@ -402,13 +423,19 @@ const ans104Unbundler = new Ans104Unbundler({
   workerCount: config.ANS104_UNBUNDLE_WORKERS,
   shouldUnbundle: shouldUnbundleDataItems,
 });
+metrics.registerQueueLengthGauge('ans104Unbundler', {
+  length: () => ans104Unbundler.queueDepth(),
+});
 
 const bundleDataImporter = new BundleDataImporter({
   log,
   contiguousDataSource,
   ans104Unbundler,
   workerCount: config.ANS104_DOWNLOAD_WORKERS,
 });
+metrics.registerQueueLengthGauge('bundleDataImporter', {
+  length: () => bundleDataImporter.queueDepth(),
+});
 
 async function queueBundle(
   item: NormalizedDataItem | PartialJsonTransaction,
@@ -540,6 +567,9 @@ const webhookEmitter = new WebhookEmitter({
   blockFilter: config.WEBHOOK_BLOCK_FILTER,
   log,
 });
+metrics.registerQueueLengthGauge('webhookEmitter', {
+  length: () => webhookEmitter.queueDepth(),
+});
 
 export const mempoolWatcher = config.ENABLE_MEMPOOL_WATCHER
   ? new MempoolWatcher({

src/workers/bundle-data-importer.ts

@@ -134,4 +134,8 @@ export class BundleDataImporter {
     await this.queue.drained();
     log.debug('Stopped successfully.');
   }
+
+  queueDepth(): number {
+    return this.queue.length();
+  }
 }

src/workers/data-content-attribute-importer.ts

@@ -89,4 +89,8 @@ export class DataContentAttributeImporter {
     await this.queue.drained();
     log.debug('Stopped successfully.');
   }
+
+  queueDepth(): number {
+    return this.queue.length();
+  }
 }

src/workers/transaction-fetcher.ts

@@ -121,4 +121,8 @@ export class TransactionFetcher {
     await this.queue.drained();
     log.debug('Stopped successfully.');
   }
+
+  queueDepth(): number {
+    return this.queue.length();
+  }
 }

src/workers/transaction-importer.ts

@@ -78,4 +78,8 @@ export class TransactionImporter {
     await this.queue.drained();
     log.debug('Stopped successfully.');
   }
+
+  queueDepth(): number {
+    return this.queue.length();
+  }
 }

src/workers/transaction-offset-importer.ts

@@ -96,4 +96,8 @@ export class TransactionOffsetImporter {
     await this.queue.drained();
     log.debug('Stopped successfully.');
   }
+
+  queueDepth(): number {
+    return this.queue.length();
+  }
 }

src/workers/webhook-emitter.ts

@@ -244,4 +244,8 @@ export class WebhookEmitter {
       this.log.error('Unexpected error while emitting webhook:', error);
     }
   }
+
+  queueDepth(): number {
+    return this.emissionQueue.length();
+  }
 }