From 092092969633bbcf6d4891a083ea497a7d5c3154 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Tue, 29 Jan 2019 08:37:25 +0100 Subject: Add hls support on server --- .../lib/schedulers/videos-redundancy-scheduler.ts | 189 ++++++++++++++------- 1 file changed, 130 insertions(+), 59 deletions(-) (limited to 'server/lib/schedulers') diff --git a/server/lib/schedulers/videos-redundancy-scheduler.ts b/server/lib/schedulers/videos-redundancy-scheduler.ts index f643ee226..1a48f2bd0 100644 --- a/server/lib/schedulers/videos-redundancy-scheduler.ts +++ b/server/lib/schedulers/videos-redundancy-scheduler.ts @@ -1,5 +1,5 @@ import { AbstractScheduler } from './abstract-scheduler' -import { CONFIG, REDUNDANCY, VIDEO_IMPORT_TIMEOUT } from '../../initializers' +import { CONFIG, HLS_REDUNDANCY_DIRECTORY, REDUNDANCY, VIDEO_IMPORT_TIMEOUT } from '../../initializers' import { logger } from '../../helpers/logger' import { VideosRedundancy } from '../../../shared/models/redundancy' import { VideoRedundancyModel } from '../../models/redundancy/video-redundancy' @@ -9,9 +9,19 @@ import { join } from 'path' 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 { VideoStreamingPlaylistModel } from '../../models/video/video-streaming-playlist' +import { VideoModel } from '../../models/video/video' +import { downloadPlaylistSegments } from '../hls' + +type CandidateToDuplicate = { + redundancy: VideosRedundancy, + video: VideoModel, + files: VideoFileModel[], + streamingPlaylists: VideoStreamingPlaylistModel[] +} export class VideosRedundancyScheduler extends AbstractScheduler { @@ -24,28 +34,32 @@ export class VideosRedundancyScheduler extends AbstractScheduler { } protected async internalExecute () { - for (const obj of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) { - logger.info('Running redundancy scheduler for strategy %s.', obj.strategy) + 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 }) } } @@ -63,25 +77,35 @@ 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() - return - } - + private async extendsRedundancy (redundancyModel: VideoRedundancyModel) { 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) + await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime) } @@ -112,49 +136,93 @@ 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) + + if (!video) { + logger.info('Video %s we want to duplicate does not existing anymore, skipping.', data.video.url) - for (const file of filesToDuplicate) { - const video = await this.loadAndRefreshVideo(file.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) + } + } - const { baseUrlHttp, baseUrlWs } = video.getBaseUrls() - const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs) + private async createVideoFileRedundancy (redundancy: VideosRedundancy, video: VideoModel, file: VideoFileModel) { + file.Video = video - const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT) + const serverActor = await getServerActor() - const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, video.getVideoFilename(file)) - await move(tmpPath, destPath) + logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy) - const createdModel = await VideoRedundancyModel.create({ - expiresOn: this.buildNewExpiration(redundancy.minLifetime), - url: getVideoCacheFileActivityPubUrl(file), - fileUrl: video.getVideoRedundancyUrl(file, CONFIG.WEBSERVER.URL), - strategy: redundancy.strategy, - videoFileId: file.id, - actorId: serverActor.id - }) - createdModel.VideoFile = file + const { baseUrlHttp, baseUrlWs } = video.getBaseUrls() + const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs) - await sendCreateCacheFile(serverActor, createdModel) + const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT) - logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url) - } + const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, video.getVideoFilename(file)) + await move(tmpPath, destPath) + + const createdModel = await VideoRedundancyModel.create({ + expiresOn: this.buildNewExpiration(redundancy.minLifetime), + url: getVideoCacheFileActivityPubUrl(file), + fileUrl: video.getVideoRedundancyUrl(file, CONFIG.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: VideoModel, playlist: VideoStreamingPlaylistModel) { + 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 = await VideoRedundancyModel.create({ + expiresOn: this.buildNewExpiration(redundancy.minLifetime), + url: getVideoCacheStreamingPlaylistActivityPubUrl(video, playlist), + fileUrl: playlist.getVideoRedundancyUrl(CONFIG.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) { @@ -168,8 +236,9 @@ export class VideosRedundancyScheduler extends AbstractScheduler { await sendUpdateCacheFile(serverActor, redundancy) } - private async purgeCacheIfNeeded (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { - while (this.isTooHeavy(redundancy, filesToDuplicate)) { + private async purgeCacheIfNeeded (candidateToDuplicate: CandidateToDuplicate) { + while (this.isTooHeavy(candidateToDuplicate)) { + const redundancy = candidateToDuplicate.redundancy const toDelete = await VideoRedundancyModel.loadOldestLocalThatAlreadyExpired(redundancy.strategy, redundancy.minLifetime) if (!toDelete) return @@ -177,11 +246,11 @@ export class VideosRedundancyScheduler extends AbstractScheduler { } } - 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 } @@ -191,13 +260,15 @@ export class VideosRedundancyScheduler extends AbstractScheduler { } private buildEntryLogId (object: VideoRedundancyModel) { - return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}` + if (object.VideoFile) return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}` + + return `${object.VideoStreamingPlaylist.playlistUrl}` } - private getTotalFileSizes (files: VideoFileModel[]) { + private getTotalFileSizes (files: VideoFileModel[], playlists: VideoStreamingPlaylistModel[]) { const fileReducer = (previous: number, current: VideoFileModel) => previous + current.size - return files.reduce(fileReducer, 0) + return files.reduce(fileReducer, 0) * playlists.length } private async loadAndRefreshVideo (videoUrl: string) { -- cgit v1.2.3