X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fschedulers%2Fvideos-redundancy-scheduler.ts;h=c1c91b6563d9488ca1b37d8ea2cce68fc544d119;hb=66fb2aa39b6f8e4677f80128c27fbafd3a8fe2e7;hp=8b7f335398acdf9b419a9d5fcd847cffbbb07966;hpb=ae28cdf327d782e629379eee1999096ca2a5d74b;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/schedulers/videos-redundancy-scheduler.ts b/server/lib/schedulers/videos-redundancy-scheduler.ts index 8b7f33539..c1c91b656 100644 --- a/server/lib/schedulers/videos-redundancy-scheduler.ts +++ b/server/lib/schedulers/videos-redundancy-scheduler.ts @@ -1,22 +1,47 @@ import { AbstractScheduler } from './abstract-scheduler' -import { CONFIG, REDUNDANCY, VIDEO_IMPORT_TIMEOUT } from '../../initializers' +import { HLS_REDUNDANCY_DIRECTORY, REDUNDANCY, VIDEO_IMPORT_TIMEOUT, WEBSERVER } from '../../initializers/constants' import { logger } from '../../helpers/logger' import { VideosRedundancy } from '../../../shared/models/redundancy' import { VideoRedundancyModel } from '../../models/redundancy/video-redundancy' -import { VideoFileModel } from '../../models/video/video-file' -import { downloadWebTorrentVideo } from '../../helpers/webtorrent' +import { downloadWebTorrentVideo, generateMagnetUri } from '../../helpers/webtorrent' import { join } from 'path' -import { rename } from 'fs-extra' +import { move } from 'fs-extra' import { getServerActor } from '../../helpers/utils' import { sendCreateCacheFile, sendUpdateCacheFile } from '../activitypub/send' -import { getVideoCacheFileActivityPubUrl } from '../activitypub/url' +import { getVideoCacheFileActivityPubUrl, getVideoCacheStreamingPlaylistActivityPubUrl } from '../activitypub/url' import { removeVideoRedundancy } from '../redundancy' import { getOrCreateVideoAndAccountAndChannel } from '../activitypub' +import { downloadPlaylistSegments } from '../hls' +import { CONFIG } from '../../initializers/config' +import { + MStreamingPlaylist, MStreamingPlaylistFiles, + MStreamingPlaylistVideo, + MVideoAccountLight, + MVideoFile, + MVideoFileVideo, + MVideoRedundancyFileVideo, + MVideoRedundancyStreamingPlaylistVideo, + MVideoRedundancyVideo, + MVideoWithAllFiles +} from '@server/typings/models' +import { getVideoFilename } from '../video-paths' + +type CandidateToDuplicate = { + redundancy: VideosRedundancy, + video: MVideoWithAllFiles, + files: MVideoFile[], + streamingPlaylists: MStreamingPlaylistFiles[] +} + +function isMVideoRedundancyFileVideo ( + o: MVideoRedundancyFileVideo | MVideoRedundancyStreamingPlaylistVideo +): o is MVideoRedundancyFileVideo { + return !!(o as MVideoRedundancyFileVideo).VideoFile +} export class VideosRedundancyScheduler extends AbstractScheduler { private static instance: AbstractScheduler - private executing = false protected schedulerIntervalMs = CONFIG.REDUNDANCY.VIDEOS.CHECK_INTERVAL @@ -24,41 +49,39 @@ export class VideosRedundancyScheduler extends AbstractScheduler { super() } - async execute () { - if (this.executing) return - - this.executing = true - - for (const obj of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) { - logger.info('Running redundancy scheduler for strategy %s.', obj.strategy) + protected async internalExecute () { + for (const redundancyConfig of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) { + logger.info('Running redundancy scheduler for strategy %s.', redundancyConfig.strategy) try { - const videoToDuplicate = await this.findVideoToDuplicate(obj) + const videoToDuplicate = await this.findVideoToDuplicate(redundancyConfig) if (!videoToDuplicate) continue - const videoFiles = videoToDuplicate.VideoFiles - videoFiles.forEach(f => f.Video = videoToDuplicate) + const candidateToDuplicate = { + video: videoToDuplicate, + redundancy: redundancyConfig, + files: videoToDuplicate.VideoFiles, + streamingPlaylists: videoToDuplicate.VideoStreamingPlaylists + } - await this.purgeCacheIfNeeded(obj, videoFiles) + await this.purgeCacheIfNeeded(candidateToDuplicate) - if (await this.isTooHeavy(obj, videoFiles)) { + if (await this.isTooHeavy(candidateToDuplicate)) { logger.info('Video %s is too big for our cache, skipping.', videoToDuplicate.url) continue } - logger.info('Will duplicate video %s in redundancy scheduler "%s".', videoToDuplicate.url, obj.strategy) + logger.info('Will duplicate video %s in redundancy scheduler "%s".', videoToDuplicate.url, redundancyConfig.strategy) - await this.createVideoRedundancy(obj, videoFiles) + await this.createVideoRedundancies(candidateToDuplicate) } catch (err) { - logger.error('Cannot run videos redundancy %s.', obj.strategy, { err }) + logger.error('Cannot run videos redundancy %s.', redundancyConfig.strategy, { err }) } } await this.extendsLocalExpiration() await this.purgeRemoteExpired() - - this.executing = false } static get Instance () { @@ -70,25 +93,38 @@ export class VideosRedundancyScheduler extends AbstractScheduler { for (const redundancyModel of expired) { try { - await this.extendsOrDeleteRedundancy(redundancyModel) + const redundancyConfig = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy) + const candidate = { + redundancy: redundancyConfig, + video: null, + files: [], + streamingPlaylists: [] + } + + // If the administrator disabled the redundancy or decreased the cache size, remove this redundancy instead of extending it + if (!redundancyConfig || await this.isTooHeavy(candidate)) { + logger.info('Destroying redundancy %s because the cache size %s is too heavy.', redundancyModel.url, redundancyModel.strategy) + await removeVideoRedundancy(redundancyModel) + } else { + await this.extendsRedundancy(redundancyModel) + } } catch (err) { - logger.error('Cannot extend expiration of %s video from our redundancy system.', this.buildEntryLogId(redundancyModel)) + logger.error( + 'Cannot extend or remove expiration of %s video from our redundancy system.', this.buildEntryLogId(redundancyModel), + { err } + ) } } } - private async extendsOrDeleteRedundancy (redundancyModel: VideoRedundancyModel) { - // Refresh the video, maybe it was deleted - const video = await this.loadAndRefreshVideo(redundancyModel.VideoFile.Video.url) - - if (!video) { - logger.info('Destroying existing redundancy %s, because the associated video does not exist anymore.', redundancyModel.url) - - await redundancyModel.destroy() + private async extendsRedundancy (redundancyModel: MVideoRedundancyVideo) { + const redundancy = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy) + // Redundancy strategy disabled, remove our redundancy instead of extending expiration + if (!redundancy) { + await removeVideoRedundancy(redundancyModel) return } - const redundancy = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy) await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime) } @@ -119,52 +155,102 @@ export class VideosRedundancyScheduler extends AbstractScheduler { } } - private async createVideoRedundancy (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { - const serverActor = await getServerActor() + private async createVideoRedundancies (data: CandidateToDuplicate) { + const video = await this.loadAndRefreshVideo(data.video.url) - for (const file of filesToDuplicate) { - const video = await this.loadAndRefreshVideo(file.Video.url) + if (!video) { + logger.info('Video %s we want to duplicate does not existing anymore, skipping.', data.video.url) + return + } + + for (const file of data.files) { const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id) if (existingRedundancy) { - await this.extendsOrDeleteRedundancy(existingRedundancy) + await this.extendsRedundancy(existingRedundancy) continue } - if (!video) { - logger.info('Video %s we want to duplicate does not existing anymore, skipping.', file.Video.url) + await this.createVideoFileRedundancy(data.redundancy, video, file) + } + + for (const streamingPlaylist of data.streamingPlaylists) { + const existingRedundancy = await VideoRedundancyModel.loadLocalByStreamingPlaylistId(streamingPlaylist.id) + if (existingRedundancy) { + await this.extendsRedundancy(existingRedundancy) continue } - logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy) + await this.createStreamingPlaylistRedundancy(data.redundancy, video, streamingPlaylist) + } + } + + private async createVideoFileRedundancy (redundancy: VideosRedundancy, video: MVideoAccountLight, fileArg: MVideoFile) { + const file = fileArg as MVideoFileVideo + file.Video = video - const { baseUrlHttp, baseUrlWs } = video.getBaseUrls() - const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs) + const serverActor = await getServerActor() - const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT) + logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy) - const destPath = join(CONFIG.STORAGE.VIDEOS_DIR, video.getVideoFilename(file)) - await rename(tmpPath, destPath) + const { baseUrlHttp, baseUrlWs } = video.getBaseUrls() + const magnetUri = generateMagnetUri(video, file, baseUrlHttp, baseUrlWs) - const createdModel = await VideoRedundancyModel.create({ - expiresOn: this.buildNewExpiration(redundancy.minLifetime), - url: getVideoCacheFileActivityPubUrl(file), - fileUrl: video.getVideoFileUrl(file, CONFIG.WEBSERVER.URL), - strategy: redundancy.strategy, - videoFileId: file.id, - actorId: serverActor.id - }) - createdModel.VideoFile = file + const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT) - await sendCreateCacheFile(serverActor, createdModel) + const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, getVideoFilename(video, file)) + await move(tmpPath, destPath, { overwrite: true }) - logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url) - } + const createdModel: MVideoRedundancyFileVideo = await VideoRedundancyModel.create({ + expiresOn: this.buildNewExpiration(redundancy.minLifetime), + url: getVideoCacheFileActivityPubUrl(file), + fileUrl: video.getVideoRedundancyUrl(file, WEBSERVER.URL), + strategy: redundancy.strategy, + videoFileId: file.id, + actorId: serverActor.id + }) + + createdModel.VideoFile = file + + await sendCreateCacheFile(serverActor, video, createdModel) + + logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url) + } + + private async createStreamingPlaylistRedundancy ( + redundancy: VideosRedundancy, + video: MVideoAccountLight, + playlistArg: MStreamingPlaylist + ) { + const playlist = playlistArg as MStreamingPlaylistVideo + playlist.Video = video + + const serverActor = await getServerActor() + + logger.info('Duplicating %s streaming playlist in videos redundancy with "%s" strategy.', video.url, redundancy.strategy) + + const destDirectory = join(HLS_REDUNDANCY_DIRECTORY, video.uuid) + await downloadPlaylistSegments(playlist.playlistUrl, destDirectory, VIDEO_IMPORT_TIMEOUT) + + const createdModel: MVideoRedundancyStreamingPlaylistVideo = await VideoRedundancyModel.create({ + expiresOn: this.buildNewExpiration(redundancy.minLifetime), + url: getVideoCacheStreamingPlaylistActivityPubUrl(video, playlist), + fileUrl: playlist.getVideoRedundancyUrl(WEBSERVER.URL), + strategy: redundancy.strategy, + videoStreamingPlaylistId: playlist.id, + actorId: serverActor.id + }) + + createdModel.VideoStreamingPlaylist = playlist + + await sendCreateCacheFile(serverActor, video, createdModel) + + logger.info('Duplicated playlist %s -> %s.', playlist.playlistUrl, createdModel.url) } - private async extendsExpirationOf (redundancy: VideoRedundancyModel, expiresAfterMs: number) { + private async extendsExpirationOf (redundancy: MVideoRedundancyVideo, expiresAfterMs: number) { logger.info('Extending expiration of %s.', redundancy.url) const serverActor = await getServerActor() @@ -175,20 +261,21 @@ export class VideosRedundancyScheduler extends AbstractScheduler { await sendUpdateCacheFile(serverActor, redundancy) } - private async purgeCacheIfNeeded (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { - while (this.isTooHeavy(redundancy, filesToDuplicate)) { - const toDelete = await VideoRedundancyModel.loadOldestLocalThatAlreadyExpired(redundancy.strategy, redundancy.minLifetime) + private async purgeCacheIfNeeded (candidateToDuplicate: CandidateToDuplicate) { + while (await this.isTooHeavy(candidateToDuplicate)) { + const redundancy = candidateToDuplicate.redundancy + const toDelete = await VideoRedundancyModel.loadOldestLocalExpired(redundancy.strategy, redundancy.minLifetime) if (!toDelete) return await removeVideoRedundancy(toDelete) } } - private async isTooHeavy (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { - const maxSize = redundancy.size + private async isTooHeavy (candidateToDuplicate: CandidateToDuplicate) { + const maxSize = candidateToDuplicate.redundancy.size - const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(redundancy.strategy) - const totalWillDuplicate = totalDuplicated + this.getTotalFileSizes(filesToDuplicate) + const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(candidateToDuplicate.redundancy.strategy) + const totalWillDuplicate = totalDuplicated + this.getTotalFileSizes(candidateToDuplicate.files, candidateToDuplicate.streamingPlaylists) return totalWillDuplicate > maxSize } @@ -197,14 +284,21 @@ export class VideosRedundancyScheduler extends AbstractScheduler { return new Date(Date.now() + expiresAfterMs) } - private buildEntryLogId (object: VideoRedundancyModel) { - return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}` + private buildEntryLogId (object: MVideoRedundancyFileVideo | MVideoRedundancyStreamingPlaylistVideo) { + if (isMVideoRedundancyFileVideo(object)) return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}` + + return `${object.VideoStreamingPlaylist.playlistUrl}` } - private getTotalFileSizes (files: VideoFileModel[]) { - const fileReducer = (previous: number, current: VideoFileModel) => previous + current.size + private getTotalFileSizes (files: MVideoFile[], playlists: MStreamingPlaylistFiles[]) { + const fileReducer = (previous: number, current: MVideoFile) => previous + current.size + + let allFiles = files + for (const p of playlists) { + allFiles = allFiles.concat(p.VideoFiles) + } - return files.reduce(fileReducer, 0) + return allFiles.reduce(fileReducer, 0) } private async loadAndRefreshVideo (videoUrl: string) {