feat(server): job metrics (#8255)

* metric repo

* add metric repo

* remove unused import

* formatting

* fix

* try disabling job metrics for e2e

* import otel in test module

server/src/app.module.ts

@@ -51,6 +51,7 @@ import { ILibraryRepository } from 'src/interfaces/library.interface';
 import { IMachineLearningRepository } from 'src/interfaces/machine-learning.interface';
 import { IMediaRepository } from 'src/interfaces/media.interface';
 import { IMetadataRepository } from 'src/interfaces/metadata.interface';
+import { IMetricRepository } from 'src/interfaces/metric.interface';
 import { IMoveRepository } from 'src/interfaces/move.interface';
 import { IPartnerRepository } from 'src/interfaces/partner.interface';
 import { IPersonRepository } from 'src/interfaces/person.interface';
@@ -83,6 +84,7 @@ import { LibraryRepository } from 'src/repositories/library.repository';
 import { MachineLearningRepository } from 'src/repositories/machine-learning.repository';
 import { MediaRepository } from 'src/repositories/media.repository';
 import { MetadataRepository } from 'src/repositories/metadata.repository';
+import { MetricRepository } from 'src/repositories/metric.repository';
 import { MoveRepository } from 'src/repositories/move.repository';
 import { PartnerRepository } from 'src/repositories/partner.repository';
 import { PersonRepository } from 'src/repositories/person.repository';
@@ -163,7 +165,6 @@ const controllers = [
 const services: Provider[] = [
   ApiService,
   MicroservicesService,
-
   APIKeyService,
   ActivityService,
   AlbumService,
@@ -208,6 +209,7 @@ const repositories: Provider[] = [
   { provide: IKeyRepository, useClass: ApiKeyRepository },
   { provide: IMachineLearningRepository, useClass: MachineLearningRepository },
   { provide: IMetadataRepository, useClass: MetadataRepository },
+  { provide: IMetricRepository, useClass: MetricRepository },
   { provide: IMoveRepository, useClass: MoveRepository },
   { provide: IPartnerRepository, useClass: PartnerRepository },
   { provide: IPersonRepository, useClass: PersonRepository },
@@ -277,6 +279,7 @@ export class ImmichAdminModule {}
     EventEmitterModule.forRoot(),
     TypeOrmModule.forRoot(databaseConfig),
     TypeOrmModule.forFeature(databaseEntities),
+    OpenTelemetryModule.forRoot(otelConfig),
   ],
   controllers: [...controllers],
   providers: [...services, ...repositories, ...middleware, SchedulerRegistry],

server/src/interfaces/metric.interface.ts

@@ -0,0 +1,13 @@
+import { MetricOptions } from '@opentelemetry/api';
+
+export interface CustomMetricOptions extends MetricOptions {
+  enabled?: boolean;
+}
+
+export const IMetricRepository = 'IMetricRepository';
+
+export interface IMetricRepository {
+  addToCounter(name: string, value: number, options?: CustomMetricOptions): void;
+  updateGauge(name: string, value: number, options?: CustomMetricOptions): void;
+  updateHistogram(name: string, value: number, options?: CustomMetricOptions): void;
+}

server/src/repositories/metric.repository.ts

@@ -0,0 +1,31 @@
+import { Inject } from '@nestjs/common';
+import { MetricService } from 'nestjs-otel';
+import { CustomMetricOptions, IMetricRepository } from 'src/interfaces/metric.interface';
+
+export class MetricRepository implements IMetricRepository {
+  constructor(@Inject(MetricService) private readonly metricService: MetricService) {}
+
+  addToCounter(name: string, value: number, options?: CustomMetricOptions): void {
+    if (options?.enabled === false) {
+      return;
+    }
+
+    this.metricService.getCounter(name, options).add(value);
+  }
+
+  updateGauge(name: string, value: number, options?: CustomMetricOptions): void {
+    if (options?.enabled === false) {
+      return;
+    }
+
+    this.metricService.getUpDownCounter(name, options).add(value);
+  }
+
+  updateHistogram(name: string, value: number, options?: CustomMetricOptions): void {
+    if (options?.enabled === false) {
+      return;
+    }
+
+    this.metricService.getHistogram(name, options).record(value);
+  }
+}

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

@@ -12,13 +12,15 @@ import {
   JobStatus,
   QueueName,
 } from 'src/interfaces/job.interface';
+import { IMetricRepository } from 'src/interfaces/metric.interface';
 import { IPersonRepository } from 'src/interfaces/person.interface';
 import { ISystemConfigRepository } from 'src/interfaces/system-config.interface';
 import { JobService } from 'src/services/job.service';
 import { assetStub } from 'test/fixtures/asset.stub';
 import { newAssetRepositoryMock } from 'test/repositories/asset.repository.mock';
 import { newEventRepositoryMock } from 'test/repositories/event.repository.mock';
 import { newJobRepositoryMock } from 'test/repositories/job.repository.mock';
+import { newMetricRepositoryMock } from 'test/repositories/metric.repository.mock';
 import { newPersonRepositoryMock } from 'test/repositories/person.repository.mock';
 import { newSystemConfigRepositoryMock } from 'test/repositories/system-config.repository.mock';
 
@@ -37,14 +39,16 @@ describe(JobService.name, () => {
   let eventMock: jest.Mocked<IEventRepository>;
   let jobMock: jest.Mocked<IJobRepository>;
   let personMock: jest.Mocked<IPersonRepository>;
+  let metricMock: jest.Mocked<IMetricRepository>;
 
   beforeEach(() => {
     assetMock = newAssetRepositoryMock();
     configMock = newSystemConfigRepositoryMock();
     eventMock = newEventRepositoryMock();
     jobMock = newJobRepositoryMock();
     personMock = newPersonRepositoryMock();
-    sut = new JobService(assetMock, eventMock, jobMock, configMock, personMock);
+    metricMock = newMetricRepositoryMock();
+    sut = new JobService(assetMock, eventMock, jobMock, configMock, personMock, metricMock);
   });
 
   it('should work', () => {

server/src/services/job.service.ts

@@ -1,4 +1,5 @@
 import { BadRequestException, Inject, Injectable } from '@nestjs/common';
+import { snakeCase } from 'lodash';
 import { FeatureFlag, SystemConfigCore } from 'src/cores/system-config.core';
 import { mapAsset } from 'src/dtos/asset-response.dto';
 import { AllJobStatusResponseDto, JobCommandDto, JobStatusDto } from 'src/dtos/job.dto';
@@ -16,8 +17,10 @@ import {
   QueueCleanType,
   QueueName,
 } from 'src/interfaces/job.interface';
+import { IMetricRepository } from 'src/interfaces/metric.interface';
 import { IPersonRepository } from 'src/interfaces/person.interface';
 import { ISystemConfigRepository } from 'src/interfaces/system-config.interface';
+import { jobMetrics } from 'src/utils/instrumentation';
 import { ImmichLogger } from 'src/utils/logger';
 
 @Injectable()
@@ -31,6 +34,7 @@ export class JobService {
     @Inject(IJobRepository) private jobRepository: IJobRepository,
     @Inject(ISystemConfigRepository) configRepository: ISystemConfigRepository,
     @Inject(IPersonRepository) private personRepository: IPersonRepository,
+    @Inject(IMetricRepository) private metricRepository: IMetricRepository,
   ) {
     this.configCore = SystemConfigCore.create(configRepository);
   }
@@ -92,6 +96,8 @@ export class JobService {
       throw new BadRequestException(`Job is already running`);
     }
 
+    this.metricRepository.addToCounter(`immich.queues.${snakeCase(name)}.started`, 1), { enabled: jobMetrics };
+
     switch (name) {
       case QueueName.VIDEO_CONVERSION: {
         return this.jobRepository.queue({ name: JobName.QUEUE_VIDEO_CONVERSION, data: { force } });
@@ -156,14 +162,21 @@ export class JobService {
       this.jobRepository.addHandler(queueName, concurrency, async (item: JobItem): Promise<void> => {
         const { name, data } = item;
 
+        const queueMetric = `immich.queues.${snakeCase(queueName)}.active`;
+        this.metricRepository.updateGauge(queueMetric, 1, { enabled: jobMetrics });
+
         try {
           const handler = jobHandlers[name];
           const status = await handler(data);
+          const jobMetric = `immich.jobs.${name.replaceAll('-', '_')}.${status}`;
+          this.metricRepository.addToCounter(jobMetric, 1, { enabled: jobMetrics });
           if (status === JobStatus.SUCCESS || status == JobStatus.SKIPPED) {
             await this.onDone(item);
           }
         } catch (error: Error | any) {
           this.logger.error(`Unable to run job handler (${queueName}/${name}): ${error}`, error?.stack, data);
+        } finally {
+          this.metricRepository.updateGauge(queueMetric, -1, { enabled: jobMetrics });
         }
       });
     }

server/src/utils/instrumentation.ts

@@ -17,12 +17,16 @@ import { excludePaths, serverVersion } from 'src/constants';
 import { DecorateAll } from 'src/decorators';
 
 let metricsEnabled = process.env.IMMICH_METRICS === 'true';
-const hostMetrics =
+export const hostMetrics =
   process.env.IMMICH_HOST_METRICS == null ? metricsEnabled : process.env.IMMICH_HOST_METRICS === 'true';
-const apiMetrics = process.env.IMMICH_API_METRICS == null ? metricsEnabled : process.env.IMMICH_API_METRICS === 'true';
-const repoMetrics = process.env.IMMICH_IO_METRICS == null ? metricsEnabled : process.env.IMMICH_IO_METRICS === 'true';
+export const apiMetrics =
+  process.env.IMMICH_API_METRICS == null ? metricsEnabled : process.env.IMMICH_API_METRICS === 'true';
+export const repoMetrics =
+  process.env.IMMICH_IO_METRICS == null ? metricsEnabled : process.env.IMMICH_IO_METRICS === 'true';
+export const jobMetrics =
+  process.env.IMMICH_JOB_METRICS == null ? metricsEnabled : process.env.IMMICH_JOB_METRICS === 'true';
 
-metricsEnabled ||= hostMetrics || apiMetrics || repoMetrics;
+metricsEnabled ||= hostMetrics || apiMetrics || repoMetrics || jobMetrics;
 if (!metricsEnabled && process.env.OTEL_SDK_DISABLED === undefined) {
   process.env.OTEL_SDK_DISABLED = 'true';
 }

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

@@ -0,0 +1,9 @@
+import { IMetricRepository } from 'src/interfaces/metric.interface';
+
+export const newMetricRepositoryMock = (): jest.Mocked<IMetricRepository> => {
+  return {
+    addToCounter: jest.fn(),
+    updateGauge: jest.fn(),
+    updateHistogram: jest.fn(),
+  };
+};