Merge pull request #193 from wormhole-foundation/processing-metrics-f…

…ixes

remove old metrics + add receivedAt timestamp to RelayJob

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@wormhole-foundation/relayer-engine",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "description": "Relayer Engine",
   "main": "lib/index.js",
   "types": "lib/index.d.ts",

relayer/application.ts

@@ -309,6 +309,8 @@ export class RelayerApp<ContextT extends Context> extends EventEmitter {
     }
     if (this.storage) {
       const job = await this.storage.addVaaToQueue(parsedVaa.bytes);
+      // TODO: it would be ideal to only emit the added event only in
+      // the cases the job was actually added (not already in queue)
       this.emit(RelayerEvents.Added, parsedVaa, job);
     } else {
       this.emit(RelayerEvents.Added, parsedVaa);

relayer/storage.metrics.ts

@@ -5,11 +5,6 @@ export interface StorageMetrics {
   waitingGauge: Gauge<string>;
   activeGauge: Gauge<string>;
   failedGauge: Gauge<string>;
-  completedCounter: Counter<string>;
-  failedWithMaxRetriesCounter: Counter<string>;
-  failedRunsCounter: Counter<string>;
-  completedDuration: Histogram<string>;
-  processedDuration: Histogram<string>;
 }
 export function createStorageMetrics(
   storageRegistry: Registry = new Registry(),
@@ -41,40 +36,6 @@ export function createStorageMetrics(
         labelNames: ["queue"],
         registers: [storageRegistry],
       }),
-      completedCounter: new Counter({
-        name: `completed_workflows_total`,
-        help: "Total number of completed jobs",
-        labelNames: ["queue"],
-        registers: [storageRegistry],
-      }),
-      failedRunsCounter: new Counter({
-        name: `failed_workflow_runs_total`,
-        help: "Total number of failed job runs",
-        labelNames: ["queue"],
-        registers: [storageRegistry],
-      }),
-      failedWithMaxRetriesCounter: new Counter({
-        name: `failed_with_max_retries_workflows_total`,
-        help: "Total number of jobs that failed after max retries. Eg: they will require manual intervention to succeed",
-        labelNames: ["queue"],
-        registers: [storageRegistry],
-      }),
-      processedDuration: new Histogram({
-        name: `worklow_processing_duration`,
-        help: "Processing time in ms for completed jobs (processing until completed)",
-        buckets: [6000, 7000, 7500, 8000, 8500, 9000, 10000, 12000],
-        labelNames: ["queue"],
-        registers: [storageRegistry],
-      }),
-      completedDuration: new Histogram({
-        name: `workflow_total_duration`,
-        help: "Completion time in ms for jobs (created until completed)",
-        buckets: [
-          7500, 10000, 20000, 30000, 45000, 60000, 90000, 120000, 240000,
-        ],
-        labelNames: ["queue"],
-        registers: [storageRegistry],
-      }),
     },
   };
 }

relayer/storage/redis-storage.ts

@@ -238,6 +238,7 @@ export class RedisStorage implements Storage {
           id: job.id,
           maxAttempts: this.opts.attempts,
           name: job.name,
+          receivedAt: job.timestamp,
           log: job.log.bind(job),
           updateProgress: job.updateProgress.bind(job),
         };
@@ -254,8 +255,6 @@ export class RedisStorage implements Storage {
     );
     this.workerId = this.worker.id;
 
-    this.worker.on("completed", this.onCompleted.bind(this));
-    this.worker.on("failed", this.onFailed.bind(this));
     this.spawnGaugeUpdateWorker();
   }
 
@@ -284,29 +283,6 @@ export class RedisStorage implements Storage {
     this.metrics.failedGauge.labels({ queue: this.vaaQueue.name }).set(failed);
   }
 
-  private async onCompleted(job: Job) {
-    const completedDuration = job.finishedOn! - job.timestamp!; // neither can be null
-    const processedDuration = job.finishedOn! - job.processedOn!; // neither can be null
-    this.metrics.completedCounter.labels({ queue: this.vaaQueue.name }).inc();
-    this.metrics.completedDuration
-      .labels({ queue: this.vaaQueue.name })
-      .observe(completedDuration);
-    this.metrics.processedDuration
-      .labels({ queue: this.vaaQueue.name })
-      .observe(processedDuration);
-  }
-
-  private async onFailed(job: Job) {
-    // TODO: Add a failed duration metric for processing time for failed jobs
-    this.metrics.failedRunsCounter.labels({ queue: this.vaaQueue.name }).inc();
-
-    if (job.attemptsMade === this.opts.attempts) {
-      this.metrics.failedWithMaxRetriesCounter
-        .labels({ queue: this.vaaQueue.name })
-        .inc();
-    }
-  }
-
   storageKoaUI(path: string) {
     // UI
     const serverAdapter = new KoaAdapter();

relayer/storage/storage.ts

@@ -16,7 +16,7 @@ export interface RelayJob {
   };
   attempts: number;
   maxAttempts: number;
-  receivedAt?: number;
+  receivedAt: number;
   log(logRow: string): Promise<number>;
   updateProgress(progress: number | object): Promise<void>;
 }