server/lib/opentelemetry/metrics.ts

   1 import { Application, Request, Response } from 'express'
   2 import { Meter, metrics } from '@opentelemetry/api-metrics'
   3 import { PrometheusExporter } from '@opentelemetry/exporter-prometheus'
   4 import { MeterProvider } from '@opentelemetry/sdk-metrics-base'
   5 import { logger } from '@server/helpers/logger'
   6 import { CONFIG } from '@server/initializers/config'
   7 import { JobQueue } from '../job-queue'
   8 import { StatsObserverBuilder } from './metric-helpers'
   9
  10 class OpenTelemetryMetrics {
  11
  12   private static instance: OpenTelemetryMetrics
  13
  14   private meter: Meter
  15
  16   private onRequestDuration: (req: Request, res: Response) => void
  17
  18   private constructor () {}
  19
  20   init (app: Application) {
  21     if (CONFIG.OPEN_TELEMETRY.METRICS.ENABLED !== true) return
  22
  23     app.use((req, res, next) => {
  24       res.once('finish', () => {
  25         if (!this.onRequestDuration) return
  26
  27         this.onRequestDuration(req as Request, res as Response)
  28       })
  29
  30       next()
  31     })
  32   }
  33
  34   registerMetrics () {
  35     if (CONFIG.OPEN_TELEMETRY.METRICS.ENABLED !== true) return
  36
  37     logger.info('Registering Open Telemetry metrics')
  38
  39     const provider = new MeterProvider()
  40
  41     provider.addMetricReader(new PrometheusExporter({ port: CONFIG.OPEN_TELEMETRY.METRICS.PROMETHEUS_EXPORTER.PORT }))
  42
  43     metrics.setGlobalMeterProvider(provider)
  44
  45     this.meter = metrics.getMeter('default')
  46
  47     this.buildMemoryObserver()
  48     this.buildRequestObserver()
  49     this.buildJobQueueObserver()
  50
  51     const statsObserverBuilder = new StatsObserverBuilder(this.meter)
  52     statsObserverBuilder.buildObservers()
  53   }
  54
  55   private buildMemoryObserver () {
  56     this.meter.createObservableGauge('nodejs_memory_usage_bytes', {
  57       description: 'Memory'
  58     }).addCallback(observableResult => {
  59       const current = process.memoryUsage()
  60
  61       observableResult.observe(current.heapTotal, { memoryType: 'heapTotal' })
  62       observableResult.observe(current.heapUsed, { memoryType: 'heapUsed' })
  63       observableResult.observe(current.arrayBuffers, { memoryType: 'arrayBuffers' })
  64       observableResult.observe(current.external, { memoryType: 'external' })
  65       observableResult.observe(current.rss, { memoryType: 'rss' })
  66     })
  67   }
  68
  69   private buildJobQueueObserver () {
  70     this.meter.createObservableGauge('peertube_job_queue_total', {
  71       description: 'Total jobs in the PeerTube job queue'
  72     }).addCallback(async observableResult => {
  73       const stats = await JobQueue.Instance.getStats()
  74
  75       for (const { jobType, counts } of stats) {
  76         for (const state of Object.keys(counts)) {
  77           observableResult.observe(counts[state], { jobType, state })
  78         }
  79       }
  80     })
  81   }
  82
  83   private buildRequestObserver () {
  84     const requestDuration = this.meter.createHistogram('http_request_duration_ms', {
  85       unit: 'milliseconds',
  86       description: 'Duration of HTTP requests in ms'
  87     })
  88
  89     this.onRequestDuration = (req: Request, res: Response) => {
  90       const duration = Date.now() - res.locals.requestStart
  91
  92       requestDuration.record(duration, {
  93         path: this.buildRequestPath(req.originalUrl),
  94         method: req.method,
  95         statusCode: res.statusCode + ''
  96       })
  97     }
  98   }
  99
 100   private buildRequestPath (path: string) {
 101     return path.split('?')[0]
 102   }
 103
 104   static get Instance () {
 105     return this.instance || (this.instance = new this())
 106   }
 107 }
 108
 109 export {
 110   OpenTelemetryMetrics
 111 }