aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/schedulers/videos-redundancy-scheduler.ts
blob: 16562ad0bcb7053de3532de13835075a73e63a89 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
import { move } from 'fs-extra'
import { join } from 'path'
import { getServerActor } from '@server/models/application/application'
import { VideoModel } from '@server/models/video/video'
import {
  MStreamingPlaylistFiles,
  MVideoAccountLight,
  MVideoFile,
  MVideoFileVideo,
  MVideoRedundancyFileVideo,
  MVideoRedundancyStreamingPlaylistVideo,
  MVideoRedundancyVideo,
  MVideoWithAllFiles
} from '@server/types/models'
import { VideosRedundancyStrategy } from '../../../shared/models/redundancy'
import { logger, loggerTagsFactory } from '../../helpers/logger'
import { downloadWebTorrentVideo } from '../../helpers/webtorrent'
import { CONFIG } from '../../initializers/config'
import { HLS_REDUNDANCY_DIRECTORY, REDUNDANCY, VIDEO_IMPORT_TIMEOUT } from '../../initializers/constants'
import { VideoRedundancyModel } from '../../models/redundancy/video-redundancy'
import { sendCreateCacheFile, sendUpdateCacheFile } from '../activitypub/send'
import { getLocalVideoCacheFileActivityPubUrl, getLocalVideoCacheStreamingPlaylistActivityPubUrl } from '../activitypub/url'
import { getOrCreateAPVideo } from '../activitypub/videos'
import { downloadPlaylistSegments } from '../hls'
import { removeVideoRedundancy } from '../redundancy'
import { generateHLSRedundancyUrl, generateWebTorrentRedundancyUrl } from '../video-urls'
import { AbstractScheduler } from './abstract-scheduler'

const lTags = loggerTagsFactory('redundancy')

type CandidateToDuplicate = {
  redundancy: VideosRedundancyStrategy
  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: VideosRedundancyScheduler

  protected schedulerIntervalMs = CONFIG.REDUNDANCY.VIDEOS.CHECK_INTERVAL

  private constructor () {
    super()
  }

  async createManualRedundancy (videoId: number) {
    const videoToDuplicate = await VideoModel.loadWithFiles(videoId)

    if (!videoToDuplicate) {
      logger.warn('Video to manually duplicate %d does not exist anymore.', videoId, lTags())
      return
    }

    return this.createVideoRedundancies({
      video: videoToDuplicate,
      redundancy: null,
      files: videoToDuplicate.VideoFiles,
      streamingPlaylists: videoToDuplicate.VideoStreamingPlaylists
    })
  }

  protected async internalExecute () {
    for (const redundancyConfig of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) {
      logger.info('Running redundancy scheduler for strategy %s.', redundancyConfig.strategy, lTags())

      try {
        const videoToDuplicate = await this.findVideoToDuplicate(redundancyConfig)
        if (!videoToDuplicate) continue

        const candidateToDuplicate = {
          video: videoToDuplicate,
          redundancy: redundancyConfig,
          files: videoToDuplicate.VideoFiles,
          streamingPlaylists: videoToDuplicate.VideoStreamingPlaylists
        }

        await this.purgeCacheIfNeeded(candidateToDuplicate)

        if (await this.isTooHeavy(candidateToDuplicate)) {
          logger.info('Video %s is too big for our cache, skipping.', videoToDuplicate.url, lTags(videoToDuplicate.uuid))
          continue
        }

        logger.info(
          'Will duplicate video %s in redundancy scheduler "%s".',
          videoToDuplicate.url, redundancyConfig.strategy, lTags(videoToDuplicate.uuid)
        )

        await this.createVideoRedundancies(candidateToDuplicate)
      } catch (err) {
        logger.error('Cannot run videos redundancy %s.', redundancyConfig.strategy, { err, ...lTags() })
      }
    }

    await this.extendsLocalExpiration()

    await this.purgeRemoteExpired()
  }

  static get Instance () {
    return this.instance || (this.instance = new this())
  }

  private async extendsLocalExpiration () {
    const expired = await VideoRedundancyModel.listLocalExpired()

    for (const redundancyModel of expired) {
      try {
        const redundancyConfig = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy)
        const { totalUsed } = await VideoRedundancyModel.getStats(redundancyConfig.strategy)

        // If the administrator disabled the redundancy or decreased the cache size, remove this redundancy instead of extending it
        if (!redundancyConfig || totalUsed > redundancyConfig.size) {
          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 or remove expiration of %s video from our redundancy system.',
          this.buildEntryLogId(redundancyModel), { err, ...lTags(redundancyModel.getVideoUUID()) }
        )
      }
    }
  }

  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
    }

    await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime)
  }

  private async purgeRemoteExpired () {
    const expired = await VideoRedundancyModel.listRemoteExpired()

    for (const redundancyModel of expired) {
      try {
        await removeVideoRedundancy(redundancyModel)
      } catch (err) {
        logger.error(
          'Cannot remove redundancy %s from our redundancy system.',
          this.buildEntryLogId(redundancyModel), lTags(redundancyModel.getVideoUUID())
        )
      }
    }
  }

  private findVideoToDuplicate (cache: VideosRedundancyStrategy) {
    if (cache.strategy === 'most-views') {
      return VideoRedundancyModel.findMostViewToDuplicate(REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR)
    }

    if (cache.strategy === 'trending') {
      return VideoRedundancyModel.findTrendingToDuplicate(REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR)
    }

    if (cache.strategy === 'recently-added') {
      const minViews = cache.minViews
      return VideoRedundancyModel.findRecentlyAddedToDuplicate(REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR, minViews)
    }
  }

  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, lTags(data.video.uuid))

      return
    }

    for (const file of data.files) {
      const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id)
      if (existingRedundancy) {
        await this.extendsRedundancy(existingRedundancy)

        continue
      }

      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
      }

      await this.createStreamingPlaylistRedundancy(data.redundancy, video, streamingPlaylist)
    }
  }

  private async createVideoFileRedundancy (redundancy: VideosRedundancyStrategy | null, video: MVideoAccountLight, fileArg: MVideoFile) {
    let strategy = 'manual'
    let expiresOn: Date = null

    if (redundancy) {
      strategy = redundancy.strategy
      expiresOn = this.buildNewExpiration(redundancy.minLifetime)
    }

    const file = fileArg as MVideoFileVideo
    file.Video = video

    const serverActor = await getServerActor()

    logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, strategy, lTags(video.uuid))

    const tmpPath = await downloadWebTorrentVideo({ uri: file.torrentUrl }, VIDEO_IMPORT_TIMEOUT)

    const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, file.filename)
    await move(tmpPath, destPath, { overwrite: true })

    const createdModel: MVideoRedundancyFileVideo = await VideoRedundancyModel.create({
      expiresOn,
      url: getLocalVideoCacheFileActivityPubUrl(file),
      fileUrl: generateWebTorrentRedundancyUrl(file),
      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, lTags(video.uuid))
  }

  private async createStreamingPlaylistRedundancy (
    redundancy: VideosRedundancyStrategy,
    video: MVideoAccountLight,
    playlistArg: MStreamingPlaylistFiles
  ) {
    let strategy = 'manual'
    let expiresOn: Date = null

    if (redundancy) {
      strategy = redundancy.strategy
      expiresOn = this.buildNewExpiration(redundancy.minLifetime)
    }

    const playlist = Object.assign(playlistArg, { Video: video })
    const serverActor = await getServerActor()

    logger.info('Duplicating %s streaming playlist in videos redundancy with "%s" strategy.', video.url, strategy, lTags(video.uuid))

    const destDirectory = join(HLS_REDUNDANCY_DIRECTORY, video.uuid)
    const masterPlaylistUrl = playlist.getMasterPlaylistUrl(video)

    const maxSizeKB = this.getTotalFileSizes([], [ playlist ]) / 1000
    const toleranceKB = maxSizeKB + ((5 * maxSizeKB) / 100) // 5% more tolerance
    await downloadPlaylistSegments(masterPlaylistUrl, destDirectory, VIDEO_IMPORT_TIMEOUT, toleranceKB)

    const createdModel: MVideoRedundancyStreamingPlaylistVideo = await VideoRedundancyModel.create({
      expiresOn,
      url: getLocalVideoCacheStreamingPlaylistActivityPubUrl(video, playlist),
      fileUrl: generateHLSRedundancyUrl(video, playlistArg),
      strategy,
      videoStreamingPlaylistId: playlist.id,
      actorId: serverActor.id
    })

    createdModel.VideoStreamingPlaylist = playlist

    await sendCreateCacheFile(serverActor, video, createdModel)

    logger.info('Duplicated playlist %s -> %s.', masterPlaylistUrl, createdModel.url, lTags(video.uuid))
  }

  private async extendsExpirationOf (redundancy: MVideoRedundancyVideo, expiresAfterMs: number) {
    logger.info('Extending expiration of %s.', redundancy.url, lTags(redundancy.getVideoUUID()))

    const serverActor = await getServerActor()

    redundancy.expiresOn = this.buildNewExpiration(expiresAfterMs)
    await redundancy.save()

    await sendUpdateCacheFile(serverActor, redundancy)
  }

  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

      const videoId = toDelete.VideoFile
        ? toDelete.VideoFile.videoId
        : toDelete.VideoStreamingPlaylist.videoId

      const redundancies = await VideoRedundancyModel.listLocalByVideoId(videoId)

      for (const redundancy of redundancies) {
        await removeVideoRedundancy(redundancy)
      }
    }
  }

  private async isTooHeavy (candidateToDuplicate: CandidateToDuplicate) {
    const maxSize = candidateToDuplicate.redundancy.size

    const { totalUsed: alreadyUsed } = await VideoRedundancyModel.getStats(candidateToDuplicate.redundancy.strategy)

    const videoSize = this.getTotalFileSizes(candidateToDuplicate.files, candidateToDuplicate.streamingPlaylists)
    const willUse = alreadyUsed + videoSize

    logger.debug('Checking candidate size.', { maxSize, alreadyUsed, videoSize, willUse, ...lTags(candidateToDuplicate.video.uuid) })

    return willUse > maxSize
  }

  private buildNewExpiration (expiresAfterMs: number) {
    return new Date(Date.now() + expiresAfterMs)
  }

  private buildEntryLogId (object: MVideoRedundancyFileVideo | MVideoRedundancyStreamingPlaylistVideo) {
    if (isMVideoRedundancyFileVideo(object)) return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}`

    return `${object.VideoStreamingPlaylist.getMasterPlaylistUrl(object.VideoStreamingPlaylist.Video)}`
  }

  private getTotalFileSizes (files: MVideoFile[], playlists: MStreamingPlaylistFiles[]): number {
    const fileReducer = (previous: number, current: MVideoFile) => previous + current.size

    let allFiles = files
    for (const p of playlists) {
      allFiles = allFiles.concat(p.VideoFiles)
    }

    return allFiles.reduce(fileReducer, 0)
  }

  private async loadAndRefreshVideo (videoUrl: string) {
    // We need more attributes and check if the video still exists
    const getVideoOptions = {
      videoObject: videoUrl,
      syncParam: { likes: false, dislikes: false, shares: false, comments: false, thumbnail: false, refreshVideo: true },
      fetchType: 'all' as 'all'
    }
    const { video } = await getOrCreateAPVideo(getVideoOptions)

    return video
  }
}