X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fjob-queue%2Fjob-queue.ts;h=4cda12b5754c5ca411384656eaaf2f4edafc192b;hb=e1a570abff3ebf375433e58e7362d56bd32d4cd8;hp=8d97434ac59b954d33510dd32780a051516ec955;hpb=a5cf76afa378aae81af2a9b0ce548e5d2582f832;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index 8d97434ac..4cda12b57 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -1,12 +1,17 @@ -import * as Bull from 'bull' +import Bull, { Job, JobOptions, Queue } from 'bull' +import { jobStates } from '@server/helpers/custom-validators/jobs' +import { CONFIG } from '@server/initializers/config' +import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy' import { ActivitypubFollowPayload, ActivitypubHttpBroadcastPayload, ActivitypubHttpFetcherPayload, ActivitypubHttpUnicastPayload, + ActorKeysPayload, EmailPayload, JobState, JobType, + MoveObjectStoragePayload, RefreshPayload, VideoFileImportPayload, VideoImportPayload, @@ -15,25 +20,28 @@ import { VideoTranscodingPayload } from '../../../shared/models' import { logger } from '../../helpers/logger' -import { Redis } from '../redis' import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants' +import { Redis } from '../redis' +import { processActivityPubCleaner } from './handlers/activitypub-cleaner' +import { processActivityPubFollow } from './handlers/activitypub-follow' import { processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast' import { processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher' import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast' -import { processEmail } from './handlers/email' -import { processVideoTranscoding } from './handlers/video-transcoding' -import { processActivityPubFollow } from './handlers/activitypub-follow' -import { processVideoImport } from './handlers/video-import' -import { processVideosViews } from './handlers/video-views' import { refreshAPObject } from './handlers/activitypub-refresher' +import { processActorKeys } from './handlers/actor-keys' +import { processEmail } from './handlers/email' +import { processMoveToObjectStorage } from './handlers/move-to-object-storage' import { processVideoFileImport } from './handlers/video-file-import' -import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy' +import { processVideoImport } from './handlers/video-import' import { processVideoLiveEnding } from './handlers/video-live-ending' +import { processVideoTranscoding } from './handlers/video-transcoding' +import { processVideosViews } from './handlers/video-views' type CreateJobArgument = { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } | + { type: 'activitypub-http-cleaner', payload: {} } | { type: 'activitypub-follow', payload: ActivitypubFollowPayload } | { type: 'video-file-import', payload: VideoFileImportPayload } | { type: 'video-transcoding', payload: VideoTranscodingPayload } | @@ -42,16 +50,20 @@ type CreateJobArgument = { type: 'activitypub-refresher', payload: RefreshPayload } | { type: 'videos-views', payload: {} } | { type: 'video-live-ending', payload: VideoLiveEndingPayload } | - { type: 'video-redundancy', payload: VideoRedundancyPayload } + { type: 'actor-keys', payload: ActorKeysPayload } | + { type: 'video-redundancy', payload: VideoRedundancyPayload } | + { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } -type CreateJobOptions = { +export type CreateJobOptions = { delay?: number + priority?: number } -const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { +const handlers: { [id in JobType]: (job: Job) => Promise } = { 'activitypub-http-broadcast': processActivityPubHttpBroadcast, 'activitypub-http-unicast': processActivityPubHttpUnicast, 'activitypub-http-fetcher': processActivityPubHttpFetcher, + 'activitypub-cleaner': processActivityPubCleaner, 'activitypub-follow': processActivityPubFollow, 'video-file-import': processVideoFileImport, 'video-transcoding': processVideoTranscoding, @@ -60,7 +72,9 @@ const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { 'videos-views': processVideosViews, 'activitypub-refresher': refreshAPObject, 'video-live-ending': processVideoLiveEnding, - 'video-redundancy': processVideoRedundancy + 'actor-keys': processActorKeys, + 'video-redundancy': processVideoRedundancy, + 'move-to-object-storage': processMoveToObjectStorage } const jobTypes: JobType[] = [ @@ -68,6 +82,7 @@ const jobTypes: JobType[] = [ 'activitypub-http-broadcast', 'activitypub-http-fetcher', 'activitypub-http-unicast', + 'activitypub-cleaner', 'email', 'video-transcoding', 'video-file-import', @@ -75,14 +90,16 @@ const jobTypes: JobType[] = [ 'videos-views', 'activitypub-refresher', 'video-redundancy', - 'video-live-ending' + 'actor-keys', + 'video-live-ending', + 'move-to-object-storage' ] class JobQueue { private static instance: JobQueue - private queues: { [id in JobType]?: Bull.Queue } = {} + private queues: { [id in JobType]?: Queue } = {} private initialized = false private jobRedisPrefix: string @@ -103,11 +120,11 @@ class JobQueue { } } - for (const handlerName of Object.keys(handlers)) { + for (const handlerName of (Object.keys(handlers) as JobType[])) { const queue = new Bull(handlerName, queueOptions) const handler = handlers[handlerName] - queue.process(JOB_CONCURRENCY[handlerName], handler) + queue.process(this.getJobConcurrency(handlerName), handler) .catch(err => logger.error('Error in job queue processor %s.', handlerName, { err })) queue.on('failed', (job, err) => { @@ -143,10 +160,11 @@ class JobQueue { return } - const jobArgs: Bull.JobOptions = { + const jobArgs: JobOptions = { backoff: { delay: 60 * 1000, type: 'exponential' }, attempts: JOB_ATTEMPTS[obj.type], timeout: JOB_TTL[obj.type], + priority: options.priority, delay: options.delay } @@ -154,14 +172,16 @@ class JobQueue { } async listForApi (options: { - state: JobState + state?: JobState start: number count: number asc?: boolean jobType: JobType - }): Promise { + }): Promise { const { state, start, count, asc, jobType } = options - let results: Bull.Job[] = [] + + const states = state ? [ state ] : jobStates + let results: Job[] = [] const filteredJobTypes = this.filterJobTypes(jobType) @@ -172,7 +192,7 @@ class JobQueue { continue } - const jobs = await queue.getJobs([ state ], 0, start + count, asc) + const jobs = await queue.getJobs(states, 0, start + count, asc) results = results.concat(jobs) } @@ -189,6 +209,7 @@ class JobQueue { } async count (state: JobState, jobType?: JobType): Promise { + const states = state ? [ state ] : jobStates let total = 0 const filteredJobTypes = this.filterJobTypes(jobType) @@ -202,7 +223,9 @@ class JobQueue { const counts = await queue.getJobCounts() - total += counts[state] + for (const s of states) { + total += counts[s] + } } return total @@ -219,6 +242,12 @@ class JobQueue { this.queues['videos-views'].add({}, { repeat: REPEAT_JOBS['videos-views'] }).catch(err => logger.error('Cannot add repeatable job.', { err })) + + if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) { + this.queues['activitypub-cleaner'].add({}, { + repeat: REPEAT_JOBS['activitypub-cleaner'] + }).catch(err => logger.error('Cannot add repeatable job.', { err })) + } } private filterJobTypes (jobType?: JobType) { @@ -227,6 +256,13 @@ class JobQueue { return jobTypes.filter(t => t === jobType) } + private getJobConcurrency (jobType: JobType) { + if (jobType === 'video-transcoding') return CONFIG.TRANSCODING.CONCURRENCY + if (jobType === 'video-import') return CONFIG.IMPORT.VIDEOS.CONCURRENCY + + return JOB_CONCURRENCY[jobType] + } + static get Instance () { return this.instance || (this.instance = new this()) }