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
2024-03-24 23:02:04 +07:00 · 2024-03-24 23:02:04 +07:00 · c58a70ac8f
parent 1855aaea99
commit c58a70ac8f
7 changed files with 83 additions and 6 deletions
--- a/server/src/app.module.ts
+++ b/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],
--- a/server/src/interfaces/metric.interface.ts
+++ b/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;
+}
--- a/server/src/repositories/metric.repository.ts
+++ b/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);
+  }
+}
--- a/server/src/services/job.service.spec.ts
+++ b/server/src/services/job.service.spec.ts
@ -12,6 +12,7 @@ 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';
@ -19,6 +20,7 @@ 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,6 +39,7 @@ 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();
@ -44,7 +47,8 @@ describe(JobService.name, () => {
    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', () => {
--- a/server/src/services/job.service.ts
+++ b/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 });
        }
      });
    }
--- a/server/src/utils/instrumentation.ts
+++ b/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';
 }
--- a/server/test/repositories/metric.repository.mock.ts
+++ b/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(),
+  };
+};