feat(server): add queued job metrics to telemetry

Adds a new observable gauge `immich.queues.<queue_name>.queued` to track the number of jobs in waiting, paused, or delayed states.

- Polls queue statistics every 5 seconds.
- Fetches queue counts in parallel.
- Only runs polling loop if job telemetry is enabled.
- Includes debug logging for failed metric updates.

Author: dangarmol

server/src/repositories/telemetry.repository.ts

@@ -24,6 +24,7 @@ type MetricGroupOptions = { enabled: boolean };
 
 export class MetricGroupRepository {
   private enabled = false;
+  private observableGauges = new Map<string, () => number>();
 
   constructor(private metricService: MetricService) {}
 
@@ -45,6 +46,15 @@ export class MetricGroupRepository {
     }
   }
 
+  setObservableGauge(name: string, valueCallback: () => number, options?: MetricOptions): void {
+    if (this.enabled && !this.observableGauges.has(name)) {
+      this.observableGauges.set(name, valueCallback);
+      this.metricService.getObservableGauge(name, options).addCallback((observableResult) => {
+        observableResult.observe(valueCallback());
+      });
+    }
+  }
+
   configure(options: MetricGroupOptions): this {
     this.enabled = options.enabled;
     return this;

server/src/services/telemetry.service.spec.ts

@@ -0,0 +1,53 @@
+import { ImmichTelemetry, QueueName } from 'src/enum';
+import { TelemetryService } from 'src/services/telemetry.service';
+import { newTestService, ServiceMocks } from 'test/utils';
+
+describe(TelemetryService.name, () => {
+  let sut: TelemetryService;
+  let mocks: ServiceMocks;
+
+  beforeEach(() => {
+    ({ sut, mocks } = newTestService(TelemetryService));
+  });
+
+  it('should work', () => {
+    expect(sut).toBeDefined();
+  });
+
+  describe('onBootstrap', () => {
+    it('should register queued metrics if enabled', async () => {
+      mocks.config.getEnv.mockReturnValue({
+        telemetry: {
+          metrics: new Set([ImmichTelemetry.Job]),
+        },
+      } as any);
+
+      mocks.job.getJobCounts.mockResolvedValue({
+        waiting: 1,
+        paused: 2,
+        delayed: 3,
+        active: 0,
+        completed: 0,
+        failed: 0,
+      });
+
+      await sut.onBootstrap();
+
+      expect(mocks.telemetry.jobs.setObservableGauge).toHaveBeenCalledTimes(Object.keys(QueueName).length);
+      expect(mocks.job.getJobCounts).toHaveBeenCalledTimes(Object.keys(QueueName).length);
+    });
+
+    it('should not register queued metrics if disabled', async () => {
+      mocks.config.getEnv.mockReturnValue({
+        telemetry: {
+          metrics: new Set([]),
+        },
+      } as any);
+
+      await sut.onBootstrap();
+
+      expect(mocks.telemetry.jobs.setObservableGauge).not.toHaveBeenCalled();
+      expect(mocks.job.getJobCounts).not.toHaveBeenCalled();
+    });
+  });
+});

server/src/services/telemetry.service.ts

@@ -1,14 +1,62 @@
 import { snakeCase } from 'lodash';
 import { OnEvent } from 'src/decorators';
-import { ImmichWorker, JobStatus } from 'src/enum';
+import { ImmichTelemetry, ImmichWorker, JobStatus, QueueName } from 'src/enum';
 import { ArgOf, ArgsOf } from 'src/repositories/event.repository';
 import { BaseService } from 'src/services/base.service';
 
+const QUEUE_METRICS_POLLING_INTERVAL = 5000;
+
 export class TelemetryService extends BaseService {
+  private queueCounts = new Map<string, number>();
+  private pollingInterval?: NodeJS.Timeout;
+
   @OnEvent({ name: 'AppBootstrap', workers: [ImmichWorker.Api] })
   async onBootstrap(): Promise<void> {
     const userCount = await this.userRepository.getCount();
     this.telemetryRepository.api.addToGauge('immich.users.total', userCount);
+
+    const { telemetry } = this.configRepository.getEnv();
+    if (telemetry.metrics.has(ImmichTelemetry.Job)) {
+      // Register observable gauges for queued metrics
+      this.registerQueuedMetrics();
+
+      // Start polling queue statistics
+      await this.updateQueuedMetrics();
+      this.pollingInterval = setInterval(() => {
+        void this.updateQueuedMetrics();
+      }, QUEUE_METRICS_POLLING_INTERVAL);
+    }
+  }
+
+  @OnEvent({ name: 'AppShutdown' })
+  onShutdown(): void {
+    if (this.pollingInterval) {
+      clearInterval(this.pollingInterval);
+    }
+  }
+
+  private registerQueuedMetrics(): void {
+    for (const queueName of Object.values(QueueName)) {
+      const metricName = `immich.queues.${snakeCase(queueName)}.queued`;
+      this.telemetryRepository.jobs.setObservableGauge(metricName, () => this.queueCounts.get(metricName) ?? 0, {
+        description: `Number of queued jobs in ${queueName} queue (waiting + paused + delayed)`,
+      });
+    }
+  }
+
+  private async updateQueuedMetrics(): Promise<void> {
+    await Promise.all(
+      Object.values(QueueName).map(async (queueName) => {
+        try {
+          const stats = await this.jobRepository.getJobCounts(queueName);
+          const queuedCount = stats.waiting + stats.paused + stats.delayed;
+          const metricName = `immich.queues.${snakeCase(queueName)}.queued`;
+          this.queueCounts.set(metricName, queuedCount);
+        } catch (error) {
+          this.logger.debug(`Failed to update queued metrics for ${queueName}: ${error}`);
+        }
+      }),
+    );
   }
 
   @OnEvent({ name: 'UserCreate' })

server/test/repositories/telemetry.repository.mock.ts

@@ -7,6 +7,7 @@ const newMetricGroupMock = () => {
     addToCounter: vitest.fn(),
     addToGauge: vitest.fn(),
     addToHistogram: vitest.fn(),
+    setObservableGauge: vitest.fn(),
     configure: vitest.fn(),
   };
 };