server/lib/schedulers/videos-redundancy-scheduler.ts

   1 import { AbstractScheduler } from './abstract-scheduler'
   2 import { HLS_REDUNDANCY_DIRECTORY, REDUNDANCY, VIDEO_IMPORT_TIMEOUT, WEBSERVER } from '../../initializers/constants'
   3 import { logger } from '../../helpers/logger'
   4 import { VideosRedundancy } from '../../../shared/models/redundancy'
   5 import { VideoRedundancyModel } from '../../models/redundancy/video-redundancy'
   6 import { downloadWebTorrentVideo, generateMagnetUri } from '../../helpers/webtorrent'
   7 import { join } from 'path'
   8 import { move } from 'fs-extra'
   9 import { getServerActor } from '../../helpers/utils'
  10 import { sendCreateCacheFile, sendUpdateCacheFile } from '../activitypub/send'
  11 import { getVideoCacheFileActivityPubUrl, getVideoCacheStreamingPlaylistActivityPubUrl } from '../activitypub/url'
  12 import { removeVideoRedundancy } from '../redundancy'
  13 import { getOrCreateVideoAndAccountAndChannel } from '../activitypub'
  14 import { downloadPlaylistSegments } from '../hls'
  15 import { CONFIG } from '../../initializers/config'
  16 import {
  17   MStreamingPlaylist, MStreamingPlaylistFiles,
  18   MStreamingPlaylistVideo,
  19   MVideoAccountLight,
  20   MVideoFile,
  21   MVideoFileVideo,
  22   MVideoRedundancyFileVideo,
  23   MVideoRedundancyStreamingPlaylistVideo,
  24   MVideoRedundancyVideo,
  25   MVideoWithAllFiles
  26 } from '@server/typings/models'
  27 import { getVideoFilename } from '../video-paths'
  28
  29 type CandidateToDuplicate = {
  30   redundancy: VideosRedundancy,
  31   video: MVideoWithAllFiles,
  32   files: MVideoFile[],
  33   streamingPlaylists: MStreamingPlaylistFiles[]
  34 }
  35
  36 function isMVideoRedundancyFileVideo (
  37   o: MVideoRedundancyFileVideo | MVideoRedundancyStreamingPlaylistVideo
  38 ): o is MVideoRedundancyFileVideo {
  39   return !!(o as MVideoRedundancyFileVideo).VideoFile
  40 }
  41
  42 export class VideosRedundancyScheduler extends AbstractScheduler {
  43
  44   private static instance: AbstractScheduler
  45
  46   protected schedulerIntervalMs = CONFIG.REDUNDANCY.VIDEOS.CHECK_INTERVAL
  47
  48   private constructor () {
  49     super()
  50   }
  51
  52   protected async internalExecute () {
  53     for (const redundancyConfig of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) {
  54       logger.info('Running redundancy scheduler for strategy %s.', redundancyConfig.strategy)
  55
  56       try {
  57         const videoToDuplicate = await this.findVideoToDuplicate(redundancyConfig)
  58         if (!videoToDuplicate) continue
  59
  60         const candidateToDuplicate = {
  61           video: videoToDuplicate,
  62           redundancy: redundancyConfig,
  63           files: videoToDuplicate.VideoFiles,
  64           streamingPlaylists: videoToDuplicate.VideoStreamingPlaylists
  65         }
  66
  67         await this.purgeCacheIfNeeded(candidateToDuplicate)
  68
  69         if (await this.isTooHeavy(candidateToDuplicate)) {
  70           logger.info('Video %s is too big for our cache, skipping.', videoToDuplicate.url)
  71           continue
  72         }
  73
  74         logger.info('Will duplicate video %s in redundancy scheduler "%s".', videoToDuplicate.url, redundancyConfig.strategy)
  75
  76         await this.createVideoRedundancies(candidateToDuplicate)
  77       } catch (err) {
  78         logger.error('Cannot run videos redundancy %s.', redundancyConfig.strategy, { err })
  79       }
  80     }
  81
  82     await this.extendsLocalExpiration()
  83
  84     await this.purgeRemoteExpired()
  85   }
  86
  87   static get Instance () {
  88     return this.instance || (this.instance = new this())
  89   }
  90
  91   private async extendsLocalExpiration () {
  92     const expired = await VideoRedundancyModel.listLocalExpired()
  93
  94     for (const redundancyModel of expired) {
  95       try {
  96         const redundancyConfig = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy)
  97         const candidate = {
  98           redundancy: redundancyConfig,
  99           video: null,
 100           files: [],
 101           streamingPlaylists: []
 102         }
 103
 104         // If the administrator disabled the redundancy or decreased the cache size, remove this redundancy instead of extending it
 105         if (!redundancyConfig || await this.isTooHeavy(candidate)) {
 106           logger.info('Destroying redundancy %s because the cache size %s is too heavy.', redundancyModel.url, redundancyModel.strategy)
 107           await removeVideoRedundancy(redundancyModel)
 108         } else {
 109           await this.extendsRedundancy(redundancyModel)
 110         }
 111       } catch (err) {
 112         logger.error(
 113           'Cannot extend or remove expiration of %s video from our redundancy system.', this.buildEntryLogId(redundancyModel),
 114           { err }
 115         )
 116       }
 117     }
 118   }
 119
 120   private async extendsRedundancy (redundancyModel: MVideoRedundancyVideo) {
 121     const redundancy = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy)
 122     // Redundancy strategy disabled, remove our redundancy instead of extending expiration
 123     if (!redundancy) {
 124       await removeVideoRedundancy(redundancyModel)
 125       return
 126     }
 127
 128     await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime)
 129   }
 130
 131   private async purgeRemoteExpired () {
 132     const expired = await VideoRedundancyModel.listRemoteExpired()
 133
 134     for (const redundancyModel of expired) {
 135       try {
 136         await removeVideoRedundancy(redundancyModel)
 137       } catch (err) {
 138         logger.error('Cannot remove redundancy %s from our redundancy system.', this.buildEntryLogId(redundancyModel))
 139       }
 140     }
 141   }
 142
 143   private findVideoToDuplicate (cache: VideosRedundancy) {
 144     if (cache.strategy === 'most-views') {
 145       return VideoRedundancyModel.findMostViewToDuplicate(REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR)
 146     }
 147
 148     if (cache.strategy === 'trending') {
 149       return VideoRedundancyModel.findTrendingToDuplicate(REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR)
 150     }
 151
 152     if (cache.strategy === 'recently-added') {
 153       const minViews = cache.minViews
 154       return VideoRedundancyModel.findRecentlyAddedToDuplicate(REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR, minViews)
 155     }
 156   }
 157
 158   private async createVideoRedundancies (data: CandidateToDuplicate) {
 159     const video = await this.loadAndRefreshVideo(data.video.url)
 160
 161     if (!video) {
 162       logger.info('Video %s we want to duplicate does not existing anymore, skipping.', data.video.url)
 163
 164       return
 165     }
 166
 167     for (const file of data.files) {
 168       const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id)
 169       if (existingRedundancy) {
 170         await this.extendsRedundancy(existingRedundancy)
 171
 172         continue
 173       }
 174
 175       await this.createVideoFileRedundancy(data.redundancy, video, file)
 176     }
 177
 178     for (const streamingPlaylist of data.streamingPlaylists) {
 179       const existingRedundancy = await VideoRedundancyModel.loadLocalByStreamingPlaylistId(streamingPlaylist.id)
 180       if (existingRedundancy) {
 181         await this.extendsRedundancy(existingRedundancy)
 182
 183         continue
 184       }
 185
 186       await this.createStreamingPlaylistRedundancy(data.redundancy, video, streamingPlaylist)
 187     }
 188   }
 189
 190   private async createVideoFileRedundancy (redundancy: VideosRedundancy, video: MVideoAccountLight, fileArg: MVideoFile) {
 191     const file = fileArg as MVideoFileVideo
 192     file.Video = video
 193
 194     const serverActor = await getServerActor()
 195
 196     logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy)
 197
 198     const { baseUrlHttp, baseUrlWs } = video.getBaseUrls()
 199     const magnetUri = generateMagnetUri(video, file, baseUrlHttp, baseUrlWs)
 200
 201     const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT)
 202
 203     const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, getVideoFilename(video, file))
 204     await move(tmpPath, destPath, { overwrite: true })
 205
 206     const createdModel: MVideoRedundancyFileVideo = await VideoRedundancyModel.create({
 207       expiresOn: this.buildNewExpiration(redundancy.minLifetime),
 208       url: getVideoCacheFileActivityPubUrl(file),
 209       fileUrl: video.getVideoRedundancyUrl(file, WEBSERVER.URL),
 210       strategy: redundancy.strategy,
 211       videoFileId: file.id,
 212       actorId: serverActor.id
 213     })
 214
 215     createdModel.VideoFile = file
 216
 217     await sendCreateCacheFile(serverActor, video, createdModel)
 218
 219     logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url)
 220   }
 221
 222   private async createStreamingPlaylistRedundancy (
 223     redundancy: VideosRedundancy,
 224     video: MVideoAccountLight,
 225     playlistArg: MStreamingPlaylist
 226   ) {
 227     const playlist = playlistArg as MStreamingPlaylistVideo
 228     playlist.Video = video
 229
 230     const serverActor = await getServerActor()
 231
 232     logger.info('Duplicating %s streaming playlist in videos redundancy with "%s" strategy.', video.url, redundancy.strategy)
 233
 234     const destDirectory = join(HLS_REDUNDANCY_DIRECTORY, video.uuid)
 235     await downloadPlaylistSegments(playlist.playlistUrl, destDirectory, VIDEO_IMPORT_TIMEOUT)
 236
 237     const createdModel: MVideoRedundancyStreamingPlaylistVideo = await VideoRedundancyModel.create({
 238       expiresOn: this.buildNewExpiration(redundancy.minLifetime),
 239       url: getVideoCacheStreamingPlaylistActivityPubUrl(video, playlist),
 240       fileUrl: playlist.getVideoRedundancyUrl(WEBSERVER.URL),
 241       strategy: redundancy.strategy,
 242       videoStreamingPlaylistId: playlist.id,
 243       actorId: serverActor.id
 244     })
 245
 246     createdModel.VideoStreamingPlaylist = playlist
 247
 248     await sendCreateCacheFile(serverActor, video, createdModel)
 249
 250     logger.info('Duplicated playlist %s -> %s.', playlist.playlistUrl, createdModel.url)
 251   }
 252
 253   private async extendsExpirationOf (redundancy: MVideoRedundancyVideo, expiresAfterMs: number) {
 254     logger.info('Extending expiration of %s.', redundancy.url)
 255
 256     const serverActor = await getServerActor()
 257
 258     redundancy.expiresOn = this.buildNewExpiration(expiresAfterMs)
 259     await redundancy.save()
 260
 261     await sendUpdateCacheFile(serverActor, redundancy)
 262   }
 263
 264   private async purgeCacheIfNeeded (candidateToDuplicate: CandidateToDuplicate) {
 265     while (await this.isTooHeavy(candidateToDuplicate)) {
 266       const redundancy = candidateToDuplicate.redundancy
 267       const toDelete = await VideoRedundancyModel.loadOldestLocalExpired(redundancy.strategy, redundancy.minLifetime)
 268       if (!toDelete) return
 269
 270       await removeVideoRedundancy(toDelete)
 271     }
 272   }
 273
 274   private async isTooHeavy (candidateToDuplicate: CandidateToDuplicate) {
 275     const maxSize = candidateToDuplicate.redundancy.size
 276
 277     const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(candidateToDuplicate.redundancy.strategy)
 278     const totalWillDuplicate = totalDuplicated + this.getTotalFileSizes(candidateToDuplicate.files, candidateToDuplicate.streamingPlaylists)
 279
 280     return totalWillDuplicate > maxSize
 281   }
 282
 283   private buildNewExpiration (expiresAfterMs: number) {
 284     return new Date(Date.now() + expiresAfterMs)
 285   }
 286
 287   private buildEntryLogId (object: MVideoRedundancyFileVideo | MVideoRedundancyStreamingPlaylistVideo) {
 288     if (isMVideoRedundancyFileVideo(object)) return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}`
 289
 290     return `${object.VideoStreamingPlaylist.playlistUrl}`
 291   }
 292
 293   private getTotalFileSizes (files: MVideoFile[], playlists: MStreamingPlaylistFiles[]) {
 294     const fileReducer = (previous: number, current: MVideoFile) => previous + current.size
 295
 296     let allFiles = files
 297     for (const p of playlists) {
 298       allFiles = allFiles.concat(p.VideoFiles)
 299     }
 300
 301     return allFiles.reduce(fileReducer, 0)
 302   }
 303
 304   private async loadAndRefreshVideo (videoUrl: string) {
 305     // We need more attributes and check if the video still exists
 306     const getVideoOptions = {
 307       videoObject: videoUrl,
 308       syncParam: { likes: false, dislikes: false, shares: false, comments: false, thumbnail: false, refreshVideo: true },
 309       fetchType: 'all' as 'all'
 310     }
 311     const { video } = await getOrCreateVideoAndAccountAndChannel(getVideoOptions)
 312
 313     return video
 314   }
 315 }