From 3a0c2a77b1a6626699514ddaf8135f4397175443 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Tue, 2 May 2023 13:51:06 +0200 Subject: [PATCH] Enable external plugins to test the PR --- .github/workflows/test.yml | 2 +- .../runner-job-list.component.html | 2 +- packages/peertube-runner/README.md | 22 ++++++++++ packages/peertube-runner/peertube-runner.ts | 8 ++-- .../server/process/shared/process-live.ts | 10 ++--- packages/peertube-runner/server/server.ts | 17 ++++++-- packages/peertube-runner/shared/http.ts | 2 +- .../peertube-runner/shared/ipc/ipc-server.ts | 4 +- scripts/build/peertube-runner.sh | 2 +- server/lib/job-queue/job-queue.ts | 7 +++- server/lib/live/shared/muxing-session.ts | 40 ++++++++++--------- .../job-handlers/abstract-job-handler.ts | 22 ++++++---- server/tests/api/videos/resumable-upload.ts | 2 +- .../tests/cli/create-import-video-file-job.ts | 2 +- shared/server-commands/requests/requests.ts | 4 +- 15 files changed, 97 insertions(+), 49 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 678b0674b..2288bcd3f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -77,7 +77,7 @@ jobs: - name: Run Test # external-plugins tests only run on schedule - if: github.event_name == 'schedule' || matrix.test_suite != 'external-plugins' + # if: github.event_name == 'schedule' || matrix.test_suite != 'external-plugins' env: AKISMET_KEY: ${{ secrets.AKISMET_KEY }} run: npm run ci -- ${{ matrix.test_suite }} diff --git a/client/src/app/+admin/system/runners/runner-job-list/runner-job-list.component.html b/client/src/app/+admin/system/runners/runner-job-list/runner-job-list.component.html index 7858b4bca..d42f600e0 100644 --- a/client/src/app/+admin/system/runners/runner-job-list/runner-job-list.component.html +++ b/client/src/app/+admin/system/runners/runner-job-list/runner-job-list.component.html @@ -84,7 +84,7 @@ 
{{ runnerJob.privatePayload }}
- 
{{ runnerJob.error }}
+ 
{{ runnerJob.error }}
diff --git a/packages/peertube-runner/README.md b/packages/peertube-runner/README.md index b7cf174d5..87c3c5354 100644 --- a/packages/peertube-runner/README.md +++ b/packages/peertube-runner/README.md @@ -1 +1,23 @@ # PeerTube runner + +Runner program to execute jobs (transcoding...) of remote PeerTube instances. + +Commands below has to be run at the root of PeerTube git repository. + +## Develop + +```bash +npm run dev:peertube-runner +``` + +## Build + +```bash +npm run build:peertube-runner +``` + +## Run + +```bash +node packages/peertube-runner/dist/peertube-runner.js --help +``` diff --git a/packages/peertube-runner/peertube-runner.ts b/packages/peertube-runner/peertube-runner.ts index 6bfd9ac0f..f02526ef9 100644 --- a/packages/peertube-runner/peertube-runner.ts +++ b/packages/peertube-runner/peertube-runner.ts @@ -26,7 +26,7 @@ program.command('server') try { await RunnerServer.Instance.run() } catch (err) { - console.error('Cannot run PeerTube runner as server mode', err) + logger.error('Cannot run PeerTube runner as server mode', err) process.exit(-1) } }) @@ -41,7 +41,7 @@ program.command('register') try { await registerRunner(options) } catch (err) { - console.error('Cannot register this PeerTube runner.', err) + logger.error('Cannot register this PeerTube runner.', err) process.exit(-1) } }) @@ -53,7 +53,7 @@ program.command('unregister') try { await unregisterRunner(options) } catch (err) { - console.error('Cannot unregister this PeerTube runner.', err) + logger.error('Cannot unregister this PeerTube runner.', err) process.exit(-1) } }) @@ -64,7 +64,7 @@ program.command('list-registered') try { await listRegistered() } catch (err) { - console.error('Cannot list registered PeerTube instances.', err) + logger.error('Cannot list registered PeerTube instances.', err) process.exit(-1) } }) diff --git a/packages/peertube-runner/server/process/shared/process-live.ts b/packages/peertube-runner/server/process/shared/process-live.ts index 5a3b596a2..b17b51c7c 100644 --- a/packages/peertube-runner/server/process/shared/process-live.ts +++ b/packages/peertube-runner/server/process/shared/process-live.ts @@ -204,8 +204,8 @@ export class ProcessLiveRTMPHLSTranscoding { // --------------------------------------------------------------------------- - private sendDeletedChunkUpdate (deletedChunk: string) { - if (this.ended) return + private sendDeletedChunkUpdate (deletedChunk: string): Promise { + if (this.ended) return Promise.resolve() logger.debug(`Sending removed live chunk ${deletedChunk} update`) @@ -230,8 +230,8 @@ export class ProcessLiveRTMPHLSTranscoding { return this.updateWithRetry(payload) } - private sendAddedChunkUpdate (addedChunk: string) { - if (this.ended) return + private sendAddedChunkUpdate (addedChunk: string): Promise { + if (this.ended) return Promise.resolve() logger.debug(`Sending added live chunk ${addedChunk} update`) @@ -257,7 +257,7 @@ export class ProcessLiveRTMPHLSTranscoding { return this.updateWithRetry(payload) } - private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1) { + private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1): Promise { if (this.ended || this.errored) return try { diff --git a/packages/peertube-runner/server/server.ts b/packages/peertube-runner/server/server.ts index 724f359bd..e851dfc7c 100644 --- a/packages/peertube-runner/server/server.ts +++ b/packages/peertube-runner/server/server.ts @@ -23,6 +23,8 @@ export class RunnerServer { private checkingAvailableJobs = false + private cleaningUp = false + private readonly sockets = new Map() private constructor () {} @@ -45,13 +47,17 @@ export class RunnerServer { try { await ipcServer.run(this) } catch (err) { - console.error('Cannot start local socket for IPC communication', err) + logger.error('Cannot start local socket for IPC communication', err) process.exit(-1) } // Cleanup on exit for (const code of [ 'SIGINT', 'SIGUSR1', 'SIGUSR2', 'uncaughtException' ]) { - process.on(code, async () => { + process.on(code, async (err, origin) => { + if (code === 'uncaughtException') { + logger.error({ err, origin }, 'uncaughtException') + } + await this.onExit() }) } @@ -244,6 +250,11 @@ export class RunnerServer { } private async onExit () { + if (this.cleaningUp) return + this.cleaningUp = true + + logger.info('Cleaning up after program exit') + try { for (const { server, job } of this.processingJobs) { await server.runnerJobs.abort({ @@ -256,7 +267,7 @@ export class RunnerServer { await this.cleanupTMP() } catch (err) { - console.error(err) + logger.error(err) process.exit(-1) } diff --git a/packages/peertube-runner/shared/http.ts b/packages/peertube-runner/shared/http.ts index d3fff70d1..df64dc168 100644 --- a/packages/peertube-runner/shared/http.ts +++ b/packages/peertube-runner/shared/http.ts @@ -47,7 +47,7 @@ export function downloadFile (options: { request.on('error', err => { remove(destination) - .catch(err => console.error(err)) + .catch(err => logger.error(err)) return rej(err) }) diff --git a/packages/peertube-runner/shared/ipc/ipc-server.ts b/packages/peertube-runner/shared/ipc/ipc-server.ts index bc340198b..922dc93e5 100644 --- a/packages/peertube-runner/shared/ipc/ipc-server.ts +++ b/packages/peertube-runner/shared/ipc/ipc-server.ts @@ -27,7 +27,7 @@ export class IPCServer { this.sendReponse(res, { success: true, data }) } catch (err) { - console.error('Cannot execute RPC call', err) + logger.error('Cannot execute RPC call', err) this.sendReponse(res, { success: false, error: err.message }) } }) @@ -56,6 +56,6 @@ export class IPCServer { body: IPCReponse ) { response(body) - .catch(err => console.error('Cannot send response after IPC request', err)) + .catch(err => logger.error('Cannot send response after IPC request', err)) } } diff --git a/scripts/build/peertube-runner.sh b/scripts/build/peertube-runner.sh index 690031af5..9c326747b 100755 --- a/scripts/build/peertube-runner.sh +++ b/scripts/build/peertube-runner.sh @@ -10,4 +10,4 @@ rm -rf ./dist rm -rf ./dist mkdir ./dist -./node_modules/.bin/esbuild ./peertube-runner.ts --bundle --platform=node --external:"./lib-cov/fluent-ffmpeg" --external:pg-hstore --outfile=dist/peertube-runner.js +./node_modules/.bin/esbuild ./peertube-runner.ts --bundle --platform=node --target=node14 --external:"./lib-cov/fluent-ffmpeg" --external:pg-hstore --outfile=dist/peertube-runner.js diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index 21bf0f226..03f6fbea7 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -96,6 +96,7 @@ export type CreateJobArgument = export type CreateJobOptions = { delay?: number priority?: number + failParentOnFailure?: boolean } const handlers: { [id in JobType]: (job: Job) => Promise } = { @@ -363,7 +364,11 @@ class JobQueue { name: 'job', data: job.payload, queueName: job.type, - opts: this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay' ])) + opts: { + failParentOnFailure: true, + + ...this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay', 'failParentOnFailure' ])) + } } } diff --git a/server/lib/live/shared/muxing-session.ts b/server/lib/live/shared/muxing-session.ts index f3f8fc886..ef4ecb83e 100644 --- a/server/lib/live/shared/muxing-session.ts +++ b/server/lib/live/shared/muxing-session.ts @@ -79,9 +79,7 @@ class MuxingSession extends EventEmitter { private streamingPlaylist: MStreamingPlaylistVideo private liveSegmentShaStore: LiveSegmentShaStore - private tsWatcher: FSWatcher - private masterWatcher: FSWatcher - private m3u8Watcher: FSWatcher + private filesWatcher: FSWatcher private masterPlaylistCreated = false private liveReady = false @@ -149,6 +147,8 @@ class MuxingSession extends EventEmitter { await this.transcodingWrapper.run() + this.filesWatcher = watch(this.outDirectory, { depth: 0 }) + this.watchMasterFile() this.watchTSFiles() this.watchM3U8File() @@ -168,9 +168,10 @@ class MuxingSession extends EventEmitter { } private watchMasterFile () { - this.masterWatcher = watch(this.outDirectory + '/' + this.streamingPlaylist.playlistFilename) + this.filesWatcher.on('add', async path => { + if (path !== join(this.outDirectory, this.streamingPlaylist.playlistFilename)) return + if (this.masterPlaylistCreated === true) return - this.masterWatcher.on('add', async () => { try { if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) { const url = await storeHLSFileFromFilename(this.streamingPlaylist, this.streamingPlaylist.playlistFilename) @@ -188,20 +189,18 @@ class MuxingSession extends EventEmitter { this.masterPlaylistCreated = true logger.info('Master playlist file for %s has been created', this.videoUUID, this.lTags()) - - this.masterWatcher.close() - .catch(err => logger.error('Cannot close master watcher of %s.', this.outDirectory, { err, ...this.lTags() })) }) } private watchM3U8File () { - this.m3u8Watcher = watch(this.outDirectory + '/*.m3u8') - const sendQueues = new Map() - const onChangeOrAdd = async (m3u8Path: string) => { + const onChange = async (m3u8Path: string) => { + if (m3u8Path.endsWith('.m3u8') !== true) return if (this.streamingPlaylist.storage !== VideoStorage.OBJECT_STORAGE) return + logger.debug('Live change handler of M3U8 file %s.', m3u8Path, this.lTags()) + try { if (!sendQueues.has(m3u8Path)) { sendQueues.set(m3u8Path, new PQueue({ concurrency: 1 })) @@ -214,18 +213,18 @@ class MuxingSession extends EventEmitter { } } - this.m3u8Watcher.on('change', onChangeOrAdd) + this.filesWatcher.on('change', onChange) } private watchTSFiles () { const startStreamDateTime = new Date().getTime() - this.tsWatcher = watch(this.outDirectory + '/*.ts') - const playlistIdMatcher = /^([\d+])-/ const addHandler = async (segmentPath: string) => { - logger.debug('Live add handler of %s.', segmentPath, this.lTags()) + if (segmentPath.endsWith('.ts') !== true) return + + logger.debug('Live add handler of TS file %s.', segmentPath, this.lTags()) const playlistId = basename(segmentPath).match(playlistIdMatcher)[0] @@ -252,6 +251,10 @@ class MuxingSession extends EventEmitter { } const deleteHandler = async (segmentPath: string) => { + if (segmentPath.endsWith('.ts') !== true) return + + logger.debug('Live delete handler of TS file %s.', segmentPath, this.lTags()) + try { await this.liveSegmentShaStore.removeSegmentSha(segmentPath) } catch (err) { @@ -267,8 +270,8 @@ class MuxingSession extends EventEmitter { } } - this.tsWatcher.on('add', p => addHandler(p)) - this.tsWatcher.on('unlink', p => deleteHandler(p)) + this.filesWatcher.on('add', p => addHandler(p)) + this.filesWatcher.on('unlink', p => deleteHandler(p)) } private async isQuotaExceeded (segmentPath: string) { @@ -371,7 +374,8 @@ class MuxingSession extends EventEmitter { setTimeout(() => { // Wait latest segments generation, and close watchers - Promise.all([ this.tsWatcher.close(), this.masterWatcher.close(), this.m3u8Watcher.close() ]) + const promise = this.filesWatcher?.close() || Promise.resolve() + promise .then(() => { // Process remaining segments hash for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) { diff --git a/server/lib/runners/job-handlers/abstract-job-handler.ts b/server/lib/runners/job-handlers/abstract-job-handler.ts index 73fc14574..74b455107 100644 --- a/server/lib/runners/job-handlers/abstract-job-handler.ts +++ b/server/lib/runners/job-handlers/abstract-job-handler.ts @@ -21,6 +21,7 @@ import { RunnerJobVODWebVideoTranscodingPayload, RunnerJobVODWebVideoTranscodingPrivatePayload } from '@shared/models' +import { throttle } from 'lodash' type CreateRunnerJobArg = { @@ -48,6 +49,8 @@ export abstract class AbstractJobHandler @@ -102,16 +105,19 @@ export abstract class AbstractJobHandler { return sequelizeTypescript.transaction(async transaction => { - if (runnerJob.changed()) { - return runnerJob.save({ transaction }) - } - - // Don't update the job too often - if (new Date().getTime() - runnerJob.updatedAt.getTime() > 2000) { - await setAsUpdated({ sequelize: sequelizeTypescript, table: 'runnerJob', id: runnerJob.id, transaction }) - } + return runnerJob.save({ transaction }) }) }) } diff --git a/server/tests/api/videos/resumable-upload.ts b/server/tests/api/videos/resumable-upload.ts index a70a7258b..2fbefb392 100644 --- a/server/tests/api/videos/resumable-upload.ts +++ b/server/tests/api/videos/resumable-upload.ts @@ -79,7 +79,7 @@ describe('Test resumable upload', function () { async function checkFileSize (uploadIdArg: string, expectedSize: number | null) { const uploadId = uploadIdArg.replace(/^upload_id=/, '') - const subPath = join('tmp', 'resumable-uploads', uploadId) + const subPath = join('tmp', 'resumable-uploads', `${rootId}-${uploadId}.mp4`) const filePath = server.servers.buildDirectory(subPath) const exists = await pathExists(filePath) diff --git a/server/tests/cli/create-import-video-file-job.ts b/server/tests/cli/create-import-video-file-job.ts index 43f53035b..3ece4f2ec 100644 --- a/server/tests/cli/create-import-video-file-job.ts +++ b/server/tests/cli/create-import-video-file-job.ts @@ -73,7 +73,7 @@ function runTests (objectStorage: boolean) { }) it('Should run a import job on video 1 with a lower resolution', async function () { - const command = `npm run create-import-video-file-job -- -v ${video1ShortId} -i server/tests/fixtures/video_short-480.webm` + const command = `npm run create-import-video-file-job -- -v ${video1ShortId} -i server/tests/fixtures/video_short_480.webm` await servers[0].cli.execWithEnv(command) await waitJobs(servers) diff --git a/shared/server-commands/requests/requests.ts b/shared/server-commands/requests/requests.ts index 96f67b4c7..e3f1817f1 100644 --- a/shared/server-commands/requests/requests.ts +++ b/shared/server-commands/requests/requests.ts @@ -159,7 +159,7 @@ function unwrapBodyOrDecodeToJSON (test: request.Test): Promise { try { return JSON.parse(new TextDecoder().decode(res.body)) } catch (err) { - console.error('Cannot decode JSON.', res.body instanceof Buffer ? res.body.toString() : res.body) + console.error('Cannot decode JSON.', { res, body: res.body instanceof Buffer ? res.body.toString() : res.body }) throw err } } @@ -168,7 +168,7 @@ function unwrapBodyOrDecodeToJSON (test: request.Test): Promise { try { return JSON.parse(res.text) } catch (err) { - console.error('Cannot decode json', res.text) + console.error('Cannot decode json', { res, text: res.text }) throw err } } -- 2.41.0