server/lib/schedulers/runner-job-watch-dog-scheduler.ts

   1 import { CONFIG } from '@server/initializers/config'
   2 import { RunnerJobModel } from '@server/models/runner/runner-job'
   3 import { logger, loggerTagsFactory } from '../../helpers/logger'
   4 import { SCHEDULER_INTERVALS_MS } from '../../initializers/constants'
   5 import { getRunnerJobHandlerClass } from '../runners'
   6 import { AbstractScheduler } from './abstract-scheduler'
   7
   8 const lTags = loggerTagsFactory('runner')
   9
  10 export class RunnerJobWatchDogScheduler extends AbstractScheduler {
  11
  12   private static instance: AbstractScheduler
  13
  14   protected schedulerIntervalMs = SCHEDULER_INTERVALS_MS.RUNNER_JOB_WATCH_DOG
  15
  16   private constructor () {
  17     super()
  18   }
  19
  20   protected async internalExecute () {
  21     const vodStalledJobs = await RunnerJobModel.listStalledJobs({
  22       staleTimeMS: CONFIG.REMOTE_RUNNERS.STALLED_JOBS.VOD,
  23       types: [ 'vod-audio-merge-transcoding', 'vod-hls-transcoding', 'vod-web-video-transcoding' ]
  24     })
  25
  26     const liveStalledJobs = await RunnerJobModel.listStalledJobs({
  27       staleTimeMS: CONFIG.REMOTE_RUNNERS.STALLED_JOBS.LIVE,
  28       types: [ 'live-rtmp-hls-transcoding' ]
  29     })
  30
  31     for (const stalled of [ ...vodStalledJobs, ...liveStalledJobs ]) {
  32       logger.info('Abort stalled runner job %s (%s)', stalled.uuid, stalled.type, lTags(stalled.uuid, stalled.type))
  33
  34       const Handler = getRunnerJobHandlerClass(stalled)
  35       await new Handler().abort({ runnerJob: stalled })
  36     }
  37   }
  38
  39   static get Instance () {
  40     return this.instance || (this.instance = new this())
  41   }
  42 }