feat: add metrics to track queue sizes and add operations

ceramicnetwork · Oct 7, 2024 · 4c8ac1b · 4c8ac1b
1 parent 8616ca3
commit 4c8ac1b
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 0 deletions.
diff --git a/packages/cli/src/s3-store.ts b/packages/cli/src/s3-store.ts
@@ -13,6 +13,7 @@ import PQueue from 'p-queue'
 import AWSSDK from 'aws-sdk'
 import { Mutex } from 'await-semaphore'
 import type { DeepNonNullable } from 'ts-essentials'
+import { ServiceMetrics as Metrics } from '@ceramicnetwork/observability'
 
 /**
  * **Remove** `undefined` fields from an S3 Level search params.
@@ -185,6 +186,8 @@ class S3KVStore implements IKVStore {
   }
 
   get(key: string): Promise<any> {
+    Metrics.count(LOAD_S3_QUEUE_ADD, 1)
+    Metrics.observe(LOAD_S3_SIZE, this.#loadingLimit.size)
     return this.#loadingLimit.add(async () => {
       const value = await this.level.get(key)
       return JSON.parse(value)

diff --git a/packages/core/src/ancillary/task-queue.ts b/packages/core/src/ancillary/task-queue.ts
@@ -1,4 +1,8 @@
 import PQueue from 'p-queue'
+import { ServiceMetrics as Metrics } from '@ceramicnetwork/observability'
+
+const TASK_QUEUE_SIZE = 'task_queue_size'
+const TASK_QUEUE_SIZE_PENDING = 'task_queue_size_pending'
 
 export const noop = () => {
   // Do Nothing
@@ -52,6 +56,8 @@ export class TaskQueue implements TaskQueueLike {
    * Size of the queue. Counts both deferred and currently running tasks.
    */
   get size(): number {
+    Metrics.observe(TASK_QUEUE_SIZE, this.#pq.size)
+    Metrics.observe(TASK_QUEUE_SIZE_PENDING, this.#pq.pending)
     return this.#pq.size + this.#pq.pending
   }
 

diff --git a/packages/core/src/state-management/named-task-queue.ts b/packages/core/src/state-management/named-task-queue.ts
@@ -1,4 +1,9 @@
 import { noop, TaskQueue } from '../ancillary/task-queue.js'
+import { ServiceMetrics as Metrics } from '@ceramicnetwork/observability'
+
+const NAMED_TASK_QUEUE_SIZE = 'named_task_queue_size'
+const NAMED_TASK_QUEUE_RUN = 'named_task_queue_run'
+const NAMED_TASK_QUEUE_ADD = 'named_task_queue_add'
 
 /**
  * Set of named PQueues.
@@ -49,6 +54,8 @@ export class NamedTaskQueue {
    */
   run<A>(name: string, task: () => Promise<A>): Promise<A> {
     const queue = this.queue(name)
+    Metrics.observe(NAMED_TASK_QUEUE_SIZE, queue.size, {'name': name})
+    Metrics.count(NAMED_TASK_QUEUE_RUN, 1, {'name': name})
     return queue.run(task).finally(() => {
       this.remove(name)
     })
@@ -62,6 +69,8 @@ export class NamedTaskQueue {
    */
   add(name: string, task: () => Promise<void>): void {
     const queue = this.queue(name)
+    Metrics.observe(NAMED_TASK_QUEUE_SIZE, queue.size, {'name': name})
+    Metrics.count(NAMED_TASK_QUEUE_ADD, 1, {'name': name})
     queue.add(
       () => task(),
       () => this.remove(name)

diff --git a/packages/indexing/src/history-sync/workers/rebuild-anchor.ts b/packages/indexing/src/history-sync/workers/rebuild-anchor.ts
@@ -13,12 +13,15 @@ import type { Worker, Job } from '@ceramicnetwork/job-queue'
 import { CID } from 'multiformats/cid'
 import { pathString } from '@ceramicnetwork/anchor-utils'
 import PQueue from 'p-queue'
+import { ServiceMetrics as Metrics } from '@ceramicnetwork/observability'
 
 // NOTE: In V' history sync will need to be reworked (ex. use CAR files, use recon)
 
 // Up to 1024 streams could be present in an anchor
 const IPFS_LOAD_CONCURRENCY = 16
 
+const REBUILD_ANCHOR_QUEUE_SIZE = 'rebuild_anchor_queue_size'
+
 const REBUILD_ANCHOR_JOB_OPTIONS: SendOptions = {
   retryLimit: 5,
   retryDelay: 60, // 1 minute
@@ -151,6 +154,7 @@ export class RebuildAnchorWorker implements Worker<RebuildAnchorJobData> {
 
     const queue = new PQueue({ concurrency: IPFS_LOAD_CONCURRENCY })
     await queue.addAll(tasks)
+    Metrics.observe(REBUILD_ANCHOR_QUEUE_SIZE, queue.size)
 
     this.logger.debug(
       `Rebuild anchor job completed for models ${jobData.models}, root ${jobData.root}, and txHash ${jobData.txHash}`