Add worker thread queue metrics

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 })
+      }
+    })
+  }
+
+}

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) {

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
+    }
+  ]
+}