1 import express, { UploadFiles } from 'express'
2 import { createReqFiles } from '@server/helpers/express-utils'
3 import { logger, loggerTagsFactory } from '@server/helpers/logger'
4 import { generateRunnerJobToken } from '@server/helpers/token-generator'
5 import { MIMETYPES } from '@server/initializers/constants'
6 import { sequelizeTypescript } from '@server/initializers/database'
7 import { getRunnerJobHandlerClass, updateLastRunnerContact } from '@server/lib/runners'
13 runnerJobsSortValidator,
16 } from '@server/middlewares'
18 abortRunnerJobValidator,
19 acceptRunnerJobValidator,
20 errorRunnerJobValidator,
21 getRunnerFromTokenValidator,
22 jobOfRunnerGetValidator,
23 runnerJobGetValidator,
24 successRunnerJobValidator,
25 updateRunnerJobValidator
26 } from '@server/middlewares/validators/runners'
27 import { RunnerModel } from '@server/models/runner/runner'
28 import { RunnerJobModel } from '@server/models/runner/runner-job'
31 AcceptRunnerJobResult,
35 LiveRTMPHLSTranscodingUpdatePayload,
36 RequestRunnerJobResult,
39 RunnerJobSuccessPayload,
42 RunnerJobUpdatePayload,
44 VODAudioMergeTranscodingSuccess,
45 VODHLSTranscodingSuccess,
46 VODWebVideoTranscodingSuccess
47 } from '@shared/models'
49 const postRunnerJobSuccessVideoFiles = createReqFiles(
50 [ 'payload[videoFile]', 'payload[resolutionPlaylistFile]' ],
51 { ...MIMETYPES.VIDEO.MIMETYPE_EXT, ...MIMETYPES.M3U8.MIMETYPE_EXT }
54 const runnerJobUpdateVideoFiles = createReqFiles(
55 [ 'payload[videoChunkFile]', 'payload[resolutionPlaylistFile]', 'payload[masterPlaylistFile]' ],
56 { ...MIMETYPES.VIDEO.MIMETYPE_EXT, ...MIMETYPES.M3U8.MIMETYPE_EXT }
59 const lTags = loggerTagsFactory('api', 'runner')
61 const runnerJobsRouter = express.Router()
63 // ---------------------------------------------------------------------------
64 // Controllers for runners
65 // ---------------------------------------------------------------------------
67 runnerJobsRouter.post('/jobs/request',
68 asyncMiddleware(getRunnerFromTokenValidator),
69 asyncMiddleware(requestRunnerJob)
72 runnerJobsRouter.post('/jobs/:jobUUID/accept',
73 asyncMiddleware(runnerJobGetValidator),
74 acceptRunnerJobValidator,
75 asyncMiddleware(getRunnerFromTokenValidator),
76 asyncMiddleware(acceptRunnerJob)
79 runnerJobsRouter.post('/jobs/:jobUUID/abort',
80 asyncMiddleware(jobOfRunnerGetValidator),
81 abortRunnerJobValidator,
82 asyncMiddleware(abortRunnerJob)
85 runnerJobsRouter.post('/jobs/:jobUUID/update',
86 runnerJobUpdateVideoFiles,
87 asyncMiddleware(jobOfRunnerGetValidator),
88 updateRunnerJobValidator,
89 asyncMiddleware(updateRunnerJobController)
92 runnerJobsRouter.post('/jobs/:jobUUID/error',
93 asyncMiddleware(jobOfRunnerGetValidator),
94 errorRunnerJobValidator,
95 asyncMiddleware(errorRunnerJob)
98 runnerJobsRouter.post('/jobs/:jobUUID/success',
99 postRunnerJobSuccessVideoFiles,
100 asyncMiddleware(jobOfRunnerGetValidator),
101 successRunnerJobValidator,
102 asyncMiddleware(postRunnerJobSuccess)
105 // ---------------------------------------------------------------------------
106 // Controllers for admins
107 // ---------------------------------------------------------------------------
109 runnerJobsRouter.post('/jobs/:jobUUID/cancel',
111 ensureUserHasRight(UserRight.MANAGE_RUNNERS),
112 asyncMiddleware(runnerJobGetValidator),
113 asyncMiddleware(cancelRunnerJob)
116 runnerJobsRouter.get('/jobs',
118 ensureUserHasRight(UserRight.MANAGE_RUNNERS),
120 runnerJobsSortValidator,
122 setDefaultPagination,
123 asyncMiddleware(listRunnerJobs)
126 // ---------------------------------------------------------------------------
132 // ---------------------------------------------------------------------------
134 // ---------------------------------------------------------------------------
135 // Controllers for runners
136 // ---------------------------------------------------------------------------
138 async function requestRunnerJob (req: express.Request, res: express.Response) {
139 const runner = res.locals.runner
140 const availableJobs = await RunnerJobModel.listAvailableJobs()
142 logger.debug('Runner %s requests for a job.', runner.name, { availableJobs, ...lTags(runner.name) })
144 const result: RequestRunnerJobResult = {
145 availableJobs: availableJobs.map(j => ({
152 updateLastRunnerContact(req, runner)
154 return res.json(result)
157 async function acceptRunnerJob (req: express.Request, res: express.Response) {
158 const runner = res.locals.runner
159 const runnerJob = res.locals.runnerJob
161 runnerJob.state = RunnerJobState.PROCESSING
162 runnerJob.processingJobToken = generateRunnerJobToken()
163 runnerJob.startedAt = new Date()
164 runnerJob.runnerId = runner.id
166 const newRunnerJob = await sequelizeTypescript.transaction(transaction => {
167 return runnerJob.save({ transaction })
169 newRunnerJob.Runner = runner as RunnerModel
171 const result: AcceptRunnerJobResult = {
173 ...newRunnerJob.toFormattedJSON(),
175 jobToken: newRunnerJob.processingJobToken
179 updateLastRunnerContact(req, runner)
182 'Remote runner %s has accepted job %s (%s)', runner.name, runnerJob.uuid, runnerJob.type,
183 lTags(runner.name, runnerJob.uuid, runnerJob.type)
186 return res.json(result)
189 async function abortRunnerJob (req: express.Request, res: express.Response) {
190 const runnerJob = res.locals.runnerJob
191 const runner = runnerJob.Runner
192 const body: AbortRunnerJobBody = req.body
195 'Remote runner %s is aborting job %s (%s)', runner.name, runnerJob.uuid, runnerJob.type,
196 { reason: body.reason, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
199 const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
200 await new RunnerJobHandler().abort({ runnerJob })
202 updateLastRunnerContact(req, runnerJob.Runner)
204 return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
207 async function errorRunnerJob (req: express.Request, res: express.Response) {
208 const runnerJob = res.locals.runnerJob
209 const runner = runnerJob.Runner
210 const body: ErrorRunnerJobBody = req.body
212 runnerJob.failures += 1
215 'Remote runner %s had an error with job %s (%s)', runner.name, runnerJob.uuid, runnerJob.type,
216 { errorMessage: body.message, totalFailures: runnerJob.failures, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
219 const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
220 await new RunnerJobHandler().error({ runnerJob, message: body.message })
222 updateLastRunnerContact(req, runnerJob.Runner)
224 return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
227 // ---------------------------------------------------------------------------
229 const jobUpdateBuilders: {
230 [id in RunnerJobType]?: (payload: RunnerJobUpdatePayload, files?: UploadFiles) => RunnerJobUpdatePayload
232 'live-rtmp-hls-transcoding': (payload: LiveRTMPHLSTranscodingUpdatePayload, files) => {
236 masterPlaylistFile: files['payload[masterPlaylistFile]']?.[0].path,
237 resolutionPlaylistFile: files['payload[resolutionPlaylistFile]']?.[0].path,
238 videoChunkFile: files['payload[videoChunkFile]']?.[0].path
243 async function updateRunnerJobController (req: express.Request, res: express.Response) {
244 const runnerJob = res.locals.runnerJob
245 const runner = runnerJob.Runner
246 const body: RunnerJobUpdateBody = req.body
248 const payloadBuilder = jobUpdateBuilders[runnerJob.type]
249 const updatePayload = payloadBuilder
250 ? payloadBuilder(body.payload, req.files as UploadFiles)
254 'Remote runner %s is updating job %s (%s)', runnerJob.Runner.name, runnerJob.uuid, runnerJob.type,
255 { body, updatePayload, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
258 const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
259 await new RunnerJobHandler().update({
261 progress: req.body.progress,
265 updateLastRunnerContact(req, runnerJob.Runner)
267 return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
270 // ---------------------------------------------------------------------------
272 const jobSuccessPayloadBuilders: {
273 [id in RunnerJobType]: (payload: RunnerJobSuccessPayload, files?: UploadFiles) => RunnerJobSuccessPayload
275 'vod-web-video-transcoding': (payload: VODWebVideoTranscodingSuccess, files) => {
279 videoFile: files['payload[videoFile]'][0].path
283 'vod-hls-transcoding': (payload: VODHLSTranscodingSuccess, files) => {
287 videoFile: files['payload[videoFile]'][0].path,
288 resolutionPlaylistFile: files['payload[resolutionPlaylistFile]'][0].path
292 'vod-audio-merge-transcoding': (payload: VODAudioMergeTranscodingSuccess, files) => {
296 videoFile: files['payload[videoFile]'][0].path
300 'live-rtmp-hls-transcoding': () => ({})
303 async function postRunnerJobSuccess (req: express.Request, res: express.Response) {
304 const runnerJob = res.locals.runnerJob
305 const runner = runnerJob.Runner
306 const body: RunnerJobSuccessBody = req.body
308 const resultPayload = jobSuccessPayloadBuilders[runnerJob.type](body.payload, req.files as UploadFiles)
311 'Remote runner %s is sending success result for job %s (%s)', runnerJob.Runner.name, runnerJob.uuid, runnerJob.type,
312 { resultPayload, ...lTags(runner.name, runnerJob.uuid, runnerJob.type) }
315 const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
316 await new RunnerJobHandler().complete({ runnerJob, resultPayload })
318 updateLastRunnerContact(req, runnerJob.Runner)
320 return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
323 // ---------------------------------------------------------------------------
324 // Controllers for admins
325 // ---------------------------------------------------------------------------
327 async function cancelRunnerJob (req: express.Request, res: express.Response) {
328 const runnerJob = res.locals.runnerJob
330 logger.info('Cancelling job %s (%s)', runnerJob.type, lTags(runnerJob.uuid, runnerJob.type))
332 const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
333 await new RunnerJobHandler().cancel({ runnerJob })
335 return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
338 async function listRunnerJobs (req: express.Request, res: express.Response) {
339 const query: ListRunnerJobsQuery = req.query
341 const resultList = await RunnerJobModel.listForApi({
349 total: resultList.total,
350 data: resultList.data.map(d => d.toFormattedAdminJSON())