server/lib/live-manager.ts

   1
   2 import * as chokidar from 'chokidar'
   3 import { FfmpegCommand } from 'fluent-ffmpeg'
   4 import { copy, ensureDir, stat } from 'fs-extra'
   5 import { basename, join } from 'path'
   6 import { isTestInstance } from '@server/helpers/core-utils'
   7 import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers/ffmpeg-utils'
   8 import { computeResolutionsToTranscode, getVideoFileFPS, getVideoFileResolution } from '@server/helpers/ffprobe-utils'
   9 import { logger } from '@server/helpers/logger'
  10 import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
  11 import { MEMOIZE_TTL, P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE, VIEW_LIFETIME, WEBSERVER } from '@server/initializers/constants'
  12 import { UserModel } from '@server/models/account/user'
  13 import { VideoModel } from '@server/models/video/video'
  14 import { VideoFileModel } from '@server/models/video/video-file'
  15 import { VideoLiveModel } from '@server/models/video/video-live'
  16 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
  17 import { MStreamingPlaylist, MUserId, MVideoLive, MVideoLiveVideo } from '@server/types/models'
  18 import { VideoState, VideoStreamingPlaylistType } from '@shared/models'
  19 import { federateVideoIfNeeded } from './activitypub/videos'
  20 import { buildSha256Segment } from './hls'
  21 import { JobQueue } from './job-queue'
  22 import { cleanupLive } from './job-queue/handlers/video-live-ending'
  23 import { PeerTubeSocket } from './peertube-socket'
  24 import { isAbleToUploadVideo } from './user'
  25 import { getHLSDirectory } from './video-paths'
  26 import { availableEncoders } from './video-transcoding-profiles'
  27
  28 import memoizee = require('memoizee')
  29
  30 const NodeRtmpServer = require('node-media-server/node_rtmp_server')
  31 const context = require('node-media-server/node_core_ctx')
  32 const nodeMediaServerLogger = require('node-media-server/node_core_logger')
  33
  34 // Disable node media server logs
  35 nodeMediaServerLogger.setLogType(0)
  36
  37 const config = {
  38   rtmp: {
  39     port: CONFIG.LIVE.RTMP.PORT,
  40     chunk_size: VIDEO_LIVE.RTMP.CHUNK_SIZE,
  41     gop_cache: VIDEO_LIVE.RTMP.GOP_CACHE,
  42     ping: VIDEO_LIVE.RTMP.PING,
  43     ping_timeout: VIDEO_LIVE.RTMP.PING_TIMEOUT
  44   },
  45   transcoding: {
  46     ffmpeg: 'ffmpeg'
  47   }
  48 }
  49
  50 class LiveManager {
  51
  52   private static instance: LiveManager
  53
  54   private readonly transSessions = new Map<string, FfmpegCommand>()
  55   private readonly videoSessions = new Map<number, string>()
  56   // Values are Date().getTime()
  57   private readonly watchersPerVideo = new Map<number, number[]>()
  58   private readonly segmentsSha256 = new Map<string, Map<string, string>>()
  59   private readonly livesPerUser = new Map<number, { liveId: number, videoId: number, size: number }[]>()
  60
  61   private readonly isAbleToUploadVideoWithCache = memoizee((userId: number) => {
  62     return isAbleToUploadVideo(userId, 1000)
  63   }, { maxAge: MEMOIZE_TTL.LIVE_ABLE_TO_UPLOAD })
  64
  65   private rtmpServer: any
  66
  67   private constructor () {
  68   }
  69
  70   init () {
  71     const events = this.getContext().nodeEvent
  72     events.on('postPublish', (sessionId: string, streamPath: string) => {
  73       logger.debug('RTMP received stream', { id: sessionId, streamPath })
  74
  75       const splittedPath = streamPath.split('/')
  76       if (splittedPath.length !== 3 || splittedPath[1] !== VIDEO_LIVE.RTMP.BASE_PATH) {
  77         logger.warn('Live path is incorrect.', { streamPath })
  78         return this.abortSession(sessionId)
  79       }
  80
  81       this.handleSession(sessionId, streamPath, splittedPath[2])
  82         .catch(err => logger.error('Cannot handle sessions.', { err }))
  83     })
  84
  85     events.on('donePublish', sessionId => {
  86       logger.info('Live session ended.', { sessionId })
  87     })
  88
  89     registerConfigChangedHandler(() => {
  90       if (!this.rtmpServer && CONFIG.LIVE.ENABLED === true) {
  91         this.run()
  92         return
  93       }
  94
  95       if (this.rtmpServer && CONFIG.LIVE.ENABLED === false) {
  96         this.stop()
  97       }
  98     })
  99
 100     // Cleanup broken lives, that were terminated by a server restart for example
 101     this.handleBrokenLives()
 102       .catch(err => logger.error('Cannot handle broken lives.', { err }))
 103
 104     setInterval(() => this.updateLiveViews(), VIEW_LIFETIME.LIVE)
 105   }
 106
 107   run () {
 108     logger.info('Running RTMP server on port %d', config.rtmp.port)
 109
 110     this.rtmpServer = new NodeRtmpServer(config)
 111     this.rtmpServer.run()
 112   }
 113
 114   stop () {
 115     logger.info('Stopping RTMP server.')
 116
 117     this.rtmpServer.stop()
 118     this.rtmpServer = undefined
 119   }
 120
 121   isRunning () {
 122     return !!this.rtmpServer
 123   }
 124
 125   getSegmentsSha256 (videoUUID: string) {
 126     return this.segmentsSha256.get(videoUUID)
 127   }
 128
 129   stopSessionOf (videoId: number) {
 130     const sessionId = this.videoSessions.get(videoId)
 131     if (!sessionId) return
 132
 133     this.videoSessions.delete(videoId)
 134     this.abortSession(sessionId)
 135   }
 136
 137   getLiveQuotaUsedByUser (userId: number) {
 138     const currentLives = this.livesPerUser.get(userId)
 139     if (!currentLives) return 0
 140
 141     return currentLives.reduce((sum, obj) => sum + obj.size, 0)
 142   }
 143
 144   addViewTo (videoId: number) {
 145     if (this.videoSessions.has(videoId) === false) return
 146
 147     let watchers = this.watchersPerVideo.get(videoId)
 148
 149     if (!watchers) {
 150       watchers = []
 151       this.watchersPerVideo.set(videoId, watchers)
 152     }
 153
 154     watchers.push(new Date().getTime())
 155   }
 156
 157   cleanupShaSegments (videoUUID: string) {
 158     this.segmentsSha256.delete(videoUUID)
 159   }
 160
 161   private getContext () {
 162     return context
 163   }
 164
 165   private abortSession (id: string) {
 166     const session = this.getContext().sessions.get(id)
 167     if (session) {
 168       session.stop()
 169       this.getContext().sessions.delete(id)
 170     }
 171
 172     const transSession = this.transSessions.get(id)
 173     if (transSession) {
 174       transSession.kill('SIGINT')
 175       this.transSessions.delete(id)
 176     }
 177   }
 178
 179   private async handleSession (sessionId: string, streamPath: string, streamKey: string) {
 180     const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
 181     if (!videoLive) {
 182       logger.warn('Unknown live video with stream key %s.', streamKey)
 183       return this.abortSession(sessionId)
 184     }
 185
 186     const video = videoLive.Video
 187     if (video.isBlacklisted()) {
 188       logger.warn('Video is blacklisted. Refusing stream %s.', streamKey)
 189       return this.abortSession(sessionId)
 190     }
 191
 192     // Cleanup old potential live files (could happen with a permanent live)
 193     this.cleanupShaSegments(video.uuid)
 194
 195     const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
 196     if (oldStreamingPlaylist) {
 197       await cleanupLive(video, oldStreamingPlaylist)
 198     }
 199
 200     this.videoSessions.set(video.id, sessionId)
 201
 202     const playlistUrl = WEBSERVER.URL + VideoStreamingPlaylistModel.getHlsMasterPlaylistStaticPath(video.uuid)
 203
 204     const session = this.getContext().sessions.get(sessionId)
 205     const rtmpUrl = 'rtmp://127.0.0.1:' + config.rtmp.port + streamPath
 206
 207     const [ resolutionResult, fps ] = await Promise.all([
 208       getVideoFileResolution(rtmpUrl),
 209       getVideoFileFPS(rtmpUrl)
 210     ])
 211
 212     const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED
 213       ? computeResolutionsToTranscode(resolutionResult.videoFileResolution, 'live')
 214       : []
 215
 216     const allResolutions = resolutionsEnabled.concat([ session.videoHeight ])
 217
 218     logger.info('Will mux/transcode live video of original resolution %d.', session.videoHeight, { allResolutions })
 219
 220     const [ videoStreamingPlaylist ] = await VideoStreamingPlaylistModel.upsert({
 221       videoId: video.id,
 222       playlistUrl,
 223       segmentsSha256Url: WEBSERVER.URL + VideoStreamingPlaylistModel.getHlsSha256SegmentsStaticPath(video.uuid, video.isLive),
 224       p2pMediaLoaderInfohashes: VideoStreamingPlaylistModel.buildP2PMediaLoaderInfoHashes(playlistUrl, allResolutions),
 225       p2pMediaLoaderPeerVersion: P2P_MEDIA_LOADER_PEER_VERSION,
 226
 227       type: VideoStreamingPlaylistType.HLS
 228     }, { returning: true }) as [ MStreamingPlaylist, boolean ]
 229
 230     return this.runMuxing({
 231       sessionId,
 232       videoLive,
 233       playlist: videoStreamingPlaylist,
 234       rtmpUrl,
 235       fps,
 236       allResolutions
 237     })
 238   }
 239
 240   private async runMuxing (options: {
 241     sessionId: string
 242     videoLive: MVideoLiveVideo
 243     playlist: MStreamingPlaylist
 244     rtmpUrl: string
 245     fps: number
 246     allResolutions: number[]
 247   }) {
 248     const { sessionId, videoLive, playlist, allResolutions, fps, rtmpUrl } = options
 249     const startStreamDateTime = new Date().getTime()
 250
 251     const user = await UserModel.loadByLiveId(videoLive.id)
 252     if (!this.livesPerUser.has(user.id)) {
 253       this.livesPerUser.set(user.id, [])
 254     }
 255
 256     const currentUserLive = { liveId: videoLive.id, videoId: videoLive.videoId, size: 0 }
 257     const livesOfUser = this.livesPerUser.get(user.id)
 258     livesOfUser.push(currentUserLive)
 259
 260     for (let i = 0; i < allResolutions.length; i++) {
 261       const resolution = allResolutions[i]
 262
 263       VideoFileModel.upsert({
 264         resolution,
 265         size: -1,
 266         extname: '.ts',
 267         infoHash: null,
 268         fps,
 269         videoStreamingPlaylistId: playlist.id
 270       }).catch(err => {
 271         logger.error('Cannot create file for live streaming.', { err })
 272       })
 273     }
 274
 275     const outPath = getHLSDirectory(videoLive.Video)
 276     await ensureDir(outPath)
 277
 278     const replayDirectory = join(outPath, VIDEO_LIVE.REPLAY_DIRECTORY)
 279
 280     if (videoLive.saveReplay === true) {
 281       await ensureDir(replayDirectory)
 282     }
 283
 284     const videoUUID = videoLive.Video.uuid
 285
 286     const ffmpegExec = CONFIG.LIVE.TRANSCODING.ENABLED
 287       ? await getLiveTranscodingCommand({
 288         rtmpUrl,
 289         outPath,
 290         resolutions: allResolutions,
 291         fps,
 292         availableEncoders,
 293         profile: 'default'
 294       })
 295       : getLiveMuxingCommand(rtmpUrl, outPath)
 296
 297     logger.info('Running live muxing/transcoding for %s.', videoUUID)
 298     this.transSessions.set(sessionId, ffmpegExec)
 299
 300     const tsWatcher = chokidar.watch(outPath + '/*.ts')
 301
 302     const segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {}
 303     const playlistIdMatcher = /^([\d+])-/
 304
 305     const processSegments = (segmentsToProcess: string[]) => {
 306       // Add sha hash of previous segments, because ffmpeg should have finished generating them
 307       for (const previousSegment of segmentsToProcess) {
 308         this.addSegmentSha(videoUUID, previousSegment)
 309           .catch(err => logger.error('Cannot add sha segment of video %s -> %s.', videoUUID, previousSegment, { err }))
 310
 311         if (videoLive.saveReplay) {
 312           const segmentName = basename(previousSegment)
 313
 314           copy(previousSegment, join(outPath, VIDEO_LIVE.REPLAY_DIRECTORY, segmentName))
 315             .catch(err => logger.error('Cannot copy segment %s to repay directory.', previousSegment, { err }))
 316         }
 317       }
 318     }
 319
 320     const addHandler = segmentPath => {
 321       logger.debug('Live add handler of %s.', segmentPath)
 322
 323       const playlistId = basename(segmentPath).match(playlistIdMatcher)[0]
 324
 325       const segmentsToProcess = segmentsToProcessPerPlaylist[playlistId] || []
 326       processSegments(segmentsToProcess)
 327
 328       segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ]
 329
 330       // Duration constraint check
 331       if (this.isDurationConstraintValid(startStreamDateTime) !== true) {
 332         logger.info('Stopping session of %s: max duration exceeded.', videoUUID)
 333
 334         this.stopSessionOf(videoLive.videoId)
 335       }
 336
 337       // Check user quota if the user enabled replay saving
 338       if (videoLive.saveReplay === true) {
 339         stat(segmentPath)
 340           .then(segmentStat => {
 341             currentUserLive.size += segmentStat.size
 342           })
 343           .then(() => this.isQuotaConstraintValid(user, videoLive))
 344           .then(quotaValid => {
 345             if (quotaValid !== true) {
 346               logger.info('Stopping session of %s: user quota exceeded.', videoUUID)
 347
 348               this.stopSessionOf(videoLive.videoId)
 349             }
 350           })
 351           .catch(err => logger.error('Cannot stat %s or check quota of %d.', segmentPath, user.id, { err }))
 352       }
 353     }
 354
 355     const deleteHandler = segmentPath => this.removeSegmentSha(videoUUID, segmentPath)
 356
 357     tsWatcher.on('add', p => addHandler(p))
 358     tsWatcher.on('unlink', p => deleteHandler(p))
 359
 360     const masterWatcher = chokidar.watch(outPath + '/master.m3u8')
 361     masterWatcher.on('add', async () => {
 362       try {
 363         const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoLive.videoId)
 364
 365         video.state = VideoState.PUBLISHED
 366         await video.save()
 367         videoLive.Video = video
 368
 369         setTimeout(() => {
 370           federateVideoIfNeeded(video, false)
 371             .catch(err => logger.error('Cannot federate live video %s.', video.url, { err }))
 372
 373           PeerTubeSocket.Instance.sendVideoLiveNewState(video)
 374         }, VIDEO_LIVE.SEGMENT_TIME_SECONDS * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION)
 375
 376       } catch (err) {
 377         logger.error('Cannot save/federate live video %d.', videoLive.videoId, { err })
 378       } finally {
 379         masterWatcher.close()
 380           .catch(err => logger.error('Cannot close master watcher of %s.', outPath, { err }))
 381       }
 382     })
 383
 384     const onFFmpegEnded = () => {
 385       logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', rtmpUrl)
 386
 387       this.transSessions.delete(sessionId)
 388
 389       this.watchersPerVideo.delete(videoLive.videoId)
 390       this.videoSessions.delete(videoLive.videoId)
 391
 392       const newLivesPerUser = this.livesPerUser.get(user.id)
 393                                                .filter(o => o.liveId !== videoLive.id)
 394       this.livesPerUser.set(user.id, newLivesPerUser)
 395
 396       setTimeout(() => {
 397         // Wait latest segments generation, and close watchers
 398
 399         Promise.all([ tsWatcher.close(), masterWatcher.close() ])
 400           .then(() => {
 401             // Process remaining segments hash
 402             for (const key of Object.keys(segmentsToProcessPerPlaylist)) {
 403               processSegments(segmentsToProcessPerPlaylist[key])
 404             }
 405           })
 406           .catch(err => logger.error('Cannot close watchers of %s or process remaining hash segments.', outPath, { err }))
 407
 408         this.onEndTransmuxing(videoLive.Video.id)
 409           .catch(err => logger.error('Error in closed transmuxing.', { err }))
 410       }, 1000)
 411     }
 412
 413     ffmpegExec.on('error', (err, stdout, stderr) => {
 414       onFFmpegEnded()
 415
 416       // Don't care that we killed the ffmpeg process
 417       if (err?.message?.includes('Exiting normally')) return
 418
 419       logger.error('Live transcoding error.', { err, stdout, stderr })
 420
 421       this.abortSession(sessionId)
 422     })
 423
 424     ffmpegExec.on('end', () => onFFmpegEnded())
 425
 426     ffmpegExec.run()
 427   }
 428
 429   private async onEndTransmuxing (videoId: number, cleanupNow = false) {
 430     try {
 431       const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
 432       if (!fullVideo) return
 433
 434       const live = await VideoLiveModel.loadByVideoId(videoId)
 435
 436       if (!live.permanentLive) {
 437         JobQueue.Instance.createJob({
 438           type: 'video-live-ending',
 439           payload: {
 440             videoId: fullVideo.id
 441           }
 442         }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY })
 443
 444         fullVideo.state = VideoState.LIVE_ENDED
 445       } else {
 446         fullVideo.state = VideoState.WAITING_FOR_LIVE
 447       }
 448
 449       await fullVideo.save()
 450
 451       PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
 452
 453       await federateVideoIfNeeded(fullVideo, false)
 454     } catch (err) {
 455       logger.error('Cannot save/federate new video state of live streaming.', { err })
 456     }
 457   }
 458
 459   private async addSegmentSha (videoUUID: string, segmentPath: string) {
 460     const segmentName = basename(segmentPath)
 461     logger.debug('Adding live sha segment %s.', segmentPath)
 462
 463     const shaResult = await buildSha256Segment(segmentPath)
 464
 465     if (!this.segmentsSha256.has(videoUUID)) {
 466       this.segmentsSha256.set(videoUUID, new Map())
 467     }
 468
 469     const filesMap = this.segmentsSha256.get(videoUUID)
 470     filesMap.set(segmentName, shaResult)
 471   }
 472
 473   private removeSegmentSha (videoUUID: string, segmentPath: string) {
 474     const segmentName = basename(segmentPath)
 475
 476     logger.debug('Removing live sha segment %s.', segmentPath)
 477
 478     const filesMap = this.segmentsSha256.get(videoUUID)
 479     if (!filesMap) {
 480       logger.warn('Unknown files map to remove sha for %s.', videoUUID)
 481       return
 482     }
 483
 484     if (!filesMap.has(segmentName)) {
 485       logger.warn('Unknown segment in files map for video %s and segment %s.', videoUUID, segmentPath)
 486       return
 487     }
 488
 489     filesMap.delete(segmentName)
 490   }
 491
 492   private isDurationConstraintValid (streamingStartTime: number) {
 493     const maxDuration = CONFIG.LIVE.MAX_DURATION
 494     // No limit
 495     if (maxDuration === null) return true
 496
 497     const now = new Date().getTime()
 498     const max = streamingStartTime + maxDuration
 499
 500     return now <= max
 501   }
 502
 503   private async isQuotaConstraintValid (user: MUserId, live: MVideoLive) {
 504     if (live.saveReplay !== true) return true
 505
 506     return this.isAbleToUploadVideoWithCache(user.id)
 507   }
 508
 509   private async updateLiveViews () {
 510     if (!this.isRunning()) return
 511
 512     if (!isTestInstance()) logger.info('Updating live video views.')
 513
 514     for (const videoId of this.watchersPerVideo.keys()) {
 515       const notBefore = new Date().getTime() - VIEW_LIFETIME.LIVE
 516
 517       const watchers = this.watchersPerVideo.get(videoId)
 518
 519       const numWatchers = watchers.length
 520
 521       const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
 522       video.views = numWatchers
 523       await video.save()
 524
 525       await federateVideoIfNeeded(video, false)
 526
 527       // Only keep not expired watchers
 528       const newWatchers = watchers.filter(w => w > notBefore)
 529       this.watchersPerVideo.set(videoId, newWatchers)
 530
 531       logger.debug('New live video views for %s is %d.', video.url, numWatchers)
 532     }
 533   }
 534
 535   private async handleBrokenLives () {
 536     const videoIds = await VideoModel.listPublishedLiveIds()
 537
 538     for (const id of videoIds) {
 539       await this.onEndTransmuxing(id, true)
 540     }
 541   }
 542
 543   static get Instance () {
 544     return this.instance || (this.instance = new this())
 545   }
 546 }
 547
 548 // ---------------------------------------------------------------------------
 549
 550 export {
 551   LiveManager
 552 }