private isRemoteRunnersEnabled () {
const config = this.server.getHTMLConfig()
- return config.transcoding.remoteRunners.enabled || config.live.transcoding.remoteRunners.enabled
+ return config.transcoding.remoteRunners.enabled ||
+ config.live.transcoding.remoteRunners.enabled ||
+ config.videoStudio.remoteRunners.enabled
}
}
return form.value['transcoding']['enabled'] === true
}
+ isStudioEnabled (form: FormGroup) {
+ return form.value['videoStudio']['enabled'] === true
+ }
+
isLiveEnabled (form: FormGroup) {
return form.value['live']['enabled'] === true
}
}
},
videoStudio: {
- enabled: null
+ enabled: null,
+ remoteRunners: {
+ enabled: null
+ }
},
autoBlacklist: {
videos: {
</ng-container>
</my-peertube-checkbox>
</div>
+
+ <div class="form-group" formGroupName="remoteRunners" [ngClass]="getStudioDisabledClass()">
+ <my-peertube-checkbox
+ inputName="videoStudioRemoteRunnersEnabled" formControlName="enabled"
+ i18n-labelText labelText="Enable remote runners"
+ >
+ <ng-container ngProjectAs="description">
+ <span i18n>
+ Use <a routerLink="/admin/system/runners/runners-list">remote runners</a> to process studio transcoding tasks.
+ Remote runners has to register on your instance first.
+ </span>
+ </ng-container>
+ </my-peertube-checkbox>
+ </div>
</ng-container>
</div>
</div>
return this.editConfigurationService.isTranscodingEnabled(this.form)
}
+ isStudioEnabled () {
+ return this.editConfigurationService.isStudioEnabled(this.form)
+ }
+
getTranscodingDisabledClass () {
return { 'disabled-checkbox-extra': !this.isTranscodingEnabled() }
}
+ getStudioDisabledClass () {
+ return { 'disabled-checkbox-extra': !this.isStudioEnabled() }
+ }
+
getTotalTranscodingThreads () {
return this.editConfigurationService.getTotalTranscodingThreads(this.form)
}
# If enabled, users can create transcoding tasks as they wish
enabled: false
+ # Enable remote runners to transcode studio tasks
+ # If enabled, your instance won't transcode the videos itself
+ # At least 1 remote runner must be configured to transcode your videos
+ remote_runners:
+ enabled: false
+
import:
# Add ability for your users to import remote videos (from YouTube, torrent...)
videos:
# If enabled, users can create transcoding tasks as they wish
enabled: false
+
+ # Enable remote runners to transcode studio tasks
+ # If enabled, your instance won't transcode the videos itself
+ # At least 1 remote runner must be configured to transcode your videos
+ remote_runners:
+ enabled: false
+
import:
# Add ability for your users to import remote videos (from YouTube, torrent...)
videos:
import { logger } from 'packages/peertube-runner/shared/logger'
import {
RunnerJobLiveRTMPHLSTranscodingPayload,
+ RunnerJobVideoEditionTranscodingPayload,
RunnerJobVODAudioMergeTranscodingPayload,
RunnerJobVODHLSTranscodingPayload,
RunnerJobVODWebVideoTranscodingPayload
} from '@shared/models'
import { processAudioMergeTranscoding, processHLSTranscoding, ProcessOptions, processWebVideoTranscoding } from './shared'
import { ProcessLiveRTMPHLSTranscoding } from './shared/process-live'
+import { processStudioTranscoding } from './shared/process-studio'
export async function processJob (options: ProcessOptions) {
const { server, job } = options
await processHLSTranscoding(options as ProcessOptions<RunnerJobVODHLSTranscodingPayload>)
} else if (job.type === 'live-rtmp-hls-transcoding') {
await new ProcessLiveRTMPHLSTranscoding(options as ProcessOptions<RunnerJobLiveRTMPHLSTranscodingPayload>).process()
+ } else if (job.type === 'video-edition-transcoding') {
+ await processStudioTranscoding(options as ProcessOptions<RunnerJobVideoEditionTranscodingPayload>)
} else {
logger.error(`Unknown job ${job.type} to process`)
return
import { ConfigManager, downloadFile, logger } from 'packages/peertube-runner/shared'
import { join } from 'path'
import { buildUUID } from '@shared/extra-utils'
-import { FFmpegLive, FFmpegVOD } from '@shared/ffmpeg'
+import { FFmpegEdition, FFmpegLive, FFmpegVOD } from '@shared/ffmpeg'
import { RunnerJob, RunnerJobPayload } from '@shared/models'
import { PeerTubeServer } from '@shared/server-commands'
import { getTranscodingLogger } from './transcoding-logger'
import { getAvailableEncoders, getEncodersToTry } from './transcoding-profiles'
+import { remove } from 'fs-extra'
export type JobWithToken <T extends RunnerJobPayload = RunnerJobPayload> = RunnerJob<T> & { jobToken: string }
const { url, job, runnerToken } = options
const destination = join(ConfigManager.Instance.getTranscodingDirectory(), buildUUID())
- await downloadFile({ url, jobToken: job.jobToken, runnerToken, destination })
+ try {
+ await downloadFile({ url, jobToken: job.jobToken, runnerToken, destination })
+ } catch (err) {
+ remove(destination)
+ .catch(err => logger.error({ err }, `Cannot remove ${destination}`))
+
+ throw err
+ }
return destination
}
return server.runnerJobs.update({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken, progress })
}
+// ---------------------------------------------------------------------------
+
export function buildFFmpegVOD (options: {
server: PeerTubeServer
runnerToken: string
.catch(err => logger.error({ err }, 'Cannot send job progress'))
}, updateInterval, { trailing: false })
- const config = ConfigManager.Instance.getConfig()
-
return new FFmpegVOD({
- niceness: config.ffmpeg.nice,
- threads: config.ffmpeg.threads,
- tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
- profile: 'default',
- availableEncoders: {
- available: getAvailableEncoders(),
- encodersToTry: getEncodersToTry()
- },
- logger: getTranscodingLogger(),
+ ...getCommonFFmpegOptions(),
+
updateJobProgress
})
}
export function buildFFmpegLive () {
+ return new FFmpegLive(getCommonFFmpegOptions())
+}
+
+export function buildFFmpegEdition () {
+ return new FFmpegEdition(getCommonFFmpegOptions())
+}
+
+function getCommonFFmpegOptions () {
const config = ConfigManager.Instance.getConfig()
- return new FFmpegLive({
+ return {
niceness: config.ffmpeg.nice,
threads: config.ffmpeg.threads,
tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
encodersToTry: getEncodersToTry()
},
logger: getTranscodingLogger()
- })
+ }
}
--- /dev/null
+import { remove } from 'fs-extra'
+import { pick } from 'lodash'
+import { logger } from 'packages/peertube-runner/shared'
+import { extname, join } from 'path'
+import { buildUUID } from '@shared/extra-utils'
+import {
+ RunnerJobVideoEditionTranscodingPayload,
+ VideoEditionTranscodingSuccess,
+ VideoStudioTask,
+ VideoStudioTaskCutPayload,
+ VideoStudioTaskIntroPayload,
+ VideoStudioTaskOutroPayload,
+ VideoStudioTaskPayload,
+ VideoStudioTaskWatermarkPayload
+} from '@shared/models'
+import { ConfigManager } from '../../../shared/config-manager'
+import { buildFFmpegEdition, downloadInputFile, JobWithToken, ProcessOptions } from './common'
+
+export async function processStudioTranscoding (options: ProcessOptions<RunnerJobVideoEditionTranscodingPayload>) {
+ const { server, job, runnerToken } = options
+ const payload = job.payload
+
+ let outputPath: string
+ const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
+ let tmpInputFilePath = inputPath
+
+ try {
+ for (const task of payload.tasks) {
+ const outputFilename = 'output-edition-' + buildUUID() + '.mp4'
+ outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), outputFilename)
+
+ await processTask({
+ inputPath: tmpInputFilePath,
+ outputPath,
+ task,
+ job,
+ runnerToken
+ })
+
+ if (tmpInputFilePath) await remove(tmpInputFilePath)
+
+ // For the next iteration
+ tmpInputFilePath = outputPath
+ }
+
+ const successBody: VideoEditionTranscodingSuccess = {
+ videoFile: outputPath
+ }
+
+ await server.runnerJobs.success({
+ jobToken: job.jobToken,
+ jobUUID: job.uuid,
+ runnerToken,
+ payload: successBody
+ })
+ } finally {
+ await remove(tmpInputFilePath)
+ await remove(outputPath)
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Private
+// ---------------------------------------------------------------------------
+
+type TaskProcessorOptions <T extends VideoStudioTaskPayload = VideoStudioTaskPayload> = {
+ inputPath: string
+ outputPath: string
+ task: T
+ runnerToken: string
+ job: JobWithToken
+}
+
+const taskProcessors: { [id in VideoStudioTask['name']]: (options: TaskProcessorOptions) => Promise<any> } = {
+ 'add-intro': processAddIntroOutro,
+ 'add-outro': processAddIntroOutro,
+ 'cut': processCut,
+ 'add-watermark': processAddWatermark
+}
+
+async function processTask (options: TaskProcessorOptions) {
+ const { task } = options
+
+ const processor = taskProcessors[options.task.name]
+ if (!process) throw new Error('Unknown task ' + task.name)
+
+ return processor(options)
+}
+
+async function processAddIntroOutro (options: TaskProcessorOptions<VideoStudioTaskIntroPayload | VideoStudioTaskOutroPayload>) {
+ const { inputPath, task, runnerToken, job } = options
+
+ logger.debug('Adding intro/outro to ' + inputPath)
+
+ const introOutroPath = await downloadInputFile({ url: task.options.file, runnerToken, job })
+
+ return buildFFmpegEdition().addIntroOutro({
+ ...pick(options, [ 'inputPath', 'outputPath' ]),
+
+ introOutroPath,
+ type: task.name === 'add-intro'
+ ? 'intro'
+ : 'outro'
+ })
+}
+
+function processCut (options: TaskProcessorOptions<VideoStudioTaskCutPayload>) {
+ const { inputPath, task } = options
+
+ logger.debug(`Cutting ${inputPath}`)
+
+ return buildFFmpegEdition().cutVideo({
+ ...pick(options, [ 'inputPath', 'outputPath' ]),
+
+ start: task.options.start,
+ end: task.options.end
+ })
+}
+
+async function processAddWatermark (options: TaskProcessorOptions<VideoStudioTaskWatermarkPayload>) {
+ const { inputPath, task, runnerToken, job } = options
+
+ logger.debug('Adding watermark to ' + inputPath)
+
+ const watermarkPath = await downloadInputFile({ url: task.options.file, runnerToken, job })
+
+ return buildFFmpegEdition().addWatermark({
+ ...pick(options, [ 'inputPath', 'outputPath' ]),
+
+ watermarkPath,
+
+ videoFilters: {
+ watermarkSizeRatio: task.options.watermarkSizeRatio,
+ horitonzalMarginRatio: task.options.horitonzalMarginRatio,
+ verticalMarginRatio: task.options.verticalMarginRatio
+ }
+ })
+}
const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
- await ffmpegVod.transcode({
- type: 'hls',
- copyCodecs: false,
- inputPath,
- hlsPlaylist: { videoFilename },
- outputPath,
-
- inputFileMutexReleaser: () => {},
-
- resolution: payload.output.resolution,
- fps: payload.output.fps
- })
-
- const successBody: VODHLSTranscodingSuccess = {
- resolutionPlaylistFile: outputPath,
- videoFile: videoPath
+ try {
+ await ffmpegVod.transcode({
+ type: 'hls',
+ copyCodecs: false,
+ inputPath,
+ hlsPlaylist: { videoFilename },
+ outputPath,
+
+ inputFileMutexReleaser: () => {},
+
+ resolution: payload.output.resolution,
+ fps: payload.output.fps
+ })
+
+ const successBody: VODHLSTranscodingSuccess = {
+ resolutionPlaylistFile: outputPath,
+ videoFile: videoPath
+ }
+
+ await server.runnerJobs.success({
+ jobToken: job.jobToken,
+ jobUUID: job.uuid,
+ runnerToken,
+ payload: successBody
+ })
+ } finally {
+ await remove(inputPath)
+ await remove(outputPath)
+ await remove(videoPath)
}
-
- await server.runnerJobs.success({
- jobToken: job.jobToken,
- jobUUID: job.uuid,
- runnerToken,
- payload: successBody
- })
-
- await remove(outputPath)
- await remove(videoPath)
}
export async function processAudioMergeTranscoding (options: ProcessOptions<RunnerJobVODAudioMergeTranscodingPayload>) {
import { IPCServer } from '../shared/ipc'
import { logger } from '../shared/logger'
import { JobWithToken, processJob } from './process'
+import { isJobSupported } from './shared'
type PeerTubeServer = PeerTubeServerCommand & {
runnerToken: string
const { availableJobs } = await server.runnerJobs.request({ runnerToken: server.runnerToken })
- if (availableJobs.length === 0) {
+ const filtered = availableJobs.filter(j => isJobSupported(j))
+
+ if (filtered.length === 0) {
logger.debug(`No job available on ${server.url} for runner ${server.runnerName}`)
return undefined
}
- return availableJobs[0]
+ return filtered[0]
}
private async tryToExecuteJobAsync (server: PeerTubeServer, jobToAccept: { uuid: string }) {
--- /dev/null
+export * from './supported-job'
--- /dev/null
+import {
+ RunnerJobLiveRTMPHLSTranscodingPayload,
+ RunnerJobPayload,
+ RunnerJobType,
+ RunnerJobVideoEditionTranscodingPayload,
+ RunnerJobVODAudioMergeTranscodingPayload,
+ RunnerJobVODHLSTranscodingPayload,
+ RunnerJobVODWebVideoTranscodingPayload,
+ VideoStudioTaskPayload
+} from '@shared/models'
+
+const supportedMatrix = {
+ 'vod-web-video-transcoding': (_payload: RunnerJobVODWebVideoTranscodingPayload) => {
+ return true
+ },
+ 'vod-hls-transcoding': (_payload: RunnerJobVODHLSTranscodingPayload) => {
+ return true
+ },
+ 'vod-audio-merge-transcoding': (_payload: RunnerJobVODAudioMergeTranscodingPayload) => {
+ return true
+ },
+ 'live-rtmp-hls-transcoding': (_payload: RunnerJobLiveRTMPHLSTranscodingPayload) => {
+ return true
+ },
+ 'video-edition-transcoding': (payload: RunnerJobVideoEditionTranscodingPayload) => {
+ const tasks = payload?.tasks
+ const supported = new Set<VideoStudioTaskPayload['name']>([ 'add-intro', 'add-outro', 'add-watermark', 'cut' ])
+
+ if (!Array.isArray(tasks)) return false
+
+ return tasks.every(t => t && supported.has(t.name))
+ }
+}
+
+export function isJobSupported (job: {
+ type: RunnerJobType
+ payload: RunnerJobPayload
+}) {
+ const fn = supportedMatrix[job.type]
+ if (!fn) return false
+
+ return fn(job.payload as any)
+}
}
},
videoStudio: {
- enabled: CONFIG.VIDEO_STUDIO.ENABLED
+ enabled: CONFIG.VIDEO_STUDIO.ENABLED,
+ remoteRunners: {
+ enabled: CONFIG.VIDEO_STUDIO.REMOTE_RUNNERS.ENABLED
+ }
},
import: {
videos: {
import { logger, loggerTagsFactory } from '@server/helpers/logger'
import { proxifyHLS, proxifyWebTorrentFile } from '@server/lib/object-storage'
import { VideoPathManager } from '@server/lib/video-path-manager'
+import { getStudioTaskFilePath } from '@server/lib/video-studio'
import { asyncMiddleware } from '@server/middlewares'
import { jobOfRunnerGetValidator } from '@server/middlewares/validators/runners'
-import { runnerJobGetVideoTranscodingFileValidator } from '@server/middlewares/validators/runners/job-files'
+import {
+ runnerJobGetVideoStudioTaskFileValidator,
+ runnerJobGetVideoTranscodingFileValidator
+} from '@server/middlewares/validators/runners/job-files'
import { VideoStorage } from '@shared/models'
const lTags = loggerTagsFactory('api', 'runner')
getMaxQualityVideoPreview
)
+runnerJobFilesRouter.post('/jobs/:jobUUID/files/videos/:videoId/studio/task-files/:filename',
+ asyncMiddleware(jobOfRunnerGetValidator),
+ asyncMiddleware(runnerJobGetVideoTranscodingFileValidator),
+ runnerJobGetVideoStudioTaskFileValidator,
+ getVideoEditionTaskFile
+)
+
// ---------------------------------------------------------------------------
export {
return res.sendFile(file.getPath())
}
+
+function getVideoEditionTaskFile (req: express.Request, res: express.Response) {
+ const runnerJob = res.locals.runnerJob
+ const runner = runnerJob.Runner
+ const video = res.locals.videoAll
+ const filename = req.params.filename
+
+ logger.info(
+ 'Get video edition task file %s of video %s of job %s for runner %s', filename, video.uuid, runnerJob.uuid, runner.name,
+ lTags(runner.name, runnerJob.id, runnerJob.type)
+ )
+
+ return res.sendFile(getStudioTaskFilePath(filename))
+}
import {
abortRunnerJobValidator,
acceptRunnerJobValidator,
+ cancelRunnerJobValidator,
errorRunnerJobValidator,
getRunnerFromTokenValidator,
jobOfRunnerGetValidator,
RunnerJobUpdateBody,
RunnerJobUpdatePayload,
UserRight,
+ VideoEditionTranscodingSuccess,
VODAudioMergeTranscodingSuccess,
VODHLSTranscodingSuccess,
VODWebVideoTranscodingSuccess
authenticate,
ensureUserHasRight(UserRight.MANAGE_RUNNERS),
asyncMiddleware(runnerJobGetValidator),
+ cancelRunnerJobValidator,
asyncMiddleware(cancelRunnerJob)
)
}
},
+ 'video-edition-transcoding': (payload: VideoEditionTranscodingSuccess, files) => {
+ return {
+ ...payload,
+
+ videoFile: files['payload[videoFile]'][0].path
+ }
+ },
+
'live-rtmp-hls-transcoding': () => ({})
}
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))
+ logger.info('Cancelling job %s (%s)', runnerJob.uuid, runnerJob.type, lTags(runnerJob.uuid, runnerJob.type))
const RunnerJobHandler = getRunnerJobHandlerClass(runnerJob)
await new RunnerJobHandler().cancel({ runnerJob })
import Bluebird from 'bluebird'
import express from 'express'
import { move } from 'fs-extra'
-import { basename, join } from 'path'
+import { basename } from 'path'
import { createAnyReqFiles } from '@server/helpers/express-utils'
-import { CONFIG } from '@server/initializers/config'
-import { MIMETYPES } from '@server/initializers/constants'
-import { JobQueue } from '@server/lib/job-queue'
-import { buildTaskFileFieldname, getTaskFileFromReq } from '@server/lib/video-studio'
+import { MIMETYPES, VIDEO_FILTERS } from '@server/initializers/constants'
+import { buildTaskFileFieldname, createVideoStudioJob, getStudioTaskFilePath, getTaskFileFromReq } from '@server/lib/video-studio'
import {
HttpStatusCode,
VideoState,
tasks: await Bluebird.mapSeries(body.tasks, (t, i) => buildTaskPayload(t, i, files))
}
- JobQueue.Instance.createJobAsync({ type: 'video-studio-edition', payload })
+ await createVideoStudioJob({
+ user: res.locals.oauth.token.User,
+ payload,
+ video
+ })
return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
}
return {
name: task.name,
options: {
- file: destination
+ file: destination,
+ watermarkSizeRatio: VIDEO_FILTERS.WATERMARK.SIZE_RATIO,
+ horitonzalMarginRatio: VIDEO_FILTERS.WATERMARK.HORIZONTAL_MARGIN_RATIO,
+ verticalMarginRatio: VIDEO_FILTERS.WATERMARK.VERTICAL_MARGIN_RATIO
}
}
}
async function moveStudioFileToPersistentTMP (file: string) {
- const destination = join(CONFIG.STORAGE.TMP_PERSISTENT_DIR, basename(file))
+ const destination = getStudioTaskFilePath(basename(file))
await move(file, destination)
})
}
-function isSafeFilename (filename: string, extension: string) {
- return typeof filename === 'string' && !!filename.match(new RegExp(`^[a-z0-9-]+\\.${extension}$`))
+function isSafeFilename (filename: string, extension?: string) {
+ const regex = extension
+ ? new RegExp(`^[a-z0-9-]+\\.${extension}$`)
+ : new RegExp(`^[a-z0-9-]+\\.[a-z0-9]{1,8}$`)
+
+ return typeof filename === 'string' && !!filename.match(regex)
}
function isSafePeerTubeFilenameWithoutExtension (filename: string) {
RunnerJobSuccessPayload,
RunnerJobType,
RunnerJobUpdatePayload,
+ VideoEditionTranscodingSuccess,
VODAudioMergeTranscodingSuccess,
VODHLSTranscodingSuccess,
VODWebVideoTranscodingSuccess
return isRunnerJobVODWebVideoResultPayloadValid(value as VODWebVideoTranscodingSuccess, type, files) ||
isRunnerJobVODHLSResultPayloadValid(value as VODHLSTranscodingSuccess, type, files) ||
isRunnerJobVODAudioMergeResultPayloadValid(value as VODHLSTranscodingSuccess, type, files) ||
- isRunnerJobLiveRTMPHLSResultPayloadValid(value as LiveRTMPHLSTranscodingSuccess, type)
+ isRunnerJobLiveRTMPHLSResultPayloadValid(value as LiveRTMPHLSTranscodingSuccess, type) ||
+ isRunnerJobVideoEditionResultPayloadValid(value as VideoEditionTranscodingSuccess, type, files)
}
// ---------------------------------------------------------------------------
function isRunnerJobUpdatePayloadValid (value: RunnerJobUpdatePayload, type: RunnerJobType, files: UploadFilesForCheck) {
return isRunnerJobVODWebVideoUpdatePayloadValid(value, type, files) ||
isRunnerJobVODHLSUpdatePayloadValid(value, type, files) ||
+ isRunnerJobVideoEditionUpdatePayloadValid(value, type, files) ||
isRunnerJobVODAudioMergeUpdatePayloadValid(value, type, files) ||
isRunnerJobLiveRTMPHLSUpdatePayloadValid(value, type, files)
}
return type === 'live-rtmp-hls-transcoding' && (!value || (typeof value === 'object' && Object.keys(value).length === 0))
}
+function isRunnerJobVideoEditionResultPayloadValid (
+ _value: VideoEditionTranscodingSuccess,
+ type: RunnerJobType,
+ files: UploadFilesForCheck
+) {
+ return type === 'video-edition-transcoding' &&
+ isFileValid({ files, field: 'payload[videoFile]', mimeTypeRegex: null, maxSize: null })
+}
+
// ---------------------------------------------------------------------------
function isRunnerJobVODWebVideoUpdatePayloadValid (
)
)
}
+
+function isRunnerJobVideoEditionUpdatePayloadValid (
+ value: RunnerJobUpdatePayload,
+ type: RunnerJobType,
+ _files: UploadFilesForCheck
+) {
+ return type === 'video-edition-transcoding' &&
+ (!value || (typeof value === 'object' && Object.keys(value).length === 0))
+}
'transcoding.resolutions.0p', 'transcoding.resolutions.144p', 'transcoding.resolutions.240p', 'transcoding.resolutions.360p',
'transcoding.resolutions.480p', 'transcoding.resolutions.720p', 'transcoding.resolutions.1080p', 'transcoding.resolutions.1440p',
'transcoding.resolutions.2160p', 'transcoding.always_transcode_original_resolution', 'transcoding.remote_runners.enabled',
- 'video_studio.enabled',
+ 'video_studio.enabled', 'video_studio.remote_runners.enabled',
'remote_runners.stalled_jobs.vod', 'remote_runners.stalled_jobs.live',
'import.videos.http.enabled', 'import.videos.torrent.enabled', 'import.videos.concurrency', 'import.videos.timeout',
'import.video_channel_synchronization.enabled', 'import.video_channel_synchronization.max_per_user',
}
},
VIDEO_STUDIO: {
- get ENABLED () { return config.get<boolean>('video_studio.enabled') }
+ get ENABLED () { return config.get<boolean>('video_studio.enabled') },
+ REMOTE_RUNNERS: {
+ get ENABLED () { return config.get<boolean>('video_studio.remote_runners.enabled') }
+ }
},
IMPORT: {
VIDEOS: {
}
}
const JOB_PRIORITY = {
- TRANSCODING: 100
+ TRANSCODING: 100,
+ VIDEO_STUDIO: 150
}
const JOB_REMOVAL_OPTIONS = {
import { Job } from 'bullmq'
-import { move, remove } from 'fs-extra'
+import { remove } from 'fs-extra'
import { join } from 'path'
import { getFFmpegCommandWrapperOptions } from '@server/helpers/ffmpeg'
-import { createTorrentAndSetInfoHashFromPath } from '@server/helpers/webtorrent'
import { CONFIG } from '@server/initializers/config'
-import { VIDEO_FILTERS } from '@server/initializers/constants'
-import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
-import { generateWebTorrentVideoFilename } from '@server/lib/paths'
-import { createOptimizeOrMergeAudioJobs } from '@server/lib/transcoding/create-transcoding-job'
import { VideoTranscodingProfilesManager } from '@server/lib/transcoding/default-transcoding-profiles'
import { isAbleToUploadVideo } from '@server/lib/user'
-import { buildFileMetadata, removeHLSPlaylist, removeWebTorrentFile } from '@server/lib/video-file'
import { VideoPathManager } from '@server/lib/video-path-manager'
-import { approximateIntroOutroAdditionalSize, safeCleanupStudioTMPFiles } from '@server/lib/video-studio'
+import { approximateIntroOutroAdditionalSize, onVideoEditionEnded, safeCleanupStudioTMPFiles } from '@server/lib/video-studio'
import { UserModel } from '@server/models/user/user'
import { VideoModel } from '@server/models/video/video'
-import { VideoFileModel } from '@server/models/video/video-file'
-import { MVideo, MVideoFile, MVideoFullLight, MVideoId, MVideoWithAllFiles } from '@server/types/models'
-import { getLowercaseExtension, pick } from '@shared/core-utils'
-import { buildUUID, getFileSize } from '@shared/extra-utils'
-import { FFmpegEdition, ffprobePromise, getVideoStreamDimensionsInfo, getVideoStreamDuration, getVideoStreamFPS } from '@shared/ffmpeg'
+import { MVideo, MVideoFullLight } from '@server/types/models'
+import { pick } from '@shared/core-utils'
+import { buildUUID } from '@shared/extra-utils'
+import { FFmpegEdition } from '@shared/ffmpeg'
import {
VideoStudioEditionPayload,
VideoStudioTask,
if (!video) {
logger.info('Can\'t process job %d, video does not exist.', job.id, lTags)
- await safeCleanupStudioTMPFiles(payload)
+ await safeCleanupStudioTMPFiles(payload.tasks)
return undefined
}
logger.info('Video edition ended for video %s.', video.uuid, lTags)
- const newFile = await buildNewFile(video, editionResultPath)
-
- const outputPath = VideoPathManager.Instance.getFSVideoFileOutputPath(video, newFile)
- await move(editionResultPath, outputPath)
-
- await safeCleanupStudioTMPFiles(payload)
-
- await createTorrentAndSetInfoHashFromPath(video, newFile, outputPath)
- await removeAllFiles(video, newFile)
-
- await newFile.save()
-
- video.duration = await getVideoStreamDuration(outputPath)
- await video.save()
-
- await federateVideoIfNeeded(video, false, undefined)
-
- const user = await UserModel.loadByVideoId(video.id)
-
- await createOptimizeOrMergeAudioJobs({ video, videoFile: newFile, isNewVideo: false, user, videoFileAlreadyLocked: false })
+ await onVideoEditionEnded({ video, editionResultPath, tasks: payload.tasks })
} catch (err) {
- await safeCleanupStudioTMPFiles(payload)
+ await safeCleanupStudioTMPFiles(payload.tasks)
throw err
}
watermarkPath: task.options.file,
videoFilters: {
- watermarkSizeRatio: VIDEO_FILTERS.WATERMARK.SIZE_RATIO,
- horitonzalMarginRatio: VIDEO_FILTERS.WATERMARK.HORIZONTAL_MARGIN_RATIO,
- verticalMarginRatio: VIDEO_FILTERS.WATERMARK.VERTICAL_MARGIN_RATIO
+ watermarkSizeRatio: task.options.watermarkSizeRatio,
+ horitonzalMarginRatio: task.options.horitonzalMarginRatio,
+ verticalMarginRatio: task.options.verticalMarginRatio
}
})
}
// ---------------------------------------------------------------------------
-async function buildNewFile (video: MVideoId, path: string) {
- const videoFile = new VideoFileModel({
- extname: getLowercaseExtension(path),
- size: await getFileSize(path),
- metadata: await buildFileMetadata(path),
- videoStreamingPlaylistId: null,
- videoId: video.id
- })
-
- const probe = await ffprobePromise(path)
-
- videoFile.fps = await getVideoStreamFPS(path, probe)
- videoFile.resolution = (await getVideoStreamDimensionsInfo(path, probe)).resolution
-
- videoFile.filename = generateWebTorrentVideoFilename(videoFile.resolution, videoFile.extname)
-
- return videoFile
-}
-
-async function removeAllFiles (video: MVideoWithAllFiles, webTorrentFileException: MVideoFile) {
- await removeHLSPlaylist(video)
-
- for (const file of video.VideoFiles) {
- if (file.id === webTorrentFileException.id) continue
-
- await removeWebTorrentFile(video, file.id)
- }
-}
-
async function checkUserQuotaOrThrow (video: MVideoFullLight, payload: VideoStudioEditionPayload) {
const user = await UserModel.loadByVideoId(video.id)
+import { throttle } from 'lodash'
import { retryTransactionWrapper } from '@server/helpers/database-utils'
import { logger, loggerTagsFactory } from '@server/helpers/logger'
import { RUNNER_JOBS } from '@server/initializers/constants'
RunnerJobSuccessPayload,
RunnerJobType,
RunnerJobUpdatePayload,
+ RunnerJobVideoEditionTranscodingPayload,
+ RunnerJobVideoEditionTranscodingPrivatePayload,
RunnerJobVODAudioMergeTranscodingPayload,
RunnerJobVODAudioMergeTranscodingPrivatePayload,
RunnerJobVODHLSTranscodingPayload,
RunnerJobVODWebVideoTranscodingPayload,
RunnerJobVODWebVideoTranscodingPrivatePayload
} from '@shared/models'
-import { throttle } from 'lodash'
type CreateRunnerJobArg =
{
type: Extract<RunnerJobType, 'live-rtmp-hls-transcoding'>
payload: RunnerJobLiveRTMPHLSTranscodingPayload
privatePayload: RunnerJobLiveRTMPHLSTranscodingPrivatePayload
+ } |
+ {
+ type: Extract<RunnerJobType, 'video-edition-transcoding'>
+ payload: RunnerJobVideoEditionTranscodingPayload
+ privatePayload: RunnerJobVideoEditionTranscodingPrivatePayload
}
export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S extends RunnerJobSuccessPayload> {
}): Promise<MRunnerJob> {
const { priority, dependsOnRunnerJob } = options
+ logger.debug('Creating runner job', { options, ...this.lTags(options.type) })
+
const runnerJob = new RunnerJobModel({
...pick(options, [ 'type', 'payload', 'privatePayload' ]),
import { moveToFailedTranscodingState, moveToNextState } from '@server/lib/video-state'
import { VideoJobInfoModel } from '@server/models/video/video-job-info'
import { MRunnerJob } from '@server/types/models/runners'
-import {
- LiveRTMPHLSTranscodingUpdatePayload,
- RunnerJobSuccessPayload,
- RunnerJobUpdatePayload,
- RunnerJobVODPrivatePayload
-} from '@shared/models'
+import { RunnerJobSuccessPayload, RunnerJobUpdatePayload, RunnerJobVODPrivatePayload } from '@shared/models'
import { AbstractJobHandler } from './abstract-job-handler'
import { loadTranscodingRunnerVideo } from './shared'
// eslint-disable-next-line max-len
export abstract class AbstractVODTranscodingJobHandler <C, U extends RunnerJobUpdatePayload, S extends RunnerJobSuccessPayload> extends AbstractJobHandler<C, U, S> {
- // ---------------------------------------------------------------------------
-
protected isAbortSupported () {
return true
}
protected specificUpdate (_options: {
runnerJob: MRunnerJob
- updatePayload?: LiveRTMPHLSTranscodingUpdatePayload
}) {
// empty
}
export * from './abstract-job-handler'
export * from './live-rtmp-hls-transcoding-job-handler'
+export * from './runner-job-handlers'
+export * from './video-edition-transcoding-job-handler'
export * from './vod-audio-merge-transcoding-job-handler'
export * from './vod-hls-transcoding-job-handler'
export * from './vod-web-video-transcoding-job-handler'
-export * from './runner-job-handlers'
// ---------------------------------------------------------------------------
- async specificUpdate (options: {
+ protected async specificUpdate (options: {
runnerJob: MRunnerJob
updatePayload: LiveRTMPHLSTranscodingUpdatePayload
}) {
import { RunnerJobSuccessPayload, RunnerJobType, RunnerJobUpdatePayload } from '@shared/models'
import { AbstractJobHandler } from './abstract-job-handler'
import { LiveRTMPHLSTranscodingJobHandler } from './live-rtmp-hls-transcoding-job-handler'
+import { VideoEditionTranscodingJobHandler } from './video-edition-transcoding-job-handler'
import { VODAudioMergeTranscodingJobHandler } from './vod-audio-merge-transcoding-job-handler'
import { VODHLSTranscodingJobHandler } from './vod-hls-transcoding-job-handler'
import { VODWebVideoTranscodingJobHandler } from './vod-web-video-transcoding-job-handler'
'vod-web-video-transcoding': VODWebVideoTranscodingJobHandler,
'vod-hls-transcoding': VODHLSTranscodingJobHandler,
'vod-audio-merge-transcoding': VODAudioMergeTranscodingJobHandler,
- 'live-rtmp-hls-transcoding': LiveRTMPHLSTranscodingJobHandler
+ 'live-rtmp-hls-transcoding': LiveRTMPHLSTranscodingJobHandler,
+ 'video-edition-transcoding': VideoEditionTranscodingJobHandler
}
export function getRunnerJobHandlerClass (job: MRunnerJob) {
--- /dev/null
+
+import { basename } from 'path'
+import { logger } from '@server/helpers/logger'
+import { onVideoEditionEnded, safeCleanupStudioTMPFiles } from '@server/lib/video-studio'
+import { MVideo } from '@server/types/models'
+import { MRunnerJob } from '@server/types/models/runners'
+import { buildUUID } from '@shared/extra-utils'
+import {
+ isVideoStudioTaskIntro,
+ isVideoStudioTaskOutro,
+ isVideoStudioTaskWatermark,
+ RunnerJobState,
+ RunnerJobUpdatePayload,
+ RunnerJobVideoEditionTranscodingPayload,
+ RunnerJobVideoEditionTranscodingPrivatePayload,
+ VideoEditionTranscodingSuccess,
+ VideoState,
+ VideoStudioTaskPayload
+} from '@shared/models'
+import { generateRunnerEditionTranscodingVideoInputFileUrl, generateRunnerTranscodingVideoInputFileUrl } from '../runner-urls'
+import { AbstractJobHandler } from './abstract-job-handler'
+import { loadTranscodingRunnerVideo } from './shared'
+
+type CreateOptions = {
+ video: MVideo
+ tasks: VideoStudioTaskPayload[]
+ priority: number
+}
+
+// eslint-disable-next-line max-len
+export class VideoEditionTranscodingJobHandler extends AbstractJobHandler<CreateOptions, RunnerJobUpdatePayload, VideoEditionTranscodingSuccess> {
+
+ async create (options: CreateOptions) {
+ const { video, priority, tasks } = options
+
+ const jobUUID = buildUUID()
+ const payload: RunnerJobVideoEditionTranscodingPayload = {
+ input: {
+ videoFileUrl: generateRunnerTranscodingVideoInputFileUrl(jobUUID, video.uuid)
+ },
+ tasks: tasks.map(t => {
+ if (isVideoStudioTaskIntro(t) || isVideoStudioTaskOutro(t)) {
+ return {
+ ...t,
+
+ options: {
+ ...t.options,
+
+ file: generateRunnerEditionTranscodingVideoInputFileUrl(jobUUID, video.uuid, basename(t.options.file))
+ }
+ }
+ }
+
+ if (isVideoStudioTaskWatermark(t)) {
+ return {
+ ...t,
+
+ options: {
+ ...t.options,
+
+ file: generateRunnerEditionTranscodingVideoInputFileUrl(jobUUID, video.uuid, basename(t.options.file))
+ }
+ }
+ }
+
+ return t
+ })
+ }
+
+ const privatePayload: RunnerJobVideoEditionTranscodingPrivatePayload = {
+ videoUUID: video.uuid,
+ originalTasks: tasks
+ }
+
+ const job = await this.createRunnerJob({
+ type: 'video-edition-transcoding',
+ jobUUID,
+ payload,
+ privatePayload,
+ priority
+ })
+
+ return job
+ }
+
+ // ---------------------------------------------------------------------------
+
+ protected isAbortSupported () {
+ return true
+ }
+
+ protected specificUpdate (_options: {
+ runnerJob: MRunnerJob
+ }) {
+ // empty
+ }
+
+ protected specificAbort (_options: {
+ runnerJob: MRunnerJob
+ }) {
+ // empty
+ }
+
+ protected async specificComplete (options: {
+ runnerJob: MRunnerJob
+ resultPayload: VideoEditionTranscodingSuccess
+ }) {
+ const { runnerJob, resultPayload } = options
+ const privatePayload = runnerJob.privatePayload as RunnerJobVideoEditionTranscodingPrivatePayload
+
+ const video = await loadTranscodingRunnerVideo(runnerJob, this.lTags)
+ if (!video) {
+ await safeCleanupStudioTMPFiles(privatePayload.originalTasks)
+
+ }
+
+ const videoFilePath = resultPayload.videoFile as string
+
+ await onVideoEditionEnded({ video, editionResultPath: videoFilePath, tasks: privatePayload.originalTasks })
+
+ logger.info(
+ 'Runner video edition transcoding job %s for %s ended.',
+ runnerJob.uuid, video.uuid, this.lTags(video.uuid, runnerJob.uuid)
+ )
+ }
+
+ protected specificError (options: {
+ runnerJob: MRunnerJob
+ nextState: RunnerJobState
+ }) {
+ if (options.nextState === RunnerJobState.ERRORED) {
+ return this.specificErrorOrCancel(options)
+ }
+
+ return Promise.resolve()
+ }
+
+ protected specificCancel (options: {
+ runnerJob: MRunnerJob
+ }) {
+ return this.specificErrorOrCancel(options)
+ }
+
+ private async specificErrorOrCancel (options: {
+ runnerJob: MRunnerJob
+ }) {
+ const { runnerJob } = options
+
+ const payload = runnerJob.privatePayload as RunnerJobVideoEditionTranscodingPrivatePayload
+ await safeCleanupStudioTMPFiles(payload.originalTasks)
+
+ const video = await loadTranscodingRunnerVideo(options.runnerJob, this.lTags)
+ if (!video) return
+
+ return video.setNewState(VideoState.PUBLISHED, false, undefined)
+ }
+}
// ---------------------------------------------------------------------------
- async specificComplete (options: {
+ protected async specificComplete (options: {
runnerJob: MRunnerJob
resultPayload: VODAudioMergeTranscodingSuccess
}) {
// ---------------------------------------------------------------------------
- async specificComplete (options: {
+ protected async specificComplete (options: {
runnerJob: MRunnerJob
resultPayload: VODHLSTranscodingSuccess
}) {
// ---------------------------------------------------------------------------
- async specificComplete (options: {
+ protected async specificComplete (options: {
runnerJob: MRunnerJob
resultPayload: VODWebVideoTranscodingSuccess
}) {
export function generateRunnerTranscodingVideoPreviewFileUrl (jobUUID: string, videoUUID: string) {
return WEBSERVER.URL + '/api/v1/runners/jobs/' + jobUUID + '/files/videos/' + videoUUID + '/previews/max-quality'
}
+
+export function generateRunnerEditionTranscodingVideoInputFileUrl (jobUUID: string, videoUUID: string, filename: string) {
+ return WEBSERVER.URL + '/api/v1/runners/jobs/' + jobUUID + '/files/videos/' + videoUUID + '/studio/task-files/' + filename
+}
}
},
videoStudio: {
- enabled: CONFIG.VIDEO_STUDIO.ENABLED
+ enabled: CONFIG.VIDEO_STUDIO.ENABLED,
+ remoteRunners: {
+ enabled: CONFIG.VIDEO_STUDIO.REMOTE_RUNNERS.ENABLED
+ }
},
import: {
videos: {
-import { JOB_PRIORITY } from '@server/initializers/constants'
-import { VideoModel } from '@server/models/video/video'
import { MUserId, MVideoFile, MVideoFullLight } from '@server/types/models'
export abstract class AbstractJobBuilder {
isNewVideo: boolean
user: MUserId | null
}): Promise<any>
-
- protected async getTranscodingJobPriority (options: {
- user: MUserId
- fallback: number
- }) {
- const { user, fallback } = options
-
- if (!user) return fallback
-
- const now = new Date()
- const lastWeek = new Date(now.getFullYear(), now.getMonth(), now.getDate() - 7)
-
- const videoUploadedByUser = await VideoModel.countVideosUploadedByUserSince(user.id, lastWeek)
-
- return JOB_PRIORITY.TRANSCODING + videoUploadedByUser
- }
}
OptimizeTranscodingPayload,
VideoTranscodingPayload
} from '@shared/models'
+import { getTranscodingJobPriority } from '../../transcoding-priority'
import { canDoQuickTranscode } from '../../transcoding-quick-transcode'
import { computeResolutionsToTranscode } from '../../transcoding-resolutions'
import { AbstractJobBuilder } from './abstract-job-builder'
return {
type: 'video-transcoding' as 'video-transcoding',
- priority: await this.getTranscodingJobPriority({ user, fallback: undefined }),
+ priority: await getTranscodingJobPriority({ user, type: 'vod', fallback: undefined }),
payload
}
}
import { MUserId, MVideoFile, MVideoFullLight, MVideoWithFileThumbnail } from '@server/types/models'
import { MRunnerJob } from '@server/types/models/runners'
import { ffprobePromise, getVideoStreamDimensionsInfo, getVideoStreamFPS, hasAudioStream, isAudioFile } from '@shared/ffmpeg'
+import { getTranscodingJobPriority } from '../../transcoding-priority'
import { computeResolutionsToTranscode } from '../../transcoding-resolutions'
import { AbstractJobBuilder } from './abstract-job-builder'
: resolution
const fps = computeOutputFPS({ inputFPS, resolution: maxResolution })
- const priority = await this.getTranscodingJobPriority({ user, fallback: 0 })
+ const priority = await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
const mainRunnerJob = videoFile.isAudio()
? await new VODAudioMergeTranscodingJobHandler().create({ video, resolution: maxResolution, fps, isNewVideo, priority })
fps,
isNewVideo,
dependsOnRunnerJob: mainRunnerJob,
- priority: await this.getTranscodingJobPriority({ user, fallback: 0 })
+ priority: await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
})
}
const maxResolution = Math.max(...resolutions)
const { fps: inputFPS } = await video.probeMaxQualityFile()
const maxFPS = computeOutputFPS({ inputFPS, resolution: maxResolution })
- const priority = await this.getTranscodingJobPriority({ user, fallback: 0 })
+ const priority = await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
const childrenResolutions = resolutions.filter(r => r !== maxResolution)
isNewVideo,
deleteWebVideoFiles: false,
dependsOnRunnerJob,
- priority: await this.getTranscodingJobPriority({ user, fallback: 0 })
+ priority: await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
})
continue
}
fps,
isNewVideo,
dependsOnRunnerJob,
- priority: await this.getTranscodingJobPriority({ user, fallback: 0 })
+ priority: await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
})
continue
}
fps,
isNewVideo,
dependsOnRunnerJob: mainRunnerJob,
- priority: await this.getTranscodingJobPriority({ user, fallback: 0 })
+ priority: await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
})
}
isNewVideo,
deleteWebVideoFiles: false,
dependsOnRunnerJob: mainRunnerJob,
- priority: await this.getTranscodingJobPriority({ user, fallback: 0 })
+ priority: await getTranscodingJobPriority({ user, type: 'vod', fallback: 0 })
})
}
}
--- /dev/null
+import { JOB_PRIORITY } from '@server/initializers/constants'
+import { VideoModel } from '@server/models/video/video'
+import { MUserId } from '@server/types/models'
+
+export async function getTranscodingJobPriority (options: {
+ user: MUserId
+ fallback: number
+ type: 'vod' | 'studio'
+}) {
+ const { user, fallback, type } = options
+
+ if (!user) return fallback
+
+ const now = new Date()
+ const lastWeek = new Date(now.getFullYear(), now.getMonth(), now.getDate() - 7)
+
+ const videoUploadedByUser = await VideoModel.countVideosUploadedByUserSince(user.id, lastWeek)
+
+ const base = type === 'vod'
+ ? JOB_PRIORITY.TRANSCODING
+ : JOB_PRIORITY.VIDEO_STUDIO
+
+ return base + videoUploadedByUser
+}
-import { logger } from '@server/helpers/logger'
-import { MVideoFullLight } from '@server/types/models'
+import { move, remove } from 'fs-extra'
+import { join } from 'path'
+import { logger, loggerTagsFactory } from '@server/helpers/logger'
+import { createTorrentAndSetInfoHashFromPath } from '@server/helpers/webtorrent'
+import { CONFIG } from '@server/initializers/config'
+import { UserModel } from '@server/models/user/user'
+import { MUser, MVideo, MVideoFile, MVideoFullLight, MVideoWithAllFiles } from '@server/types/models'
import { getVideoStreamDuration } from '@shared/ffmpeg'
-import { VideoStudioEditionPayload, VideoStudioTask } from '@shared/models'
-import { remove } from 'fs-extra'
+import { VideoStudioEditionPayload, VideoStudioTask, VideoStudioTaskPayload } from '@shared/models'
+import { federateVideoIfNeeded } from './activitypub/videos'
+import { JobQueue } from './job-queue'
+import { VideoEditionTranscodingJobHandler } from './runners'
+import { createOptimizeOrMergeAudioJobs } from './transcoding/create-transcoding-job'
+import { getTranscodingJobPriority } from './transcoding/transcoding-priority'
+import { buildNewFile, removeHLSPlaylist, removeWebTorrentFile } from './video-file'
+import { VideoPathManager } from './video-path-manager'
-function buildTaskFileFieldname (indice: number, fieldName = 'file') {
+const lTags = loggerTagsFactory('video-edition')
+
+export function buildTaskFileFieldname (indice: number, fieldName = 'file') {
return `tasks[${indice}][options][${fieldName}]`
}
-function getTaskFileFromReq (files: Express.Multer.File[], indice: number, fieldName = 'file') {
+export function getTaskFileFromReq (files: Express.Multer.File[], indice: number, fieldName = 'file') {
return files.find(f => f.fieldname === buildTaskFileFieldname(indice, fieldName))
}
-async function safeCleanupStudioTMPFiles (payload: VideoStudioEditionPayload) {
- for (const task of payload.tasks) {
+export function getStudioTaskFilePath (filename: string) {
+ return join(CONFIG.STORAGE.TMP_PERSISTENT_DIR, filename)
+}
+
+export async function safeCleanupStudioTMPFiles (tasks: VideoStudioTaskPayload[]) {
+ logger.info('Removing studio task files', { tasks, ...lTags() })
+
+ for (const task of tasks) {
try {
if (task.name === 'add-intro' || task.name === 'add-outro') {
await remove(task.options.file)
}
}
-async function approximateIntroOutroAdditionalSize (video: MVideoFullLight, tasks: VideoStudioTask[], fileFinder: (i: number) => string) {
+// ---------------------------------------------------------------------------
+
+export async function approximateIntroOutroAdditionalSize (
+ video: MVideoFullLight,
+ tasks: VideoStudioTask[],
+ fileFinder: (i: number) => string
+) {
let additionalDuration = 0
for (let i = 0; i < tasks.length; i++) {
return (video.getMaxQualityFile().size / video.duration) * additionalDuration
}
-export {
- approximateIntroOutroAdditionalSize,
- buildTaskFileFieldname,
- getTaskFileFromReq,
- safeCleanupStudioTMPFiles
+// ---------------------------------------------------------------------------
+
+export async function createVideoStudioJob (options: {
+ video: MVideo
+ user: MUser
+ payload: VideoStudioEditionPayload
+}) {
+ const { video, user, payload } = options
+
+ const priority = await getTranscodingJobPriority({ user, type: 'studio', fallback: 0 })
+
+ if (CONFIG.VIDEO_STUDIO.REMOTE_RUNNERS.ENABLED) {
+ await new VideoEditionTranscodingJobHandler().create({ video, tasks: payload.tasks, priority })
+ return
+ }
+
+ await JobQueue.Instance.createJob({ type: 'video-studio-edition', payload, priority })
+}
+
+export async function onVideoEditionEnded (options: {
+ editionResultPath: string
+ tasks: VideoStudioTaskPayload[]
+ video: MVideoFullLight
+}) {
+ const { video, tasks, editionResultPath } = options
+
+ const newFile = await buildNewFile({ path: editionResultPath, mode: 'web-video' })
+ newFile.videoId = video.id
+
+ const outputPath = VideoPathManager.Instance.getFSVideoFileOutputPath(video, newFile)
+ await move(editionResultPath, outputPath)
+
+ await safeCleanupStudioTMPFiles(tasks)
+
+ await createTorrentAndSetInfoHashFromPath(video, newFile, outputPath)
+ await removeAllFiles(video, newFile)
+
+ await newFile.save()
+
+ video.duration = await getVideoStreamDuration(outputPath)
+ await video.save()
+
+ await federateVideoIfNeeded(video, false, undefined)
+
+ const user = await UserModel.loadByVideoId(video.id)
+
+ await createOptimizeOrMergeAudioJobs({ video, videoFile: newFile, isNewVideo: false, user, videoFileAlreadyLocked: false })
+}
+
+// ---------------------------------------------------------------------------
+// Private
+// ---------------------------------------------------------------------------
+
+async function removeAllFiles (video: MVideoWithAllFiles, webTorrentFileException: MVideoFile) {
+ await removeHLSPlaylist(video)
+
+ for (const file of video.VideoFiles) {
+ if (file.id === webTorrentFileException.id) continue
+
+ await removeWebTorrentFile(video, file.id)
+ }
}
body('transcoding.hls.enabled').isBoolean(),
body('videoStudio.enabled').isBoolean(),
+ body('videoStudio.remoteRunners.enabled').isBoolean(),
body('import.videos.concurrency').isInt({ min: 0 }),
body('import.videos.http.enabled').isBoolean(),
import express from 'express'
-import { HttpStatusCode } from '@shared/models'
+import { param } from 'express-validator'
+import { basename } from 'path'
+import { isSafeFilename } from '@server/helpers/custom-validators/misc'
+import { hasVideoStudioTaskFile, HttpStatusCode, RunnerJobVideoEditionTranscodingPayload } from '@shared/models'
import { areValidationErrors, doesVideoExist, isValidVideoIdParam } from '../shared'
const tags = [ 'runner' ]
return next()
}
]
+
+export const runnerJobGetVideoStudioTaskFileValidator = [
+ param('filename').custom(v => isSafeFilename(v)),
+
+ (req: express.Request, res: express.Response, next: express.NextFunction) => {
+ if (areValidationErrors(req, res)) return
+
+ const filename = req.params.filename
+
+ const payload = res.locals.runnerJob.payload as RunnerJobVideoEditionTranscodingPayload
+
+ const found = Array.isArray(payload?.tasks) && payload.tasks.some(t => {
+ if (hasVideoStudioTaskFile(t)) {
+ return basename(t.options.file) === filename
+ }
+
+ return false
+ })
+
+ if (!found) {
+ return res.fail({
+ status: HttpStatusCode.BAD_REQUEST_400,
+ message: 'File is not associated to this edition task',
+ tags: [ ...tags, res.locals.videoAll.uuid ]
+ })
+ }
+
+ return next()
+ }
+]
}
]
+export const cancelRunnerJobValidator = [
+ (req: express.Request, res: express.Response, next: express.NextFunction) => {
+ const runnerJob = res.locals.runnerJob
+
+ const allowedStates = new Set<RunnerJobState>([
+ RunnerJobState.PENDING,
+ RunnerJobState.PROCESSING,
+ RunnerJobState.WAITING_FOR_PARENT_JOB
+ ])
+
+ if (allowedStates.has(runnerJob.state) !== true) {
+ return res.fail({
+ status: HttpStatusCode.BAD_REQUEST_400,
+ message: 'Cannot cancel this job that is not in "pending", "processing" or "waiting for parent job" state',
+ tags
+ })
+ }
+
+ return next()
+ }
+]
+
export const runnerJobGetValidator = [
param('jobUUID').custom(isUUIDValid),
}
},
videoStudio: {
- enabled: true
+ enabled: true,
+ remoteRunners: {
+ enabled: true
+ }
},
import: {
videos: {
+import { basename } from 'path'
/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
import { checkBadCountPagination, checkBadSortPagination, checkBadStartPagination } from '@server/tests/shared'
-import { HttpStatusCode, RunnerJob, RunnerJobState, RunnerJobSuccessPayload, RunnerJobUpdatePayload, VideoPrivacy } from '@shared/models'
+import {
+ HttpStatusCode,
+ isVideoStudioTaskIntro,
+ RunnerJob,
+ RunnerJobState,
+ RunnerJobSuccessPayload,
+ RunnerJobUpdatePayload,
+ RunnerJobVideoEditionTranscodingPayload,
+ VideoPrivacy,
+ VideoStudioTaskIntro
+} from '@shared/models'
import {
cleanupTests,
createSingleServer,
setAccessTokensToServers,
setDefaultVideoChannel,
stopFfmpeg,
+ VideoStudioCommand,
waitJobs
} from '@shared/server-commands'
registrationTokenId = data[0].id
await server.config.enableTranscoding(true, true)
+ await server.config.enableStudio()
await server.config.enableRemoteTranscoding()
+ await server.config.enableRemoteStudio()
+
runnerToken = await server.runners.autoRegisterRunner()
runnerToken2 = await server.runners.autoRegisterRunner()
await server.runnerJobs.cancelByAdmin({ jobUUID, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
})
+ it('Should fail with an already cancelled job', async function () {
+ await server.runnerJobs.cancelByAdmin({ jobUUID: cancelledJobUUID, expectedStatus: HttpStatusCode.BAD_REQUEST_400 })
+ })
+
it('Should succeed with the correct params', async function () {
await server.runnerJobs.cancelByAdmin({ jobUUID })
})
let pendingUUID: string
+ let videoStudioUUID: string
+ let studioFile: string
+
let liveAcceptedJob: RunnerJob & { jobToken: string }
+ let studioAcceptedJob: RunnerJob & { jobToken: string }
- async function fetchFiles (options: {
+ async function fetchVideoInputFiles (options: {
jobUUID: string
videoUUID: string
runnerToken: string
}
}
+ async function fetchStudioFiles (options: {
+ jobUUID: string
+ videoUUID: string
+ runnerToken: string
+ jobToken: string
+ studioFile?: string
+ expectedStatus: HttpStatusCode
+ }) {
+ const { jobUUID, expectedStatus, videoUUID, runnerToken, jobToken, studioFile } = options
+
+ const path = `/api/v1/runners/jobs/${jobUUID}/files/videos/${videoUUID}/studio/task-files/${studioFile}`
+
+ await makePostBodyRequest({ url: server.url, path, fields: { runnerToken, jobToken }, expectedStatus })
+ }
+
before(async function () {
this.timeout(120000)
pendingUUID = availableJobs[0].uuid
}
+ {
+ await server.config.disableTranscoding()
+
+ const { uuid } = await server.videos.quickUpload({ name: 'video studio' })
+ videoStudioUUID = uuid
+
+ await server.config.enableTranscoding(true, true)
+ await server.config.enableStudio()
+
+ await server.videoStudio.createEditionTasks({
+ videoId: videoStudioUUID,
+ tasks: VideoStudioCommand.getComplexTask()
+ })
+
+ const { job } = await server.runnerJobs.autoAccept({ runnerToken, type: 'video-edition-transcoding' })
+ studioAcceptedJob = job
+
+ const tasks = (job.payload as RunnerJobVideoEditionTranscodingPayload).tasks
+ const fileUrl = (tasks.find(t => isVideoStudioTaskIntro(t)) as VideoStudioTaskIntro).options.file as string
+ studioFile = basename(fileUrl)
+ }
+
{
await server.config.enableLive({
allowReplay: false,
jobToken: string
expectedStatus: HttpStatusCode
}) {
- await fetchFiles({ ...options, videoUUID })
-
await server.runnerJobs.abort({ ...options, reason: 'reason' })
await server.runnerJobs.update({ ...options })
await server.runnerJobs.error({ ...options, message: 'message' })
}
it('Should fail with an invalid job uuid', async function () {
- await testEndpoints({ jobUUID: 'a', runnerToken, jobToken, expectedStatus: HttpStatusCode.BAD_REQUEST_400 })
+ const options = { jobUUID: 'a', runnerToken, expectedStatus: HttpStatusCode.BAD_REQUEST_400 }
+
+ await testEndpoints({ ...options, jobToken })
+ await fetchVideoInputFiles({ ...options, videoUUID, jobToken })
+ await fetchStudioFiles({ ...options, videoUUID, jobToken: studioAcceptedJob.jobToken, studioFile })
})
it('Should fail with an unknown job uuid', async function () {
- const jobUUID = badUUID
- await testEndpoints({ jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+ const options = { jobUUID: badUUID, runnerToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 }
+
+ await testEndpoints({ ...options, jobToken })
+ await fetchVideoInputFiles({ ...options, videoUUID, jobToken })
+ await fetchStudioFiles({ ...options, jobToken: studioAcceptedJob.jobToken, videoUUID, studioFile })
})
it('Should fail with an invalid runner token', async function () {
- await testEndpoints({ jobUUID, runnerToken: '', jobToken, expectedStatus: HttpStatusCode.BAD_REQUEST_400 })
+ const options = { runnerToken: '', expectedStatus: HttpStatusCode.BAD_REQUEST_400 }
+
+ await testEndpoints({ ...options, jobUUID, jobToken })
+ await fetchVideoInputFiles({ ...options, jobUUID, videoUUID, jobToken })
+ await fetchStudioFiles({
+ ...options,
+ jobToken: studioAcceptedJob.jobToken,
+ jobUUID: studioAcceptedJob.uuid,
+ videoUUID: videoStudioUUID,
+ studioFile
+ })
})
it('Should fail with an unknown runner token', async function () {
- const runnerToken = badUUID
- await testEndpoints({ jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+ const options = { runnerToken: badUUID, expectedStatus: HttpStatusCode.NOT_FOUND_404 }
+
+ await testEndpoints({ ...options, jobUUID, jobToken })
+ await fetchVideoInputFiles({ ...options, jobUUID, videoUUID, jobToken })
+ await fetchStudioFiles({
+ ...options,
+ jobToken: studioAcceptedJob.jobToken,
+ jobUUID: studioAcceptedJob.uuid,
+ videoUUID: videoStudioUUID,
+ studioFile
+ })
})
it('Should fail with an invalid job token job uuid', async function () {
- await testEndpoints({ jobUUID, runnerToken, jobToken: '', expectedStatus: HttpStatusCode.BAD_REQUEST_400 })
+ const options = { runnerToken, jobToken: '', expectedStatus: HttpStatusCode.BAD_REQUEST_400 }
+
+ await testEndpoints({ ...options, jobUUID })
+ await fetchVideoInputFiles({ ...options, jobUUID, videoUUID })
+ await fetchStudioFiles({ ...options, jobUUID: studioAcceptedJob.uuid, videoUUID: videoStudioUUID, studioFile })
})
it('Should fail with an unknown job token job uuid', async function () {
- const jobToken = badUUID
- await testEndpoints({ jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+ const options = { runnerToken, jobToken: badUUID, expectedStatus: HttpStatusCode.NOT_FOUND_404 }
+
+ await testEndpoints({ ...options, jobUUID })
+ await fetchVideoInputFiles({ ...options, jobUUID, videoUUID })
+ await fetchStudioFiles({ ...options, jobUUID: studioAcceptedJob.uuid, videoUUID: videoStudioUUID, studioFile })
})
it('Should fail with a runner token not associated to this job', async function () {
- await testEndpoints({ jobUUID, runnerToken: runnerToken2, jobToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+ const options = { runnerToken: runnerToken2, expectedStatus: HttpStatusCode.NOT_FOUND_404 }
+
+ await testEndpoints({ ...options, jobUUID, jobToken })
+ await fetchVideoInputFiles({ ...options, jobUUID, videoUUID, jobToken })
+ await fetchStudioFiles({
+ ...options,
+ jobToken: studioAcceptedJob.jobToken,
+ jobUUID: studioAcceptedJob.uuid,
+ videoUUID: videoStudioUUID,
+ studioFile
+ })
})
it('Should fail with a job uuid not associated to the job token', async function () {
- await testEndpoints({ jobUUID: jobUUID2, runnerToken, jobToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
- await testEndpoints({ jobUUID, runnerToken, jobToken: jobToken2, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+ {
+ const options = { jobUUID: jobUUID2, runnerToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 }
+
+ await testEndpoints({ ...options, jobToken })
+ await fetchVideoInputFiles({ ...options, jobToken, videoUUID })
+ await fetchStudioFiles({ ...options, jobToken: studioAcceptedJob.jobToken, videoUUID: videoStudioUUID, studioFile })
+ }
+
+ {
+ const options = { runnerToken, jobToken: jobToken2, expectedStatus: HttpStatusCode.NOT_FOUND_404 }
+
+ await testEndpoints({ ...options, jobUUID })
+ await fetchVideoInputFiles({ ...options, jobUUID, videoUUID })
+ await fetchStudioFiles({ ...options, jobUUID: studioAcceptedJob.uuid, videoUUID: videoStudioUUID, studioFile })
+ }
})
})
})
})
})
+
+ describe('Video studio', function () {
+
+ it('Should fail with an invalid video edition transcoding payload', async function () {
+ await server.runnerJobs.success({
+ jobUUID: studioAcceptedJob.uuid,
+ jobToken: studioAcceptedJob.jobToken,
+ payload: { hello: 'video_short.mp4' } as any,
+ runnerToken,
+ expectedStatus: HttpStatusCode.BAD_REQUEST_400
+ })
+ })
+ })
})
describe('Job files', function () {
- describe('Video files', function () {
+ describe('Check video param for common job file routes', function () {
+
+ async function fetchFiles (options: {
+ videoUUID?: string
+ expectedStatus: HttpStatusCode
+ }) {
+ await fetchVideoInputFiles({ videoUUID, ...options, jobToken, jobUUID, runnerToken })
+
+ await fetchStudioFiles({
+ videoUUID: videoStudioUUID,
+
+ ...options,
+
+ jobToken: studioAcceptedJob.jobToken,
+ jobUUID: studioAcceptedJob.uuid,
+ runnerToken,
+ studioFile
+ })
+ }
it('Should fail with an invalid video id', async function () {
- await fetchFiles({ videoUUID: 'a', jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.BAD_REQUEST_400 })
+ await fetchFiles({
+ videoUUID: 'a',
+ expectedStatus: HttpStatusCode.BAD_REQUEST_400
+ })
})
it('Should fail with an unknown video id', async function () {
const videoUUID = '910ec12a-d9e6-458b-a274-0abb655f9464'
- await fetchFiles({ videoUUID, jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+
+ await fetchFiles({
+ videoUUID,
+ expectedStatus: HttpStatusCode.NOT_FOUND_404
+ })
})
it('Should fail with a video id not associated to this job', async function () {
- await fetchFiles({ videoUUID: videoUUID2, jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.FORBIDDEN_403 })
+ await fetchFiles({
+ videoUUID: videoUUID2,
+ expectedStatus: HttpStatusCode.FORBIDDEN_403
+ })
})
it('Should succeed with the correct params', async function () {
- await fetchFiles({ videoUUID, jobUUID, runnerToken, jobToken, expectedStatus: HttpStatusCode.OK_200 })
+ await fetchFiles({ expectedStatus: HttpStatusCode.OK_200 })
+ })
+ })
+
+ describe('Video edition tasks file routes', function () {
+
+ it('Should fail with an invalid studio filename', async function () {
+ await fetchStudioFiles({
+ videoUUID: videoStudioUUID,
+ jobUUID: studioAcceptedJob.uuid,
+ runnerToken,
+ jobToken: studioAcceptedJob.jobToken,
+ studioFile: 'toto',
+ expectedStatus: HttpStatusCode.BAD_REQUEST_400
+ })
})
})
})
export * from './runner-common'
export * from './runner-live-transcoding'
export * from './runner-socket'
+export * from './runner-studio-transcoding'
export * from './runner-vod-transcoding'
import { expect } from 'chai'
import { wait } from '@shared/core-utils'
-import { HttpStatusCode, Runner, RunnerJob, RunnerJobAdmin, RunnerJobState, RunnerRegistrationToken } from '@shared/models'
+import {
+ HttpStatusCode,
+ Runner,
+ RunnerJob,
+ RunnerJobAdmin,
+ RunnerJobState,
+ RunnerJobVODWebVideoTranscodingPayload,
+ RunnerRegistrationToken
+} from '@shared/models'
import {
cleanupTests,
createSingleServer,
for (const job of availableJobs) {
expect(job.uuid).to.exist
expect(job.payload.input).to.exist
- expect(job.payload.output).to.exist
+ expect((job.payload as RunnerJobVODWebVideoTranscodingPayload).output).to.exist
expect((job as RunnerJobAdmin).privatePayload).to.not.exist
}
--- /dev/null
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { readFile } from 'fs-extra'
+import { checkPersistentTmpIsEmpty, checkVideoDuration } from '@server/tests/shared'
+import { buildAbsoluteFixturePath } from '@shared/core-utils'
+import {
+ RunnerJobVideoEditionTranscodingPayload,
+ VideoEditionTranscodingSuccess,
+ VideoState,
+ VideoStudioTask,
+ VideoStudioTaskIntro
+} from '@shared/models'
+import {
+ cleanupTests,
+ createMultipleServers,
+ doubleFollow,
+ PeerTubeServer,
+ setAccessTokensToServers,
+ setDefaultVideoChannel,
+ VideoStudioCommand,
+ waitJobs
+} from '@shared/server-commands'
+
+describe('Test runner video studio transcoding', function () {
+ let servers: PeerTubeServer[] = []
+ let runnerToken: string
+ let videoUUID: string
+ let jobUUID: string
+
+ async function renewStudio (tasks: VideoStudioTask[] = VideoStudioCommand.getComplexTask()) {
+ const { uuid } = await servers[0].videos.quickUpload({ name: 'video' })
+ videoUUID = uuid
+
+ await waitJobs(servers)
+
+ await servers[0].videoStudio.createEditionTasks({ videoId: uuid, tasks })
+ await waitJobs(servers)
+
+ const { availableJobs } = await servers[0].runnerJobs.request({ runnerToken })
+ expect(availableJobs).to.have.lengthOf(1)
+
+ jobUUID = availableJobs[0].uuid
+ }
+
+ before(async function () {
+ this.timeout(120_000)
+
+ servers = await createMultipleServers(2)
+
+ await setAccessTokensToServers(servers)
+ await setDefaultVideoChannel(servers)
+
+ await doubleFollow(servers[0], servers[1])
+
+ await servers[0].config.enableTranscoding(true, true)
+ await servers[0].config.enableStudio()
+ await servers[0].config.enableRemoteStudio()
+
+ runnerToken = await servers[0].runners.autoRegisterRunner()
+ })
+
+ it('Should error a studio transcoding job', async function () {
+ this.timeout(60000)
+
+ await renewStudio()
+
+ for (let i = 0; i < 5; i++) {
+ const { job } = await servers[0].runnerJobs.accept({ runnerToken, jobUUID })
+ const jobToken = job.jobToken
+
+ await servers[0].runnerJobs.error({ runnerToken, jobUUID, jobToken, message: 'Error' })
+ }
+
+ const video = await servers[0].videos.get({ id: videoUUID })
+ expect(video.state.id).to.equal(VideoState.PUBLISHED)
+
+ await checkPersistentTmpIsEmpty(servers[0])
+ })
+
+ it('Should cancel a transcoding job', async function () {
+ this.timeout(60000)
+
+ await renewStudio()
+
+ await servers[0].runnerJobs.cancelByAdmin({ jobUUID })
+
+ const video = await servers[0].videos.get({ id: videoUUID })
+ expect(video.state.id).to.equal(VideoState.PUBLISHED)
+
+ await checkPersistentTmpIsEmpty(servers[0])
+ })
+
+ it('Should execute a remote studio job', async function () {
+ this.timeout(240_000)
+
+ const tasks = [
+ {
+ name: 'add-outro' as 'add-outro',
+ options: {
+ file: 'video_short.webm'
+ }
+ },
+ {
+ name: 'add-watermark' as 'add-watermark',
+ options: {
+ file: 'thumbnail.png'
+ }
+ },
+ {
+ name: 'add-intro' as 'add-intro',
+ options: {
+ file: 'video_very_short_240p.mp4'
+ }
+ }
+ ]
+
+ await renewStudio(tasks)
+
+ for (const server of servers) {
+ await checkVideoDuration(server, videoUUID, 5)
+ }
+
+ const { job } = await servers[0].runnerJobs.accept<RunnerJobVideoEditionTranscodingPayload>({ runnerToken, jobUUID })
+ const jobToken = job.jobToken
+
+ expect(job.type === 'video-edition-transcoding')
+ expect(job.payload.input.videoFileUrl).to.exist
+
+ // Check video input file
+ {
+ await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ }
+
+ // Check task files
+ for (let i = 0; i < tasks.length; i++) {
+ const task = tasks[i]
+ const payloadTask = job.payload.tasks[i]
+
+ expect(payloadTask.name).to.equal(task.name)
+
+ const inputFile = await readFile(buildAbsoluteFixturePath(task.options.file))
+
+ const { body } = await servers[0].runnerJobs.getJobFile({
+ url: (payloadTask as VideoStudioTaskIntro).options.file as string,
+ jobToken,
+ runnerToken
+ })
+
+ expect(body).to.deep.equal(inputFile)
+ }
+
+ const payload: VideoEditionTranscodingSuccess = { videoFile: 'video_very_short_240p.mp4' }
+ await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+ await waitJobs(servers)
+
+ for (const server of servers) {
+ await checkVideoDuration(server, videoUUID, 2)
+ }
+
+ await checkPersistentTmpIsEmpty(servers[0])
+ })
+
+ after(async function () {
+ await cleanupTests(servers)
+ })
+})
expect(job.payload.output.resolution).to.equal(720)
expect(job.payload.output.fps).to.equal(25)
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath('video_short.webm'))
expect(body).to.deep.equal(inputFile)
const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
jobToken = job.jobToken
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
expect(body).to.deep.equal(inputFile)
const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
jobToken = job.jobToken
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
expect(body).to.deep.equal(inputFile)
const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
jobToken = job.jobToken
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
expect(body).to.deep.equal(inputFile)
const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
jobToken = job.jobToken
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath(maxQualityFile))
expect(body).to.deep.equal(inputFile)
expect(job.payload.output.resolution).to.equal(480)
{
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.audioFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.audioFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath('sample.ogg'))
expect(body).to.deep.equal(inputFile)
}
{
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.previewFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.previewFileUrl, jobToken, runnerToken })
const video = await servers[0].videos.get({ id: videoUUID })
const { body: inputFile } = await makeGetRequest({
const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
jobToken = job.jobToken
- const { body } = await servers[0].runnerJobs.getInputFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+ const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
const inputFile = await readFile(buildAbsoluteFixturePath('video_short_480p.mp4'))
expect(body).to.deep.equal(inputFile)
expect(data.live.transcoding.alwaysTranscodeOriginalResolution).to.be.true
expect(data.videoStudio.enabled).to.be.false
+ expect(data.videoStudio.remoteRunners.enabled).to.be.false
expect(data.import.videos.concurrency).to.equal(2)
expect(data.import.videos.http.enabled).to.be.true
expect(data.live.transcoding.alwaysTranscodeOriginalResolution).to.be.false
expect(data.videoStudio.enabled).to.be.true
+ expect(data.videoStudio.remoteRunners.enabled).to.be.true
expect(data.import.videos.concurrency).to.equal(4)
expect(data.import.videos.http.enabled).to.be.false
}
},
videoStudio: {
- enabled: true
+ enabled: true,
+ remoteRunners: {
+ enabled: true
+ }
},
import: {
videos: {
import { expect } from 'chai'
-import { checkPersistentTmpIsEmpty, expectStartWith } from '@server/tests/shared'
+import { checkPersistentTmpIsEmpty, checkVideoDuration, expectStartWith } from '@server/tests/shared'
import { areMockObjectStorageTestsDisabled, getAllFiles } from '@shared/core-utils'
import { VideoStudioTask } from '@shared/models'
import {
let servers: PeerTubeServer[] = []
let videoUUID: string
- async function checkDuration (server: PeerTubeServer, duration: number) {
- const video = await server.videos.get({ id: videoUUID })
-
- expect(video.duration).to.be.approximately(duration, 1)
-
- for (const file of video.files) {
- const metadata = await server.videos.getFileMetadata({ url: file.metadataUrl })
-
- for (const stream of metadata.streams) {
- expect(Math.round(stream.duration)).to.be.approximately(duration, 1)
- }
- }
- }
-
async function renewVideo (fixture = 'video_short.webm') {
const video = await servers[0].videos.quickUpload({ name: 'video', fixture })
videoUUID = video.uuid
])
for (const server of servers) {
- await checkDuration(server, 3)
+ await checkVideoDuration(server, videoUUID, 3)
const video = await server.videos.get({ id: videoUUID })
expect(new Date(video.publishedAt)).to.be.below(beforeTasks)
])
for (const server of servers) {
- await checkDuration(server, 2)
+ await checkVideoDuration(server, videoUUID, 2)
}
})
])
for (const server of servers) {
- await checkDuration(server, 4)
+ await checkVideoDuration(server, videoUUID, 4)
}
})
})
])
for (const server of servers) {
- await checkDuration(server, 10)
+ await checkVideoDuration(server, videoUUID, 10)
}
})
])
for (const server of servers) {
- await checkDuration(server, 7)
+ await checkVideoDuration(server, videoUUID, 7)
}
})
])
for (const server of servers) {
- await checkDuration(server, 12)
+ await checkVideoDuration(server, videoUUID, 12)
}
})
])
for (const server of servers) {
- await checkDuration(server, 7)
+ await checkVideoDuration(server, videoUUID, 7)
}
})
])
for (const server of servers) {
- await checkDuration(server, 10)
+ await checkVideoDuration(server, videoUUID, 10)
}
})
])
for (const server of servers) {
- await checkDuration(server, 10)
+ await checkVideoDuration(server, videoUUID, 10)
}
})
})
await createTasks(VideoStudioCommand.getComplexTask())
for (const server of servers) {
- await checkDuration(server, 9)
+ await checkVideoDuration(server, videoUUID, 9)
}
})
})
const video = await server.videos.get({ id: videoUUID })
expect(video.files).to.have.lengthOf(0)
- await checkDuration(server, 9)
+ await checkVideoDuration(server, videoUUID, 9)
}
})
})
expectStartWith(hlsFile.fileUrl, ObjectStorageCommand.getMockPlaylistBaseUrl())
}
- await checkDuration(server, 9)
+ await checkVideoDuration(server, videoUUID, 9)
}
})
})
await waitJobs(servers)
for (const server of servers) {
- await checkDuration(server, 9)
+ await checkVideoDuration(server, videoUUID, 9)
}
})
export * from './client-cli'
export * from './live-transcoding'
+export * from './studio-transcoding'
export * from './vod-transcoding'
import { expect } from 'chai'
/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
-import { expectStartWith, PeerTubeRunnerProcess, SQLCommand, testLiveVideoResolutions } from '@server/tests/shared'
+import {
+ checkPeerTubeRunnerCacheIsEmpty,
+ expectStartWith,
+ PeerTubeRunnerProcess,
+ SQLCommand,
+ testLiveVideoResolutions
+} from '@server/tests/shared'
import { areMockObjectStorageTestsDisabled, wait } from '@shared/core-utils'
import { HttpStatusCode, VideoPrivacy } from '@shared/models'
import {
runSuite({ objectStorage: true })
})
+ describe('Check cleanup', function () {
+
+ it('Should have an empty cache directory', async function () {
+ await checkPeerTubeRunnerCacheIsEmpty()
+ })
+ })
+
after(async function () {
await peertubeRunner.unregisterPeerTubeInstance({ server: servers[0] })
peertubeRunner.kill()
--- /dev/null
+
+import { expect } from 'chai'
+import { checkPeerTubeRunnerCacheIsEmpty, checkVideoDuration, expectStartWith, PeerTubeRunnerProcess } from '@server/tests/shared'
+import { areMockObjectStorageTestsDisabled, getAllFiles, wait } from '@shared/core-utils'
+import {
+ cleanupTests,
+ createMultipleServers,
+ doubleFollow,
+ ObjectStorageCommand,
+ PeerTubeServer,
+ setAccessTokensToServers,
+ setDefaultVideoChannel,
+ VideoStudioCommand,
+ waitJobs
+} from '@shared/server-commands'
+
+describe('Test studio transcoding in peertube-runner program', function () {
+ let servers: PeerTubeServer[] = []
+ let peertubeRunner: PeerTubeRunnerProcess
+
+ function runSuite (options: {
+ objectStorage: boolean
+ }) {
+ const { objectStorage } = options
+
+ it('Should run a complex studio transcoding', async function () {
+ this.timeout(120000)
+
+ const { uuid } = await servers[0].videos.quickUpload({ name: 'mp4', fixture: 'video_short.mp4' })
+ await waitJobs(servers)
+
+ const video = await servers[0].videos.get({ id: uuid })
+ const oldFileUrls = getAllFiles(video).map(f => f.fileUrl)
+
+ await servers[0].videoStudio.createEditionTasks({ videoId: uuid, tasks: VideoStudioCommand.getComplexTask() })
+ await waitJobs(servers, { runnerJobs: true })
+
+ for (const server of servers) {
+ const video = await server.videos.get({ id: uuid })
+ const files = getAllFiles(video)
+
+ for (const f of files) {
+ expect(oldFileUrls).to.not.include(f.fileUrl)
+ }
+
+ if (objectStorage) {
+ for (const webtorrentFile of video.files) {
+ expectStartWith(webtorrentFile.fileUrl, ObjectStorageCommand.getMockWebTorrentBaseUrl())
+ }
+
+ for (const hlsFile of video.streamingPlaylists[0].files) {
+ expectStartWith(hlsFile.fileUrl, ObjectStorageCommand.getMockPlaylistBaseUrl())
+ }
+ }
+
+ await checkVideoDuration(server, uuid, 9)
+ }
+ })
+ }
+
+ before(async function () {
+ this.timeout(120_000)
+
+ servers = await createMultipleServers(2)
+
+ await setAccessTokensToServers(servers)
+ await setDefaultVideoChannel(servers)
+
+ await doubleFollow(servers[0], servers[1])
+
+ await servers[0].config.enableTranscoding(true, true)
+ await servers[0].config.enableStudio()
+ await servers[0].config.enableRemoteStudio()
+
+ const registrationToken = await servers[0].runnerRegistrationTokens.getFirstRegistrationToken()
+
+ peertubeRunner = new PeerTubeRunnerProcess()
+ await peertubeRunner.runServer({ hideLogs: false })
+ await peertubeRunner.registerPeerTubeInstance({ server: servers[0], registrationToken, runnerName: 'runner' })
+ })
+
+ describe('With videos on local filesystem storage', function () {
+ runSuite({ objectStorage: false })
+ })
+
+ describe('With videos on object storage', function () {
+ if (areMockObjectStorageTestsDisabled()) return
+
+ before(async function () {
+ await ObjectStorageCommand.prepareDefaultMockBuckets()
+
+ await servers[0].kill()
+
+ await servers[0].run(ObjectStorageCommand.getDefaultMockConfig())
+
+ // Wait for peertube runner socket reconnection
+ await wait(1500)
+ })
+
+ runSuite({ objectStorage: true })
+ })
+
+ describe('Check cleanup', function () {
+
+ it('Should have an empty cache directory', async function () {
+ await checkPeerTubeRunnerCacheIsEmpty()
+ })
+ })
+
+ after(async function () {
+ await peertubeRunner.unregisterPeerTubeInstance({ server: servers[0] })
+ peertubeRunner.kill()
+
+ await cleanupTests(servers)
+ })
+})
/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
import { expect } from 'chai'
-import { completeCheckHlsPlaylist, completeWebVideoFilesCheck, PeerTubeRunnerProcess } from '@server/tests/shared'
+import {
+ checkPeerTubeRunnerCacheIsEmpty,
+ completeCheckHlsPlaylist,
+ completeWebVideoFilesCheck,
+ PeerTubeRunnerProcess
+} from '@server/tests/shared'
import { areMockObjectStorageTestsDisabled, getAllFiles, wait } from '@shared/core-utils'
import { VideoPrivacy } from '@shared/models'
import {
})
})
+ describe('Check cleanup', function () {
+
+ it('Should have an empty cache directory', async function () {
+ await checkPeerTubeRunnerCacheIsEmpty()
+ })
+ })
+
after(async function () {
await peertubeRunner.unregisterPeerTubeInstance({ server: servers[0] })
peertubeRunner.kill()
})
}
+// ---------------------------------------------------------------------------
+
+async function checkVideoDuration (server: PeerTubeServer, videoUUID: string, duration: number) {
+ const video = await server.videos.get({ id: videoUUID })
+
+ expect(video.duration).to.be.approximately(duration, 1)
+
+ for (const file of video.files) {
+ const metadata = await server.videos.getFileMetadata({ url: file.metadataUrl })
+
+ for (const stream of metadata.streams) {
+ expect(Math.round(stream.duration)).to.be.approximately(duration, 1)
+ }
+ }
+}
+
export {
dateIsValid,
testImageSize,
checkBadStartPagination,
checkBadCountPagination,
checkBadSortPagination,
+ checkVideoDuration,
expectLogContain
}
import { expect } from 'chai'
import { pathExists, readdir } from 'fs-extra'
+import { homedir } from 'os'
+import { join } from 'path'
import { PeerTubeServer } from '@shared/server-commands'
-async function checkTmpIsEmpty (server: PeerTubeServer) {
+export async function checkTmpIsEmpty (server: PeerTubeServer) {
await checkDirectoryIsEmpty(server, 'tmp', [ 'plugins-global.css', 'hls', 'resumable-uploads' ])
if (await pathExists(server.getDirectoryPath('tmp/hls'))) {
}
}
-async function checkPersistentTmpIsEmpty (server: PeerTubeServer) {
+export async function checkPersistentTmpIsEmpty (server: PeerTubeServer) {
await checkDirectoryIsEmpty(server, 'tmp-persistent')
}
-async function checkDirectoryIsEmpty (server: PeerTubeServer, directory: string, exceptions: string[] = []) {
+export async function checkDirectoryIsEmpty (server: PeerTubeServer, directory: string, exceptions: string[] = []) {
const directoryPath = server.getDirectoryPath(directory)
const directoryExists = await pathExists(directoryPath)
expect(filtered).to.have.lengthOf(0)
}
-export {
- checkTmpIsEmpty,
- checkPersistentTmpIsEmpty,
- checkDirectoryIsEmpty
+export async function checkPeerTubeRunnerCacheIsEmpty () {
+ const directoryPath = join(homedir(), '.cache', 'peertube-runner-nodejs', 'test', 'transcoding')
+
+ const directoryExists = await pathExists(directoryPath)
+ expect(directoryExists).to.be.true
+
+ const files = await readdir(directoryPath)
+
+ expect(files).to.have.lengthOf(0)
}
+import { VideoStudioTaskPayload } from '../server'
+
export type RunnerJobVODPayload =
RunnerJobVODWebVideoTranscodingPayload |
RunnerJobVODHLSTranscodingPayload |
export type RunnerJobPayload =
RunnerJobVODPayload |
- RunnerJobLiveRTMPHLSTranscodingPayload
+ RunnerJobLiveRTMPHLSTranscodingPayload |
+ RunnerJobVideoEditionTranscodingPayload
// ---------------------------------------------------------------------------
}
}
+export interface RunnerJobVideoEditionTranscodingPayload {
+ input: {
+ videoFileUrl: string
+ }
+
+ tasks: VideoStudioTaskPayload[]
+}
+
// ---------------------------------------------------------------------------
export function isAudioMergeTranscodingPayload (payload: RunnerJobPayload): payload is RunnerJobVODAudioMergeTranscodingPayload {
+import { VideoStudioTaskPayload } from '../server'
+
export type RunnerJobVODPrivatePayload =
RunnerJobVODWebVideoTranscodingPrivatePayload |
RunnerJobVODAudioMergeTranscodingPrivatePayload |
export type RunnerJobPrivatePayload =
RunnerJobVODPrivatePayload |
- RunnerJobLiveRTMPHLSTranscodingPrivatePayload
+ RunnerJobLiveRTMPHLSTranscodingPrivatePayload |
+ RunnerJobVideoEditionTranscodingPrivatePayload
// ---------------------------------------------------------------------------
masterPlaylistName: string
outputDirectory: string
}
+
+// ---------------------------------------------------------------------------
+
+export interface RunnerJobVideoEditionTranscodingPrivatePayload {
+ videoUUID: string
+ originalTasks: VideoStudioTaskPayload[]
+}
VODWebVideoTranscodingSuccess |
VODHLSTranscodingSuccess |
VODAudioMergeTranscodingSuccess |
- LiveRTMPHLSTranscodingSuccess
+ LiveRTMPHLSTranscodingSuccess |
+ VideoEditionTranscodingSuccess
export interface VODWebVideoTranscodingSuccess {
videoFile: Blob | string
}
+export interface VideoEditionTranscodingSuccess {
+ videoFile: Blob | string
+}
+
export function isWebVideoOrAudioMergeTranscodingPayloadSuccess (
payload: RunnerJobSuccessPayload
): payload is VODHLSTranscodingSuccess | VODAudioMergeTranscodingSuccess {
'vod-web-video-transcoding' |
'vod-hls-transcoding' |
'vod-audio-merge-transcoding' |
- 'live-rtmp-hls-transcoding'
+ 'live-rtmp-hls-transcoding' |
+ 'video-edition-transcoding'
videoStudio: {
enabled: boolean
+
+ remoteRunners: {
+ enabled: boolean
+ }
}
import: {
options: {
file: string
+
+ watermarkSizeRatio: number
+ horitonzalMarginRatio: number
+ verticalMarginRatio: number
}
}
-import { VideoPrivacy } from '../videos/video-privacy.enum'
import { ClientScriptJSON } from '../plugins/plugin-package-json.model'
import { NSFWPolicyType } from '../videos/nsfw-policy.type'
+import { VideoPrivacy } from '../videos/video-privacy.enum'
import { BroadcastMessageLevel } from './broadcast-message-level.type'
export interface ServerConfigPlugin {
videoStudio: {
enabled: boolean
+
+ remoteRunners: {
+ enabled: boolean
+ }
}
import: {
file: Blob | string
}
}
+
+// ---------------------------------------------------------------------------
+
+export function isVideoStudioTaskIntro (v: VideoStudioTask): v is VideoStudioTaskIntro {
+ return v.name === 'add-intro'
+}
+
+export function isVideoStudioTaskOutro (v: VideoStudioTask): v is VideoStudioTaskOutro {
+ return v.name === 'add-outro'
+}
+
+export function isVideoStudioTaskWatermark (v: VideoStudioTask): v is VideoStudioTaskWatermark {
+ return v.name === 'add-watermark'
+}
+
+export function hasVideoStudioTaskFile (v: VideoStudioTask): v is VideoStudioTaskIntro | VideoStudioTaskOutro | VideoStudioTaskWatermark {
+ return isVideoStudioTaskIntro(v) || isVideoStudioTaskOutro(v) || isVideoStudioTaskWatermark(v)
+}
})
}
- getInputFile (options: OverrideCommandOptions & { url: string, jobToken: string, runnerToken: string }) {
+ getJobFile (options: OverrideCommandOptions & { url: string, jobToken: string, runnerToken: string }) {
const { host, protocol, pathname } = new URL(options.url)
return this.postBodyRequest({
const { data } = await this.list({ count: 100 })
+ const allowedStates = new Set<RunnerJobState>([
+ RunnerJobState.PENDING,
+ RunnerJobState.PROCESSING,
+ RunnerJobState.WAITING_FOR_PARENT_JOB
+ ])
+
for (const job of data) {
if (state && job.state.id !== state) continue
+ else if (allowedStates.has(job.state.id) !== true) continue
await this.cancelByAdmin({ jobUUID: job.uuid })
}
})
}
+ enableRemoteStudio () {
+ return this.updateExistingSubConfig({
+ newConfig: {
+ videoStudio: {
+ remoteRunners: {
+ enabled: true
+ }
+ }
+ }
+ })
+ }
+
// ---------------------------------------------------------------------------
enableStudio () {
}
},
videoStudio: {
- enabled: false
+ enabled: false,
+ remoteRunners: {
+ enabled: false
+ }
},
import: {
videos: {
projects / github / Chocobozzz / PeerTube.git / commitdiff
