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

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