aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/schedulers
diff options
context:
space:
mode:
authorChocobozzz <me@florianbigard.com>2019-02-11 14:09:23 +0100
committerChocobozzz <me@florianbigard.com>2019-02-11 14:09:23 +0100
commitb718fd22374d64534bcfe69932cf562894abed6a (patch)
tree311d3c67e2a4d1f33ebdd1dc163527de9d33d0f7 /server/lib/schedulers
parentadb115f5522bea4d52456a9fc5eb4140bb064476 (diff)
parent501e961199578129629cf0567033d13efced9904 (diff)
downloadPeerTube-b718fd22374d64534bcfe69932cf562894abed6a.tar.gz
PeerTube-b718fd22374d64534bcfe69932cf562894abed6a.tar.zst
PeerTube-b718fd22374d64534bcfe69932cf562894abed6a.zip
Merge branch 'develop' into pr/1285
Diffstat (limited to 'server/lib/schedulers')
-rw-r--r--server/lib/schedulers/videos-redundancy-scheduler.ts189
1 files changed, 130 insertions, 59 deletions
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 @@
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'
@@ -9,9 +9,19 @@ import { join } from 'path'
9import { move } 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
@@ -24,28 +34,32 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
24 } 34 }
25 35
26 protected async internalExecute () { 36 protected async internalExecute () {
27 for (const obj of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) { 37 for (const redundancyConfig of CONFIG.REDUNDANCY.VIDEOS.STRATEGIES) {
28 logger.info('Running redundancy scheduler for strategy %s.', obj.strategy) 38 logger.info('Running redundancy scheduler for strategy %s.', redundancyConfig.strategy)
29 39
30 try { 40 try {
31 const videoToDuplicate = await this.findVideoToDuplicate(obj) 41 const videoToDuplicate = await this.findVideoToDuplicate(redundancyConfig)
32 if (!videoToDuplicate) continue 42 if (!videoToDuplicate) continue
33 43
34 const videoFiles = videoToDuplicate.VideoFiles 44 const candidateToDuplicate = {
35 videoFiles.forEach(f => f.Video = videoToDuplicate) 45 video: videoToDuplicate,
46 redundancy: redundancyConfig,
47 files: videoToDuplicate.VideoFiles,
48 streamingPlaylists: videoToDuplicate.VideoStreamingPlaylists
49 }
36 50
37 await this.purgeCacheIfNeeded(obj, videoFiles) 51 await this.purgeCacheIfNeeded(candidateToDuplicate)
38 52
39 if (await this.isTooHeavy(obj, videoFiles)) { 53 if (await this.isTooHeavy(candidateToDuplicate)) {
40 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)
41 continue 55 continue
42 } 56 }
43 57
44 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)
45 59
46 await this.createVideoRedundancy(obj, videoFiles) 60 await this.createVideoRedundancies(candidateToDuplicate)
47 } catch (err) { 61 } catch (err) {
48 logger.error('Cannot run videos redundancy %s.', obj.strategy, { err }) 62 logger.error('Cannot run videos redundancy %s.', redundancyConfig.strategy, { err })
49 } 63 }
50 } 64 }
51 65
@@ -63,25 +77,35 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
63 77
64 for (const redundancyModel of expired) { 78 for (const redundancyModel of expired) {
65 try { 79 try {
66 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 }
67 } catch (err) { 95 } catch (err) {
68 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 )
69 } 100 }
70 } 101 }
71 } 102 }
72 103
73 private async extendsOrDeleteRedundancy (redundancyModel: VideoRedundancyModel) { 104 private async extendsRedundancy (redundancyModel: VideoRedundancyModel) {
74 // Refresh the video, maybe it was deleted
75 const video = await this.loadAndRefreshVideo(redundancyModel.VideoFile.Video.url)
76
77 if (!video) {
78 logger.info('Destroying existing redundancy %s, because the associated video does not exist anymore.', redundancyModel.url)
79
80 await redundancyModel.destroy()
81 return
82 }
83
84 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
85 await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime) 109 await this.extendsExpirationOf(redundancyModel, redundancy.minLifetime)
86 } 110 }
87 111
@@ -112,49 +136,93 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
112 } 136 }
113 } 137 }
114 138
115 private async createVideoRedundancy (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { 139 private async createVideoRedundancies (data: CandidateToDuplicate) {
116 const serverActor = await getServerActor() 140 const video = await this.loadAndRefreshVideo(data.video.url)
141
142 if (!video) {
143 logger.info('Video %s we want to duplicate does not existing anymore, skipping.', data.video.url)
117 144
118 for (const file of filesToDuplicate) { 145 return
119 const video = await this.loadAndRefreshVideo(file.Video.url) 146 }
120 147
148 for (const file of data.files) {
121 const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id) 149 const existingRedundancy = await VideoRedundancyModel.loadLocalByFileId(file.id)
122 if (existingRedundancy) { 150 if (existingRedundancy) {
123 await this.extendsOrDeleteRedundancy(existingRedundancy) 151 await this.extendsRedundancy(existingRedundancy)
124 152
125 continue 153 continue
126 } 154 }
127 155
128 if (!video) { 156 await this.createVideoFileRedundancy(data.redundancy, video, file)
129 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)
130 163
131 continue 164 continue
132 } 165 }
133 166
134 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 }
135 170
136 const { baseUrlHttp, baseUrlWs } = video.getBaseUrls() 171 private async createVideoFileRedundancy (redundancy: VideosRedundancy, video: VideoModel, file: VideoFileModel) {
137 const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs) 172 file.Video = video
138 173
139 const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT) 174 const serverActor = await getServerActor()
140 175
141 const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, video.getVideoFilename(file)) 176 logger.info('Duplicating %s - %d in videos redundancy with "%s" strategy.', video.url, file.resolution, redundancy.strategy)
142 await move(tmpPath, destPath)
143 177
144 const createdModel = await VideoRedundancyModel.create({ 178 const { baseUrlHttp, baseUrlWs } = video.getBaseUrls()
145 expiresOn: this.buildNewExpiration(redundancy.minLifetime), 179 const magnetUri = video.generateMagnetUri(file, baseUrlHttp, baseUrlWs)
146 url: getVideoCacheFileActivityPubUrl(file),
147 fileUrl: video.getVideoRedundancyUrl(file, CONFIG.WEBSERVER.URL),
148 strategy: redundancy.strategy,
149 videoFileId: file.id,
150 actorId: serverActor.id
151 })
152 createdModel.VideoFile = file
153 180
154 await sendCreateCacheFile(serverActor, createdModel) 181 const tmpPath = await downloadWebTorrentVideo({ magnetUri }, VIDEO_IMPORT_TIMEOUT)
155 182
156 logger.info('Duplicated %s - %d -> %s.', video.url, file.resolution, createdModel.url) 183 const destPath = join(CONFIG.STORAGE.REDUNDANCY_DIR, video.getVideoFilename(file))
157 } 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)
158 } 226 }
159 227
160 private async extendsExpirationOf (redundancy: VideoRedundancyModel, expiresAfterMs: number) { 228 private async extendsExpirationOf (redundancy: VideoRedundancyModel, expiresAfterMs: number) {
@@ -168,8 +236,9 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
168 await sendUpdateCacheFile(serverActor, redundancy) 236 await sendUpdateCacheFile(serverActor, redundancy)
169 } 237 }
170 238
171 private async purgeCacheIfNeeded (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { 239 private async purgeCacheIfNeeded (candidateToDuplicate: CandidateToDuplicate) {
172 while (this.isTooHeavy(redundancy, filesToDuplicate)) { 240 while (this.isTooHeavy(candidateToDuplicate)) {
241 const redundancy = candidateToDuplicate.redundancy
173 const toDelete = await VideoRedundancyModel.loadOldestLocalThatAlreadyExpired(redundancy.strategy, redundancy.minLifetime) 242 const toDelete = await VideoRedundancyModel.loadOldestLocalThatAlreadyExpired(redundancy.strategy, redundancy.minLifetime)
174 if (!toDelete) return 243 if (!toDelete) return
175 244
@@ -177,11 +246,11 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
177 } 246 }
178 } 247 }
179 248
180 private async isTooHeavy (redundancy: VideosRedundancy, filesToDuplicate: VideoFileModel[]) { 249 private async isTooHeavy (candidateToDuplicate: CandidateToDuplicate) {
181 const maxSize = redundancy.size 250 const maxSize = candidateToDuplicate.redundancy.size
182 251
183 const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(redundancy.strategy) 252 const totalDuplicated = await VideoRedundancyModel.getTotalDuplicated(candidateToDuplicate.redundancy.strategy)
184 const totalWillDuplicate = totalDuplicated + this.getTotalFileSizes(filesToDuplicate) 253 const totalWillDuplicate = totalDuplicated + this.getTotalFileSizes(candidateToDuplicate.files, candidateToDuplicate.streamingPlaylists)
185 254
186 return totalWillDuplicate > maxSize 255 return totalWillDuplicate > maxSize
187 } 256 }
@@ -191,13 +260,15 @@ export class VideosRedundancyScheduler extends AbstractScheduler {
191 } 260 }
192 261
193 private buildEntryLogId (object: VideoRedundancyModel) { 262 private buildEntryLogId (object: VideoRedundancyModel) {
194 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}`
195 } 266 }
196 267
197 private getTotalFileSizes (files: VideoFileModel[]) { 268 private getTotalFileSizes (files: VideoFileModel[], playlists: VideoStreamingPlaylistModel[]) {
198 const fileReducer = (previous: number, current: VideoFileModel) => previous + current.size 269 const fileReducer = (previous: number, current: VideoFileModel) => previous + current.size
199 270
200 return files.reduce(fileReducer, 0) 271 return files.reduce(fileReducer, 0) * playlists.length
201 } 272 }
202 273
203 private async loadAndRefreshVideo (videoUrl: string) { 274 private async loadAndRefreshVideo (videoUrl: string) {