interesting
/
PeerTube
mirror of https://github.com/Chocobozzz/PeerTube.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Prevent stalled jobs

This commit is contained in:
Chocobozzz 2023-06-22 15:25:39 +02:00
parent bc3918b2ae
commit d68b88bac4
No known key found for this signature in database
GPG Key ID: 583A612D890159BE
3 changed files with 126 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
packages/peertube-runner/server/process/shared/common.ts
View File

@ -35,49 +35,48 @@ export async function downloadInputFile (options: {
return destination
}
export async function updateTranscodingProgress (options: {
export function scheduleTranscodingProgress (options: {
server: PeerTubeServer
runnerToken: string
job: JobWithToken
progress: number
progressGetter: () => number
}) {
const { server, job, runnerToken, progress } = options
return server.runnerJobs.update({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken, progress })
}
// ---------------------------------------------------------------------------
export function buildFFmpegVOD (options: {
server: PeerTubeServer
runnerToken: string
job: JobWithToken
}) {
const { server, job, runnerToken } = options
const { job, server, progressGetter, runnerToken } = options
const updateInterval = ConfigManager.Instance.isTestInstance()
? 500
: 60000
let progress: number
const update = () => {
server.runnerJobs.update({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken, progress: progressGetter() })
.catch(err => logger.error({ err }, 'Cannot send job progress'))
}
const interval = setInterval(() => {
updateTranscodingProgress({ server, job, runnerToken, progress })
.catch(err => logger.error({ err }, 'Cannot send job progress'))
update()
}, updateInterval)
update()
return interval
}
// ---------------------------------------------------------------------------
export function buildFFmpegVOD (options: {
onJobProgress: (progress: number) => void
}) {
const { onJobProgress } = options
return new FFmpegVOD({
...getCommonFFmpegOptions(),
onError: () => clearInterval(interval),
onEnd: () => clearInterval(interval),
updateJobProgress: arg => {
if (arg < 0 || arg > 100) {
progress = undefined
} else {
progress = arg
}
const progress = arg < 0 || arg > 100
? undefined
: arg
onJobProgress(progress)
}
})
}

35
packages/peertube-runner/server/process/shared/process-studio.ts
View File

@ -5,30 +5,42 @@ import { join } from 'path'
import { buildUUID } from '@shared/extra-utils'
import {
RunnerJobStudioTranscodingPayload,
VideoStudioTranscodingSuccess,
VideoStudioTask,
VideoStudioTaskCutPayload,
VideoStudioTaskIntroPayload,
VideoStudioTaskOutroPayload,
VideoStudioTaskPayload,
VideoStudioTaskWatermarkPayload
VideoStudioTaskWatermarkPayload,
VideoStudioTranscodingSuccess
} from '@shared/models'
import { ConfigManager } from '../../../shared/config-manager'
import { buildFFmpegEdition, downloadInputFile, JobWithToken, ProcessOptions } from './common'
import { buildFFmpegEdition, downloadInputFile, JobWithToken, ProcessOptions, scheduleTranscodingProgress } from './common'
export async function processStudioTranscoding (options: ProcessOptions<RunnerJobStudioTranscodingPayload>) {
const { server, job, runnerToken } = options
const payload = job.payload
logger.info(`Downloading input file ${payload.input.videoFileUrl} for job ${job.jobToken}`)
let inputPath: string
let outputPath: string
const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
let tmpInputFilePath = inputPath
let tmpInputFilePath: string
logger.info(`Input file ${payload.input.videoFileUrl} downloaded for job ${job.jobToken}. Running studio transcoding tasks.`)
let tasksProgress = 0
const updateProgressInterval = scheduleTranscodingProgress({
job,
server,
runnerToken,
progressGetter: () => tasksProgress
})
try {
logger.info(`Downloading input file ${payload.input.videoFileUrl} for job ${job.jobToken}`)
inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
tmpInputFilePath = inputPath
logger.info(`Input file ${payload.input.videoFileUrl} downloaded for job ${job.jobToken}. Running studio transcoding tasks.`)
for (const task of payload.tasks) {
const outputFilename = 'output-edition-' + buildUUID() + '.mp4'
outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), outputFilename)
@ -45,6 +57,8 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
// For the next iteration
tmpInputFilePath = outputPath
tasksProgress += Math.floor(100 / payload.tasks.length)
}
const successBody: VideoStudioTranscodingSuccess = {
@ -58,8 +72,9 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
payload: successBody
})
} finally {
await remove(tmpInputFilePath)
await remove(outputPath)
if (tmpInputFilePath) await remove(tmpInputFilePath)
if (outputPath) await remove(outputPath)
if (updateProgressInterval) clearInterval(updateProgressInterval)
}
}

112
packages/peertube-runner/server/process/shared/process-vod.ts
View File

@ -11,23 +11,36 @@ import {
VODWebVideoTranscodingSuccess
} from '@shared/models'
import { ConfigManager } from '../../../shared/config-manager'
import { buildFFmpegVOD, downloadInputFile, ProcessOptions } from './common'
import { buildFFmpegVOD, downloadInputFile, ProcessOptions, scheduleTranscodingProgress } from './common'
export async function processWebVideoTranscoding (options: ProcessOptions<RunnerJobVODWebVideoTranscodingPayload>) {
const { server, job, runnerToken } = options
const payload = job.payload
logger.info(`Downloading input file ${payload.input.videoFileUrl} for web video transcoding job ${job.jobToken}`)
const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running web video transcoding.`)
const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
let ffmpegProgress: number
let inputPath: string
const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
const updateProgressInterval = scheduleTranscodingProgress({
job,
server,
runnerToken,
progressGetter: () => ffmpegProgress
})
try {
logger.info(`Downloading input file ${payload.input.videoFileUrl} for web video transcoding job ${job.jobToken}`)
inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running web video transcoding.`)
const ffmpegVod = buildFFmpegVOD({
onJobProgress: progress => { ffmpegProgress = progress }
})
await ffmpegVod.transcode({
type: 'video',
@ -52,8 +65,9 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
payload: successBody
})
} finally {
await remove(inputPath)
await remove(outputPath)
if (inputPath) await remove(inputPath)
if (outputPath) await remove(outputPath)
if (updateProgressInterval) clearInterval(updateProgressInterval)
}
}
@ -61,21 +75,32 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
const { server, job, runnerToken } = options
const payload = job.payload
logger.info(`Downloading input file ${payload.input.videoFileUrl} for HLS transcoding job ${job.jobToken}`)
const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running HLS transcoding.`)
let ffmpegProgress: number
let inputPath: string
const uuid = buildUUID()
const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `${uuid}-${payload.output.resolution}.m3u8`)
const videoFilename = `${uuid}-${payload.output.resolution}-fragmented.mp4`
const videoPath = join(join(ConfigManager.Instance.getTranscodingDirectory(), videoFilename))
const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
const updateProgressInterval = scheduleTranscodingProgress({
job,
server,
runnerToken,
progressGetter: () => ffmpegProgress
})
try {
logger.info(`Downloading input file ${payload.input.videoFileUrl} for HLS transcoding job ${job.jobToken}`)
inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running HLS transcoding.`)
const ffmpegVod = buildFFmpegVOD({
onJobProgress: progress => { ffmpegProgress = progress }
})
await ffmpegVod.transcode({
type: 'hls',
copyCodecs: false,
@ -101,9 +126,10 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
payload: successBody
})
} finally {
await remove(inputPath)
await remove(outputPath)
await remove(videoPath)
if (inputPath) await remove(inputPath)
if (outputPath) await remove(outputPath)
if (videoPath) await remove(videoPath)
if (updateProgressInterval) clearInterval(updateProgressInterval)
}
}
@ -111,24 +137,37 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
const { server, job, runnerToken } = options
const payload = job.payload
logger.info(
`Downloading input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
`for audio merge transcoding job ${job.jobToken}`
)
const audioPath = await downloadInputFile({ url: payload.input.audioFileUrl, runnerToken, job })
const inputPath = await downloadInputFile({ url: payload.input.previewFileUrl, runnerToken, job })
logger.info(
`Downloaded input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
`for job ${job.jobToken}. Running audio merge transcoding.`
)
let ffmpegProgress: number
let audioPath: string
let inputPath: string
const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
const updateProgressInterval = scheduleTranscodingProgress({
job,
server,
runnerToken,
progressGetter: () => ffmpegProgress
})
try {
logger.info(
`Downloading input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
`for audio merge transcoding job ${job.jobToken}`
)
audioPath = await downloadInputFile({ url: payload.input.audioFileUrl, runnerToken, job })
inputPath = await downloadInputFile({ url: payload.input.previewFileUrl, runnerToken, job })
logger.info(
`Downloaded input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
`for job ${job.jobToken}. Running audio merge transcoding.`
)
const ffmpegVod = buildFFmpegVOD({
onJobProgress: progress => { ffmpegProgress = progress }
})
await ffmpegVod.transcode({
type: 'merge-audio',
@ -154,8 +193,9 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
payload: successBody
})
} finally {
await remove(audioPath)
await remove(inputPath)
await remove(outputPath)
if (audioPath) await remove(audioPath)
if (inputPath) await remove(inputPath)
if (outputPath) await remove(outputPath)
if (updateProgressInterval) clearInterval(updateProgressInterval)
}
}