server/lib/live-manager.ts

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