Implement remote runner jobs in server

Move ffmpeg functions to @shared

5 files changed, 648 insertions, 0 deletions

diff --git a/server/controllers/api/runners/index.ts b/server/controllers/api/runners/index.ts
new file mode 100644
index 000000000..c98ded354
--- /dev/null
+++ b/server/controllers/api/runners/index.ts
@@ -0,0 +1,18 @@
+import express from 'express'
+import { runnerJobsRouter } from './jobs'
+import { runnerJobFilesRouter } from './jobs-files'
+import { manageRunnersRouter } from './manage-runners'
+import { runnerRegistrationTokensRouter } from './registration-tokens'
+
+const runnersRouter = express.Router()
+
+runnersRouter.use('/', manageRunnersRouter)
+runnersRouter.use('/', runnerJobsRouter)
+runnersRouter.use('/', runnerJobFilesRouter)
+runnersRouter.use('/', runnerRegistrationTokensRouter)
+
+// ---------------------------------------------------------------------------
+
+export {
+  runnersRouter
+}
diff --git a/server/controllers/api/runners/jobs-files.ts b/server/controllers/api/runners/jobs-files.ts
new file mode 100644
index 000000000..e43ce35f5
--- /dev/null
+++ b/server/controllers/api/runners/jobs-files.ts
@@ -0,0 +1,84 @@
+import express from 'express'
+import { logger, loggerTagsFactory } from '@server/helpers/logger'
+import { proxifyHLS, proxifyWebTorrentFile } from '@server/lib/object-storage'
+import { VideoPathManager } from '@server/lib/video-path-manager'
+import { asyncMiddleware } from '@server/middlewares'
+import { jobOfRunnerGetValidator } from '@server/middlewares/validators/runners'
+import { runnerJobGetVideoTranscodingFileValidator } from '@server/middlewares/validators/runners/job-files'
+import { VideoStorage } from '@shared/models'
+
+const lTags = loggerTagsFactory('api', 'runner')
+
+const runnerJobFilesRouter = express.Router()
+
+runnerJobFilesRouter.post('/jobs/:jobUUID/files/videos/:videoId/max-quality',
+  asyncMiddleware(jobOfRunnerGetValidator),
+  asyncMiddleware(runnerJobGetVideoTranscodingFileValidator),
+  asyncMiddleware(getMaxQualityVideoFile)
+)
+
+runnerJobFilesRouter.post('/jobs/:jobUUID/files/videos/:videoId/previews/max-quality',
+  asyncMiddleware(jobOfRunnerGetValidator),
+  asyncMiddleware(runnerJobGetVideoTranscodingFileValidator),
+  getMaxQualityVideoPreview
+)
+
+// ---------------------------------------------------------------------------
+
+export {
+  runnerJobFilesRouter
+}
+
+// ---------------------------------------------------------------------------
+
+async function getMaxQualityVideoFile (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+  const runner = runnerJob.Runner
+  const video = res.locals.videoAll
+
+  logger.info(
+    'Get max quality file of video %s of job %s for runner %s', video.uuid, runnerJob.uuid, runner.name,
+    lTags(runner.name, runnerJob.id, runnerJob.type)
+  )
+
+  const file = video.getMaxQualityFile()
+
+  if (file.storage === VideoStorage.OBJECT_STORAGE) {
+    if (file.isHLS()) {
+      return proxifyHLS({
+        req,
+        res,
+        filename: file.filename,
+        playlist: video.getHLSPlaylist(),
+        reinjectVideoFileToken: false,
+        video
+      })
+    }
+
+    // Web video
+    return proxifyWebTorrentFile({
+      req,
+      res,
+      filename: file.filename
+    })
+  }
+
+  return VideoPathManager.Instance.makeAvailableVideoFile(file, videoPath => {
+    return res.sendFile(videoPath)
+  })
+}
+
+function getMaxQualityVideoPreview (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+  const runner = runnerJob.Runner
+  const video = res.locals.videoAll
+
+  logger.info(
+    'Get max quality preview file of video %s of job %s for runner %s', video.uuid, runnerJob.uuid, runner.name,
+    lTags(runner.name, runnerJob.id, runnerJob.type)
+  )
+
+  const file = video.getPreview()
+
+  return res.sendFile(file.getPath())
+}
diff --git a/server/controllers/api/runners/jobs.ts b/server/controllers/api/runners/jobs.ts
new file mode 100644
index 000000000..7d488ec11
--- /dev/null
+++ b/server/controllers/api/runners/jobs.ts
@@ -0,0 +1,352 @@
+import express, { UploadFiles } from 'express'
+import { createReqFiles } from '@server/helpers/express-utils'
+import { logger, loggerTagsFactory } from '@server/helpers/logger'
+import { generateRunnerJobToken } from '@server/helpers/token-generator'
+import { MIMETYPES } from '@server/initializers/constants'
+import { sequelizeTypescript } from '@server/initializers/database'
+import { getRunnerJobHandlerClass, updateLastRunnerContact } from '@server/lib/runners'
+import {
+  asyncMiddleware,
+  authenticate,
+  ensureUserHasRight,
+  paginationValidator,
+  runnerJobsSortValidator,
+  setDefaultPagination,
+  setDefaultSort
+} from '@server/middlewares'
+import {
+  abortRunnerJobValidator,
+  acceptRunnerJobValidator,
+  errorRunnerJobValidator,
+  getRunnerFromTokenValidator,
+  jobOfRunnerGetValidator,
+  runnerJobGetValidator,
+  successRunnerJobValidator,
+  updateRunnerJobValidator
+} from '@server/middlewares/validators/runners'
+import { RunnerModel } from '@server/models/runner/runner'
+import { RunnerJobModel } from '@server/models/runner/runner-job'
+import {
+  AbortRunnerJobBody,
+  AcceptRunnerJobResult,
+  ErrorRunnerJobBody,
+  HttpStatusCode,
+  ListRunnerJobsQuery,
+  LiveRTMPHLSTranscodingUpdatePayload,
+  RequestRunnerJobResult,
+  RunnerJobState,
+  RunnerJobSuccessBody,
+  RunnerJobSuccessPayload,
+  RunnerJobType,
+  RunnerJobUpdateBody,
+  RunnerJobUpdatePayload,
+  UserRight,
+  VODAudioMergeTranscodingSuccess,
+  VODHLSTranscodingSuccess,
+  VODWebVideoTranscodingSuccess
+} from '@shared/models'
+
+const postRunnerJobSuccessVideoFiles = createReqFiles(
+  [ 'payload[videoFile]', 'payload[resolutionPlaylistFile]' ],
+  { ...MIMETYPES.VIDEO.MIMETYPE_EXT, ...MIMETYPES.M3U8.MIMETYPE_EXT }
+)
+
+const runnerJobUpdateVideoFiles = createReqFiles(
+  [ 'payload[videoChunkFile]', 'payload[resolutionPlaylistFile]', 'payload[masterPlaylistFile]' ],
+  { ...MIMETYPES.VIDEO.MIMETYPE_EXT, ...MIMETYPES.M3U8.MIMETYPE_EXT }
+)
+
+const lTags = loggerTagsFactory('api', 'runner')
+
+const runnerJobsRouter = express.Router()
+
+// ---------------------------------------------------------------------------
+// Controllers for runners
+// ---------------------------------------------------------------------------
+
+runnerJobsRouter.post('/jobs/request',
+  asyncMiddleware(getRunnerFromTokenValidator),
+  asyncMiddleware(requestRunnerJob)
+)
+
+runnerJobsRouter.post('/jobs/:jobUUID/accept',
+  asyncMiddleware(runnerJobGetValidator),
+  acceptRunnerJobValidator,
+  asyncMiddleware(getRunnerFromTokenValidator),
+  asyncMiddleware(acceptRunnerJob)
+)
+
+runnerJobsRouter.post('/jobs/:jobUUID/abort',
+  asyncMiddleware(jobOfRunnerGetValidator),
+  abortRunnerJobValidator,
+  asyncMiddleware(abortRunnerJob)
+)
+
+runnerJobsRouter.post('/jobs/:jobUUID/update',
+  runnerJobUpdateVideoFiles,
+  asyncMiddleware(jobOfRunnerGetValidator),
+  updateRunnerJobValidator,
+  asyncMiddleware(updateRunnerJobController)
+)
+
+runnerJobsRouter.post('/jobs/:jobUUID/error',
+  asyncMiddleware(jobOfRunnerGetValidator),
+  errorRunnerJobValidator,
+  asyncMiddleware(errorRunnerJob)
+)
+
+runnerJobsRouter.post('/jobs/:jobUUID/success',
+  postRunnerJobSuccessVideoFiles,
+  asyncMiddleware(jobOfRunnerGetValidator),
+  successRunnerJobValidator,
+  asyncMiddleware(postRunnerJobSuccess)
+)
+
+// ---------------------------------------------------------------------------
+// Controllers for admins
+// ---------------------------------------------------------------------------
+
+runnerJobsRouter.post('/jobs/:jobUUID/cancel',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  asyncMiddleware(runnerJobGetValidator),
+  asyncMiddleware(cancelRunnerJob)
+)
+
+runnerJobsRouter.get('/jobs',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  paginationValidator,
+  runnerJobsSortValidator,
+  setDefaultSort,
+  setDefaultPagination,
+  asyncMiddleware(listRunnerJobs)
+)
+
+// ---------------------------------------------------------------------------
+
+export {
+  runnerJobsRouter
+}
+
+// ---------------------------------------------------------------------------
+
+// ---------------------------------------------------------------------------
+// Controllers for runners
+// ---------------------------------------------------------------------------
+
+async function requestRunnerJob (req: express.Request, res: express.Response) {
+  const runner = res.locals.runner
+  const availableJobs = await RunnerJobModel.listAvailableJobs()
+
+  logger.debug('Runner %s requests for a job.', runner.name, { availableJobs, ...lTags(runner.name) })
+
+  const result: RequestRunnerJobResult = {
+    availableJobs: availableJobs.map(j => ({
+      uuid: j.uuid,
+      type: j.type,
+      payload: j.payload
+    }))
+  }
+
+  updateLastRunnerContact(req, runner)
+
+  return res.json(result)
+}
+
+async function acceptRunnerJob (req: express.Request, res: express.Response) {
+  const runner = res.locals.runner
+  const runnerJob = res.locals.runnerJob
+
+  runnerJob.state = RunnerJobState.PROCESSING
+  runnerJob.processingJobToken = generateRunnerJobToken()
+  runnerJob.startedAt = new Date()
+  runnerJob.runnerId = runner.id
+
+  const newRunnerJob = await sequelizeTypescript.transaction(transaction => {
+    return runnerJob.save({ transaction })
+  })
+  newRunnerJob.Runner = runner as RunnerModel
+
+  const result: AcceptRunnerJobResult = {
+    job: {
+      ...newRunnerJob.toFormattedJSON(),
+
+      jobToken: newRunnerJob.processingJobToken
+    }
+  }
+
+  updateLastRunnerContact(req, runner)
+
+  logger.info(
+    'Remote runner %s has accepted job %s (%s)', runner.name, runnerJob.uuid, runnerJob.type,
+    lTags(runner.name, runnerJob.uuid, runnerJob.type)
+  )
+
+  return res.json(result)
+}
+
+async function abortRunnerJob (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+  const runner = runnerJob.Runner
+  const body: AbortRunnerJobBody = req.body
+
+  logger.info(
+    'Remote runner %s is aborting job %s (%s)', runner.name, runnerJob.uuid, runnerJob.type,
+    { reason: body.reason, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
+  )
+
+  const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
+  await new RunnerJobHandler().abort({ runnerJob })
+
+  updateLastRunnerContact(req, runnerJob.Runner)
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+async function errorRunnerJob (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+  const runner = runnerJob.Runner
+  const body: ErrorRunnerJobBody = req.body
+
+  runnerJob.failures += 1
+
+  logger.error(
+    'Remote runner %s had an error with job %s (%s)', runner.name, runnerJob.uuid, runnerJob.type,
+    { errorMessage: body.message, totalFailures: runnerJob.failures, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
+  )
+
+  const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
+  await new RunnerJobHandler().error({ runnerJob, message: body.message })
+
+  updateLastRunnerContact(req, runnerJob.Runner)
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+// ---------------------------------------------------------------------------
+
+const jobUpdateBuilders: {
+  [id in RunnerJobType]?: (payload: RunnerJobUpdatePayload, files?: UploadFiles) => RunnerJobUpdatePayload
+} = {
+  'live-rtmp-hls-transcoding': (payload: LiveRTMPHLSTranscodingUpdatePayload, files) => {
+    return {
+      ...payload,
+
+      masterPlaylistFile: files['payload[masterPlaylistFile]']?.[0].path,
+      resolutionPlaylistFile: files['payload[resolutionPlaylistFile]']?.[0].path,
+      videoChunkFile: files['payload[videoChunkFile]']?.[0].path
+    }
+  }
+}
+
+async function updateRunnerJobController (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+  const runner = runnerJob.Runner
+  const body: RunnerJobUpdateBody = req.body
+
+  const payloadBuilder = jobUpdateBuilders[runnerJob.type]
+  const updatePayload = payloadBuilder
+    ? payloadBuilder(body.payload, req.files as UploadFiles)
+    : undefined
+
+  logger.debug(
+    'Remote runner %s is updating job %s (%s)', runnerJob.Runner.name, runnerJob.uuid, runnerJob.type,
+    { body, updatePayload, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
+  )
+
+  const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
+  await new RunnerJobHandler().update({
+    runnerJob,
+    progress: req.body.progress,
+    updatePayload
+  })
+
+  updateLastRunnerContact(req, runnerJob.Runner)
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+// ---------------------------------------------------------------------------
+
+const jobSuccessPayloadBuilders: {
+  [id in RunnerJobType]: (payload: RunnerJobSuccessPayload, files?: UploadFiles) => RunnerJobSuccessPayload
+} = {
+  'vod-web-video-transcoding': (payload: VODWebVideoTranscodingSuccess, files) => {
+    return {
+      ...payload,
+
+      videoFile: files['payload[videoFile]'][0].path
+    }
+  },
+
+  'vod-hls-transcoding': (payload: VODHLSTranscodingSuccess, files) => {
+    return {
+      ...payload,
+
+      videoFile: files['payload[videoFile]'][0].path,
+      resolutionPlaylistFile: files['payload[resolutionPlaylistFile]'][0].path
+    }
+  },
+
+  'vod-audio-merge-transcoding': (payload: VODAudioMergeTranscodingSuccess, files) => {
+    return {
+      ...payload,
+
+      videoFile: files['payload[videoFile]'][0].path
+    }
+  },
+
+  'live-rtmp-hls-transcoding': () => ({})
+}
+
+async function postRunnerJobSuccess (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+  const runner = runnerJob.Runner
+  const body: RunnerJobSuccessBody = req.body
+
+  const resultPayload = jobSuccessPayloadBuilders[runnerJob.type](body.payload, req.files as UploadFiles)
+
+  logger.info(
+    'Remote runner %s is sending success result for job %s (%s)', runnerJob.Runner.name, runnerJob.uuid, runnerJob.type,
+    { resultPayload, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
+  )
+
+  const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
+  await new RunnerJobHandler().complete({ runnerJob, resultPayload })
+
+  updateLastRunnerContact(req, runnerJob.Runner)
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+// ---------------------------------------------------------------------------
+// Controllers for admins
+// ---------------------------------------------------------------------------
+
+async function cancelRunnerJob (req: express.Request, res: express.Response) {
+  const runnerJob = res.locals.runnerJob
+
+  logger.info('Cancelling job %s (%s)', runnerJob.type, lTags(runnerJob.uuid, runnerJob.type))
+
+  const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
+  await new RunnerJobHandler().cancel({ runnerJob })
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+async function listRunnerJobs (req: express.Request, res: express.Response) {
+  const query: ListRunnerJobsQuery = req.query
+
+  const resultList = await RunnerJobModel.listForApi({
+    start: query.start,
+    count: query.count,
+    sort: query.sort,
+    search: query.search
+  })
+
+  return res.json({
+    total: resultList.total,
+    data: resultList.data.map(d => d.toFormattedAdminJSON())
+  })
+}
diff --git a/server/controllers/api/runners/manage-runners.ts b/server/controllers/api/runners/manage-runners.ts
new file mode 100644
index 000000000..eb08c4b1d
--- /dev/null
+++ b/server/controllers/api/runners/manage-runners.ts
@@ -0,0 +1,107 @@
+import express from 'express'
+import { logger, loggerTagsFactory } from '@server/helpers/logger'
+import { generateRunnerToken } from '@server/helpers/token-generator'
+import {
+  asyncMiddleware,
+  authenticate,
+  ensureUserHasRight,
+  paginationValidator,
+  runnersSortValidator,
+  setDefaultPagination,
+  setDefaultSort
+} from '@server/middlewares'
+import { deleteRunnerValidator, getRunnerFromTokenValidator, registerRunnerValidator } from '@server/middlewares/validators/runners'
+import { RunnerModel } from '@server/models/runner/runner'
+import { HttpStatusCode, ListRunnersQuery, RegisterRunnerBody, UserRight } from '@shared/models'
+
+const lTags = loggerTagsFactory('api', 'runner')
+
+const manageRunnersRouter = express.Router()
+
+manageRunnersRouter.post('/register',
+  asyncMiddleware(registerRunnerValidator),
+  asyncMiddleware(registerRunner)
+)
+manageRunnersRouter.post('/unregister',
+  asyncMiddleware(getRunnerFromTokenValidator),
+  asyncMiddleware(unregisterRunner)
+)
+
+manageRunnersRouter.delete('/:runnerId',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  asyncMiddleware(deleteRunnerValidator),
+  asyncMiddleware(deleteRunner)
+)
+
+manageRunnersRouter.get('/',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  paginationValidator,
+  runnersSortValidator,
+  setDefaultSort,
+  setDefaultPagination,
+  asyncMiddleware(listRunners)
+)
+
+// ---------------------------------------------------------------------------
+
+export {
+  manageRunnersRouter
+}
+
+// ---------------------------------------------------------------------------
+
+async function registerRunner (req: express.Request, res: express.Response) {
+  const body: RegisterRunnerBody = req.body
+
+  const runnerToken = generateRunnerToken()
+
+  const runner = new RunnerModel({
+    runnerToken,
+    name: body.name,
+    description: body.description,
+    lastContact: new Date(),
+    ip: req.ip,
+    runnerRegistrationTokenId: res.locals.runnerRegistrationToken.id
+  })
+
+  await runner.save()
+
+  logger.info('Registered new runner %s', runner.name, { ...lTags(runner.name) })
+
+  return res.json({ id: runner.id, runnerToken })
+}
+async function unregisterRunner (req: express.Request, res: express.Response) {
+  const runner = res.locals.runner
+  await runner.destroy()
+
+  logger.info('Unregistered runner %s', runner.name, { ...lTags(runner.name) })
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+async function deleteRunner (req: express.Request, res: express.Response) {
+  const runner = res.locals.runner
+
+  await runner.destroy()
+
+  logger.info('Deleted runner %s', runner.name, { ...lTags(runner.name) })
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+async function listRunners (req: express.Request, res: express.Response) {
+  const query: ListRunnersQuery = req.query
+
+  const resultList = await RunnerModel.listForApi({
+    start: query.start,
+    count: query.count,
+    sort: query.sort
+  })
+
+  return res.json({
+    total: resultList.total,
+    data: resultList.data.map(d => d.toFormattedJSON())
+  })
+}
diff --git a/server/controllers/api/runners/registration-tokens.ts b/server/controllers/api/runners/registration-tokens.ts
new file mode 100644
index 000000000..5ac3773fe
--- /dev/null
+++ b/server/controllers/api/runners/registration-tokens.ts
@@ -0,0 +1,87 @@
+import express from 'express'
+import { generateRunnerRegistrationToken } from '@server/helpers/token-generator'
+import {
+  asyncMiddleware,
+  authenticate,
+  ensureUserHasRight,
+  paginationValidator,
+  runnerRegistrationTokensSortValidator,
+  setDefaultPagination,
+  setDefaultSort
+} from '@server/middlewares'
+import { deleteRegistrationTokenValidator } from '@server/middlewares/validators/runners'
+import { RunnerRegistrationTokenModel } from '@server/models/runner/runner-registration-token'
+import { HttpStatusCode, ListRunnerRegistrationTokensQuery, UserRight } from '@shared/models'
+import { logger, loggerTagsFactory } from '@server/helpers/logger'
+
+const lTags = loggerTagsFactory('api', 'runner')
+
+const runnerRegistrationTokensRouter = express.Router()
+
+runnerRegistrationTokensRouter.post('/registration-tokens/generate',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  asyncMiddleware(generateRegistrationToken)
+)
+
+runnerRegistrationTokensRouter.delete('/registration-tokens/:id',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  asyncMiddleware(deleteRegistrationTokenValidator),
+  asyncMiddleware(deleteRegistrationToken)
+)
+
+runnerRegistrationTokensRouter.get('/registration-tokens',
+  authenticate,
+  ensureUserHasRight(UserRight.MANAGE_RUNNERS),
+  paginationValidator,
+  runnerRegistrationTokensSortValidator,
+  setDefaultSort,
+  setDefaultPagination,
+  asyncMiddleware(listRegistrationTokens)
+)
+
+// ---------------------------------------------------------------------------
+
+export {
+  runnerRegistrationTokensRouter
+}
+
+// ---------------------------------------------------------------------------
+
+async function generateRegistrationToken (req: express.Request, res: express.Response) {
+  logger.info('Generating new runner registration token.', lTags())
+
+  const registrationToken = new RunnerRegistrationTokenModel({
+    registrationToken: generateRunnerRegistrationToken()
+  })
+
+  await registrationToken.save()
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+async function deleteRegistrationToken (req: express.Request, res: express.Response) {
+  logger.info('Removing runner registration token.', lTags())
+
+  const runnerRegistrationToken = res.locals.runnerRegistrationToken
+
+  await runnerRegistrationToken.destroy()
+
+  return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
+}
+
+async function listRegistrationTokens (req: express.Request, res: express.Response) {
+  const query: ListRunnerRegistrationTokensQuery = req.query
+
+  const resultList = await RunnerRegistrationTokenModel.listForApi({
+    start: query.start,
+    count: query.count,
+    sort: query.sort
+  })
+
+  return res.json({
+    total: resultList.total,
+    data: resultList.data.map(d => d.toFormattedJSON())
+  })
+}