aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/schedulers/videos-redundancy-scheduler.ts
diff options
context:
space:
mode:
Diffstat (limited to 'server/lib/schedulers/videos-redundancy-scheduler.ts')
-rw-r--r--server/lib/schedulers/videos-redundancy-scheduler.ts201
1 files changed, 133 insertions, 68 deletions
diff --git a/server/lib/schedulers/videos-redundancy-scheduler.ts b/server/lib/schedulers/videos-redundancy-scheduler.ts
index c49a8c89a..1a48f2bd0 100644
--- a/server/lib/schedulers/videos-redundancy-scheduler.ts
+++ b/server/lib/schedulers/videos-redundancy-scheduler.ts
@@ -1,22 +1,31 @@
1import { AbstractScheduler } from './abstract-scheduler' 1import { AbstractScheduler } from './abstract-scheduler'
2import { CONFIG, REDUNDANCY, VIDEO_IMPORT_TIMEOUT } from '../../initializers' 2import { CONFIG, HLS_REDUNDANCY_DIRECTORY, REDUNDANCY, VIDEO_IMPORT_TIMEOUT } from '../../initializers'
3import { logger } from '../../helpers/logger' 3import { logger } from '../../helpers/logger'
4import { VideosRedundancy } from '../../../shared/models/redundancy' 4import { VideosRedundancy } from '../../../shared/models/redundancy'
5import { VideoRedundancyModel } from '../../models/redundancy/video-redundancy' 5import { VideoRedundancyModel } from '../../models/redundancy/video-redundancy'
6import { VideoFileModel } from '../../models/video/video-file' 6import { VideoFileModel } from '../../models/video/video-file'
7import { downloadWebTorrentVideo } from '../../helpers/webtorrent' 7import { downloadWebTorrentVideo } from '../../helpers/webtorrent'
8import { join } from 'path' 8import { join } from 'path'
9import { rename } from 'fs-extra' 9import { move } from 'fs-extra'
10import { getServerActor } from '../../helpers/utils' 10import { getServerActor } from '../../helpers/utils'
11import { sendCreateCacheFile, sendUpdateCacheFile } from '../activitypub/send' 11import { sendCreateCacheFile, sendUpdateCacheFile } from '../activitypub/send'
12import { getVideoCacheFileActivityPubUrl } from '../activitypub/url' 12import { getVideoCacheFileActivityPubUrl, getVideoCacheStreamingPlaylistActivityPubUrl } from '../activitypub/url'
13import { removeVideoRedundancy } from '../redundancy' 13import { removeVideoRedundancy } from '../redundancy'
14import { getOrCreateVideoAndAccountAndChannel } from '../activitypub' 14import { getOrCreateVideoAndAccountAndChannel } from '../activitypub'
15import { VideoStreamingPlaylistModel } from '../../models/video/video-streaming-playlist'
16import { VideoModel } from '../../models/video/video'
17import { downloadPlaylistSegments } from '../hls'
18
19type CandidateToDuplicate = {
20 redundancy: VideosRedundancy,
21 video: VideoModel,
22 files: VideoFileModel[],
23 streamingPlaylists: VideoStreamingPlaylistModel[]
24}
15 25
16export class VideosRedundancyScheduler extends AbstractScheduler { 26export class VideosRedundancyScheduler extends AbstractScheduler {
17 27
18 private static instance: AbstractScheduler 28 private static instance: AbstractScheduler
19 private executing = false
20 29
21 protected schedulerIntervalMs = CONFIG.REDUNDANCY.VIDEOS.CHECK_INTERVAL 30 protected schedulerIntervalMs = CONFIG.REDUNDANCY.VIDEOS.CHECK_INTERVAL
22 31
@@ -24,41 +33,39 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
24 super() 33 super()
25 } 34 }
26 35
27 async execute () { 36 protected async internalExecute () {
28 if (this.executing) return 37 for (const redundancyConfig of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) {
29 38 logger.info('Running redundancy scheduler for strategy %s.', redundancyConfig.strategy)
30 this.executing = true
31
32 for (const obj of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) {
33 logger.info('Running redundancy scheduler for strategy %s.', obj.strategy)
34 39
35 try { 40 try {
36 const videoToDuplicate = await this.findVideoToDuplicate(obj) 41 const videoToDuplicate = await this.findVideoToDuplicate(redundancyConfig)
37 if (!videoToDuplicate) continue 42 if (!videoToDuplicate) continue
38 43
39 const videoFiles = videoToDuplicate.VideoFiles 44 const candidateToDuplicate = {
40 videoFiles.forEach(f => f.Video = videoToDuplicate) 45 video: videoToDuplicate,
46 redundancy: redundancyConfig,
47 files: videoToDuplicate.VideoFiles,
48 streamingPlaylists: videoToDuplicate.VideoStreamingPlaylists
49 }
41 50
42 await this.purgeCacheIfNeeded(obj, videoFiles) 51 await this.purgeCacheIfNeeded(candidateToDuplicate)
43 52
44 if (await this.isTooHeavy(obj, videoFiles)) { 53 if (await this.isTooHeavy(candidateToDuplicate)) {
45 logger.info('Video %s is too big for our cache, skipping.', videoToDuplicate.url) 54 logger.info('Video %s is too big for our cache, skipping.', videoToDuplicate.url)
46 continue 55 continue
47 } 56 }
48 57
49 logger.info('Will duplicate video %s in redundancy scheduler "%s".', videoToDuplicate.url, obj.strategy) 58 logger.info('Will duplicate video %s in redundancy scheduler "%s".', videoToDuplicate.url, redundancyConfig.strategy)
50 59
51 await this.createVideoRedundancy(obj, videoFiles) 60 await this.createVideoRedundancies(candidateToDuplicate)
52 } catch (err) { 61 } catch (err) {
53 logger.error('Cannot run videos redundancy %s.', obj.strategy, { err }) 62 logger.error('Cannot run videos redundancy %s.', redundancyConfig.strategy, { err })
54 } 63 }
55 } 64 }
56 65
57 await this.extendsLocalExpiration() 66 await this.extendsLocalExpiration()
58 67
59 await this.purgeRemoteExpired() 68 await this.purgeRemoteExpired()
60
61 this.executing = false
62 } 69 }
63 70
64 static get Instance () { 71 static get Instance () {
@@ -70,25 +77,35 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
70 77
71 for (const redundancyModel of expired) { 78 for (const redundancyModel of expired) {
72 try { 79 try {
73 await this.extendsOrDeleteRedundancy(redundancyModel) 80 const redundancyConfig = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy)
81 const candidate = {
82 redundancy: redundancyConfig,
83 video: null,
84 files: [],
85 streamingPlaylists: []
86 }
87
88 // If the administrator disabled the redundancy or decreased the cache size, remove this redundancy instead of extending it
89 if (!redundancyConfig || await this.isTooHeavy(candidate)) {
90 logger.info('Destroying redundancy %s because the cache size %s is too heavy.', redundancyModel.url, redundancyModel.strategy)
91 await removeVideoRedundancy(redundancyModel)
92 } else {
93 await this.extendsRedundancy(redundancyModel)
94 }
74 } catch (err) { 95 } catch (err) {
75 logger.error('Cannot extend expiration of %s video from our redundancy system.', this.buildEntryLogId(redundancyModel)) 96 logger.error(
97 'Cannot extend or remove expiration of %s video from our redundancy system.', this.buildEntryLogId(redundancyModel),
98 { err }
99 )
76 } 100 }
77 } 101 }
78 } 102 }
79 103
80 private async extendsOrDeleteRedundancy (redundancyModel: VideoRedundancyModel) { 104 private async extendsRedundancy (redundancyModel: VideoRedundancyModel) {
81 // Refresh the video, maybe it was deleted
82 const video = await this.loadAndRefreshVideo(redundancyModel.VideoFile.Video.url)
83
84 if (!video) {
85 logger.info('Destroying existing redundancy %s, because the associated video does not exist anymore.', redundancyModel.url)
86
87 await redundancyModel.destroy()
88 return
89 }
90
91 const redundancy = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy) 105 const redundancy = CONFIG.REDUNDANCY.VIDEOS.STRATEGIES.find(s => s.strategy === redundancyModel.strategy)
106 // Redundancy strategy disabled, remove our redundancy instead of extending expiration
107 if (!redundancy) await removeVideoRedundancy(redundancyModel)
108
92 await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime) 109 await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime)
93 } 110 }
94 111
@@ -119,49 +136,93 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
119 } 136 }
120 } 137 }
121 138
122 private async createVideoRedundancy (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { 139 private async createVideoRedundancies (data: CandidateToDuplicate) {
123 const serverActor = await getServerActor() 140 const video = await this.loadAndRefreshVideo(data.video.url)
124 141
125 for (const file of filesToDuplicate) { 142 if (!video) {
126 const video = await this.loadAndRefreshVideo(file.Video.url) 143 logger.info('Video %s we want to duplicate does not existing anymore, skipping.', data.video.url)
127 144
145 return
146 }
147
148 for (const file of data.files) {
128 const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id) 149 const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id)
129 if (existingRedundancy) { 150 if (existingRedundancy) {
130 await this.extendsOrDeleteRedundancy(existingRedundancy) 151 await this.extendsRedundancy(existingRedundancy)
131 152
132 continue 153 continue
133 } 154 }
134 155
135 if (!video) { 156 await this.createVideoFileRedundancy(data.redundancy, video, file)
136 logger.info('Video %s we want to duplicate does not existing anymore, skipping.', file.Video.url) 157 }
158
159 for (const streamingPlaylist of data.streamingPlaylists) {
160 const existingRedundancy = await VideoRedundancyModel.loadLocalByStreamingPlaylistId(streamingPlaylist.id)
161 if (existingRedundancy) {
162 await this.extendsRedundancy(existingRedundancy)
137 163
138 continue 164 continue
139 } 165 }
140 166
141 logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy) 167 await this.createStreamingPlaylistRedundancy(data.redundancy, video, streamingPlaylist)
168 }
169 }
142 170
143 const { baseUrlHttp, baseUrlWs } = video.getBaseUrls() 171 private async createVideoFileRedundancy (redundancy: VideosRedundancy, video: VideoModel, file: VideoFileModel) {
144 const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs) 172 file.Video = video
145 173
146 const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT) 174 const serverActor = await getServerActor()
147 175
148 const destPath = join(CONFIG.STORAGE.VIDEOS_DIR, video.getVideoFilename(file)) 176 logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy)
149 await rename(tmpPath, destPath)
150 177
151 const createdModel = await VideoRedundancyModel.create({ 178 const { baseUrlHttp, baseUrlWs } = video.getBaseUrls()
152 expiresOn: this.buildNewExpiration(redundancy.minLifetime), 179 const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs)
153 url: getVideoCacheFileActivityPubUrl(file),
154 fileUrl: video.getVideoFileUrl(file, CONFIG.WEBSERVER.URL),
155 strategy: redundancy.strategy,
156 videoFileId: file.id,
157 actorId: serverActor.id
158 })
159 createdModel.VideoFile = file
160 180
161 await sendCreateCacheFile(serverActor, createdModel) 181 const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT)
162 182
163 logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url) 183 const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, video.getVideoFilename(file))
164 } 184 await move(tmpPath, destPath)
185
186 const createdModel = await VideoRedundancyModel.create({
187 expiresOn: this.buildNewExpiration(redundancy.minLifetime),
188 url: getVideoCacheFileActivityPubUrl(file),
189 fileUrl: video.getVideoRedundancyUrl(file, CONFIG.WEBSERVER.URL),
190 strategy: redundancy.strategy,
191 videoFileId: file.id,
192 actorId: serverActor.id
193 })
194
195 createdModel.VideoFile = file
196
197 await sendCreateCacheFile(serverActor, video, createdModel)
198
199 logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url)
200 }
201
202 private async createStreamingPlaylistRedundancy (redundancy: VideosRedundancy, video: VideoModel, playlist: VideoStreamingPlaylistModel) {
203 playlist.Video = video
204
205 const serverActor = await getServerActor()
206
207 logger.info('Duplicating %s streaming playlist in videos redundancy with "%s" strategy.', video.url, redundancy.strategy)
208
209 const destDirectory = join(HLS_REDUNDANCY_DIRECTORY, video.uuid)
210 await downloadPlaylistSegments(playlist.playlistUrl, destDirectory, VIDEO_IMPORT_TIMEOUT)
211
212 const createdModel = await VideoRedundancyModel.create({
213 expiresOn: this.buildNewExpiration(redundancy.minLifetime),
214 url: getVideoCacheStreamingPlaylistActivityPubUrl(video, playlist),
215 fileUrl: playlist.getVideoRedundancyUrl(CONFIG.WEBSERVER.URL),
216 strategy: redundancy.strategy,
217 videoStreamingPlaylistId: playlist.id,
218 actorId: serverActor.id
219 })
220
221 createdModel.VideoStreamingPlaylist = playlist
222
223 await sendCreateCacheFile(serverActor, video, createdModel)
224
225 logger.info('Duplicated playlist %s -> %s.', playlist.playlistUrl, createdModel.url)
165 } 226 }
166 227
167 private async extendsExpirationOf (redundancy: VideoRedundancyModel, expiresAfterMs: number) { 228 private async extendsExpirationOf (redundancy: VideoRedundancyModel, expiresAfterMs: number) {
@@ -175,8 +236,9 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
175 await sendUpdateCacheFile(serverActor, redundancy) 236 await sendUpdateCacheFile(serverActor, redundancy)
176 } 237 }
177 238
178 private async purgeCacheIfNeeded (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { 239 private async purgeCacheIfNeeded (candidateToDuplicate: CandidateToDuplicate) {
179 while (this.isTooHeavy(redundancy, filesToDuplicate)) { 240 while (this.isTooHeavy(candidateToDuplicate)) {
241 const redundancy = candidateToDuplicate.redundancy
180 const toDelete = await VideoRedundancyModel.loadOldestLocalThatAlreadyExpired(redundancy.strategy, redundancy.minLifetime) 242 const toDelete = await VideoRedundancyModel.loadOldestLocalThatAlreadyExpired(redundancy.strategy, redundancy.minLifetime)
181 if (!toDelete) return 243 if (!toDelete) return
182 244
@@ -184,12 +246,13 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
184 } 246 }
185 } 247 }
186 248
187 private async isTooHeavy (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { 249 private async isTooHeavy (candidateToDuplicate: CandidateToDuplicate) {
188 const maxSize = redundancy.size - this.getTotalFileSizes(filesToDuplicate) 250 const maxSize = candidateToDuplicate.redundancy.size
189 251
190 const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(redundancy.strategy) 252 const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(candidateToDuplicate.redundancy.strategy)
253 const totalWillDuplicate = totalDuplicated + this.getTotalFileSizes(candidateToDuplicate.files, candidateToDuplicate.streamingPlaylists)
191 254
192 return totalDuplicated > maxSize 255 return totalWillDuplicate > maxSize
193 } 256 }
194 257
195 private buildNewExpiration (expiresAfterMs: number) { 258 private buildNewExpiration (expiresAfterMs: number) {
@@ -197,13 +260,15 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
197 } 260 }
198 261
199 private buildEntryLogId (object: VideoRedundancyModel) { 262 private buildEntryLogId (object: VideoRedundancyModel) {
200 return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}` 263 if (object.VideoFile) return `${object.VideoFile.Video.url}-${object.VideoFile.resolution}`
264
265 return `${object.VideoStreamingPlaylist.playlistUrl}`
201 } 266 }
202 267
203 private getTotalFileSizes (files: VideoFileModel[]) { 268 private getTotalFileSizes (files: VideoFileModel[], playlists: VideoStreamingPlaylistModel[]) {
204 const fileReducer = (previous: number, current: VideoFileModel) => previous + current.size 269 const fileReducer = (previous: number, current: VideoFileModel) => previous + current.size
205 270
206 return files.reduce(fileReducer, 0) 271 return files.reduce(fileReducer, 0) * playlists.length
207 } 272 }
208 273
209 private async loadAndRefreshVideo (videoUrl: string) { 274 private async loadAndRefreshVideo (videoUrl: string) {