- 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 }}
<pre>{{ runnerJob.privatePayload }}</pre>
</div>
- <pre *ngIf="runnerJob.error" class=".text-danger" >{{ runnerJob.error }}</pre>
+ <pre *ngIf="runnerJob.error" class="text-danger" >{{ runnerJob.error }}</pre>
</td>
</tr>
</ng-template>
# 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
+```
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)
}
})
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)
}
})
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)
}
})
try {
await listRegistered()
} catch (err) {
- console.error('Cannot list registered PeerTube instances.', err)
+ logger.error('Cannot list registered PeerTube instances.', err)
process.exit(-1)
}
})
// ---------------------------------------------------------------------------
- private sendDeletedChunkUpdate (deletedChunk: string) {
- if (this.ended) return
+ private sendDeletedChunkUpdate (deletedChunk: string): Promise<any> {
+ if (this.ended) return Promise.resolve()
logger.debug(`Sending removed live chunk ${deletedChunk} update`)
return this.updateWithRetry(payload)
}
- private sendAddedChunkUpdate (addedChunk: string) {
- if (this.ended) return
+ private sendAddedChunkUpdate (addedChunk: string): Promise<any> {
+ if (this.ended) return Promise.resolve()
logger.debug(`Sending added live chunk ${addedChunk} update`)
return this.updateWithRetry(payload)
}
- private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1) {
+ private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1): Promise<any> {
if (this.ended || this.errored) return
try {
private checkingAvailableJobs = false
+ private cleaningUp = false
+
private readonly sockets = new Map<PeerTubeServer, Socket>()
private constructor () {}
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()
})
}
}
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({
await this.cleanupTMP()
} catch (err) {
- console.error(err)
+ logger.error(err)
process.exit(-1)
}
request.on('error', err => {
remove(destination)
- .catch(err => console.error(err))
+ .catch(err => logger.error(err))
return rej(err)
})
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 })
}
})
body: IPCReponse<T>
) {
response(body)
- .catch(err => console.error('Cannot send response after IPC request', err))
+ .catch(err => logger.error('Cannot send response after IPC request', err))
}
}
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
export type CreateJobOptions = {
delay?: number
priority?: number
+ failParentOnFailure?: boolean
}
const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
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' ]))
+ }
}
}
private streamingPlaylist: MStreamingPlaylistVideo
private liveSegmentShaStore: LiveSegmentShaStore
- private tsWatcher: FSWatcher
- private masterWatcher: FSWatcher
- private m3u8Watcher: FSWatcher
+ private filesWatcher: FSWatcher
private masterPlaylistCreated = false
private liveReady = false
await this.transcodingWrapper.run()
+ this.filesWatcher = watch(this.outDirectory, { depth: 0 })
+
this.watchMasterFile()
this.watchTSFiles()
this.watchM3U8File()
}
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)
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<string, PQueue>()
- 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 }))
}
}
- 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]
}
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) {
}
}
- 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) {
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)) {
RunnerJobVODWebVideoTranscodingPayload,
RunnerJobVODWebVideoTranscodingPrivatePayload
} from '@shared/models'
+import { throttle } from 'lodash'
type CreateRunnerJobArg =
{
protected readonly lTags = loggerTagsFactory('runner')
+ static setJobAsUpdatedThrottled = throttle(setAsUpdated, 2000)
+
// ---------------------------------------------------------------------------
abstract create (options: C): Promise<MRunnerJob>
if (progress) runnerJob.progress = progress
+ if (!runnerJob.changed()) {
+ try {
+ await AbstractJobHandler.setJobAsUpdatedThrottled({ sequelize: sequelizeTypescript, table: 'runnerJob', id: runnerJob.id })
+ } catch (err) {
+ logger.warn('Cannot set remote job as updated', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
+ }
+
+ return
+ }
+
await retryTransactionWrapper(() => {
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 })
})
})
}
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)
})
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)
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
}
}
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
}
}
projects / github / Chocobozzz / PeerTube.git / commitdiff
