From a2d5c2d44f12d2bda4d8e50ea6d2b68bbb461dbf Mon Sep 17 00:00:00 2001
From: Chocobozzz <me@florianbigard.com>
Date: Thu, 26 Oct 2023 16:34:54 +0200
Subject: [PATCH] Add worker thread queue metrics

---
 .../worker-threads-observers.ts               | 22 +++++++++++
 server/core/lib/opentelemetry/metrics.ts      | 25 ++++--------
 server/core/lib/worker/parent-process.ts      | 39 +++++++++++++++++++
 3 files changed, 69 insertions(+), 17 deletions(-)
 create mode 100644 server/core/lib/opentelemetry/metric-helpers/worker-threads-observers.ts

diff --git a/server/core/lib/opentelemetry/metric-helpers/worker-threads-observers.ts b/server/core/lib/opentelemetry/metric-helpers/worker-threads-observers.ts
new file mode 100644
index 000000000..608010cef
--- /dev/null
+++ b/server/core/lib/opentelemetry/metric-helpers/worker-threads-observers.ts
@@ -0,0 +1,22 @@
+import { Meter } from '@opentelemetry/api'
+import { getWorkersQueueSize } from '@server/lib/worker/parent-process.js'
+
+export class WorkerThreadsObserversBuilder {
+
+  constructor (private readonly meter: Meter) {
+
+  }
+
+  buildObservers () {
+    this.meter.createObservableGauge('peertube_worker_thread_queue_total', {
+      description: 'Total tasks waiting for a PeerTube worker thread'
+    }).addCallback(observableResult => {
+      const stats = getWorkersQueueSize()
+
+      for (const stat of stats) {
+        observableResult.observe(stat.queueSize, { state: 'waiting', workerThread: stat.label })
+      }
+    })
+  }
+
+}
diff --git a/server/core/lib/opentelemetry/metrics.ts b/server/core/lib/opentelemetry/metrics.ts
index 7276182ef..9d70fbe19 100644
--- a/server/core/lib/opentelemetry/metrics.ts
+++ b/server/core/lib/opentelemetry/metrics.ts
@@ -15,6 +15,7 @@ import {
   StatsObserversBuilder,
   ViewersObserversBuilder
 } from './metric-helpers/index.js'
+import { WorkerThreadsObserversBuilder } from './metric-helpers/worker-threads-observers.js'
 
 class OpenTelemetryMetrics {
 
@@ -69,23 +70,13 @@ class OpenTelemetryMetrics {
     this.playbackMetrics = new PlaybackMetrics(this.meter)
     this.playbackMetrics.buildCounters()
 
-    const nodeJSObserversBuilder = new NodeJSObserversBuilder(this.meter)
-    nodeJSObserversBuilder.buildObservers()
-
-    const jobQueueObserversBuilder = new JobQueueObserversBuilder(this.meter)
-    jobQueueObserversBuilder.buildObservers()
-
-    const statsObserversBuilder = new StatsObserversBuilder(this.meter)
-    statsObserversBuilder.buildObservers()
-
-    const livesObserversBuilder = new LivesObserversBuilder(this.meter)
-    livesObserversBuilder.buildObservers()
-
-    const viewersObserversBuilder = new ViewersObserversBuilder(this.meter)
-    viewersObserversBuilder.buildObservers()
-
-    const bittorrentTrackerObserversBuilder = new BittorrentTrackerObserversBuilder(this.meter, options.trackerServer)
-    bittorrentTrackerObserversBuilder.buildObservers()
+    new NodeJSObserversBuilder(this.meter).buildObservers()
+    new JobQueueObserversBuilder(this.meter).buildObservers()
+    new StatsObserversBuilder(this.meter).buildObservers()
+    new LivesObserversBuilder(this.meter).buildObservers()
+    new ViewersObserversBuilder(this.meter).buildObservers()
+    new WorkerThreadsObserversBuilder(this.meter).buildObservers()
+    new BittorrentTrackerObserversBuilder(this.meter, options.trackerServer).buildObservers()
   }
 
   observePlaybackMetric (video: MVideoImmutable, metrics: PlaybackMetricCreate) {
diff --git a/server/core/lib/worker/parent-process.ts b/server/core/lib/worker/parent-process.ts
index 64d264202..a4bbed06d 100644
--- a/server/core/lib/worker/parent-process.ts
+++ b/server/core/lib/worker/parent-process.ts
@@ -146,3 +146,42 @@ export function buildDigestFromWorker (
 
   return buildDigestWorker.run(options)
 }
+
+// ---------------------------------------------------------------------------
+
+export function getWorkersQueueSize () {
+  return [
+    {
+      label: 'downloadImage',
+      queueSize: downloadImageWorker?.queueSize || 0
+    },
+    {
+      label: 'processImageWorker',
+      queueSize: processImageWorker?.queueSize || 0
+    },
+    {
+      label: 'getImageSizeWorker',
+      queueSize: getImageSizeWorker?.queueSize || 0
+    },
+    {
+      label: 'parallelHTTPBroadcastWorker',
+      queueSize: parallelHTTPBroadcastWorker?.queueSize || 0
+    },
+    {
+      label: 'sequentialHTTPBroadcastWorker',
+      queueSize: sequentialHTTPBroadcastWorker?.queueSize || 0
+    },
+    {
+      label: 'httpUnicastWorker',
+      queueSize: httpUnicastWorker?.queueSize || 0
+    },
+    {
+      label: 'signJsonLDObjectWorker',
+      queueSize: signJsonLDObjectWorker?.queueSize || 0
+    },
+    {
+      label: 'buildDigestWorker',
+      queueSize: buildDigestWorker?.queueSize || 0
+    }
+  ]
+}