server/lib/live/live-manager.ts

   1 import { readdir, readFile } from 'fs-extra'
   2 import { createServer, Server } from 'net'
   3 import { join } from 'path'
   4 import { createServer as createServerTLS, Server as ServerTLS } from 'tls'
   5 import { logger, loggerTagsFactory } from '@server/helpers/logger'
   6 import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
   7 import { VIDEO_LIVE, WEBSERVER } from '@server/initializers/constants'
   8 import { sequelizeTypescript } from '@server/initializers/database'
   9 import { RunnerJobModel } from '@server/models/runner/runner-job'
  10 import { UserModel } from '@server/models/user/user'
  11 import { VideoModel } from '@server/models/video/video'
  12 import { VideoLiveModel } from '@server/models/video/video-live'
  13 import { VideoLiveReplaySettingModel } from '@server/models/video/video-live-replay-setting'
  14 import { VideoLiveSessionModel } from '@server/models/video/video-live-session'
  15 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
  16 import { MVideo, MVideoLiveSession, MVideoLiveVideo, MVideoLiveVideoWithSetting } from '@server/types/models'
  17 import { pick, wait } from '@shared/core-utils'
  18 import { ffprobePromise, getVideoStreamBitrate, getVideoStreamDimensionsInfo, getVideoStreamFPS, hasAudioStream } from '@shared/ffmpeg'
  19 import { LiveVideoError, VideoState } from '@shared/models'
  20 import { federateVideoIfNeeded } from '../activitypub/videos'
  21 import { JobQueue } from '../job-queue'
  22 import { getLiveReplayBaseDirectory } from '../paths'
  23 import { PeerTubeSocket } from '../peertube-socket'
  24 import { Hooks } from '../plugins/hooks'
  25 import { computeResolutionsToTranscode } from '../transcoding/transcoding-resolutions'
  26 import { LiveQuotaStore } from './live-quota-store'
  27 import { cleanupAndDestroyPermanentLive, getLiveSegmentTime } from './live-utils'
  28 import { MuxingSession } from './shared'
  29
  30 const NodeRtmpSession = require('node-media-server/src/node_rtmp_session')
  31 const context = require('node-media-server/src/node_core_ctx')
  32 const nodeMediaServerLogger = require('node-media-server/src/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 }
  46
  47 const lTags = loggerTagsFactory('live')
  48
  49 class LiveManager {
  50
  51   private static instance: LiveManager
  52
  53   private readonly muxingSessions = new Map<string, MuxingSession>()
  54   private readonly videoSessions = new Map<string, string>()
  55
  56   private rtmpServer: Server
  57   private rtmpsServer: ServerTLS
  58
  59   private running = false
  60
  61   private constructor () {
  62   }
  63
  64   init () {
  65     const events = this.getContext().nodeEvent
  66     events.on('postPublish', (sessionId: string, streamPath: string) => {
  67       logger.debug('RTMP received stream', { id: sessionId, streamPath, ...lTags(sessionId) })
  68
  69       const splittedPath = streamPath.split('/')
  70       if (splittedPath.length !== 3 || splittedPath[1] !== VIDEO_LIVE.RTMP.BASE_PATH) {
  71         logger.warn('Live path is incorrect.', { streamPath, ...lTags(sessionId) })
  72         return this.abortSession(sessionId)
  73       }
  74
  75       const session = this.getContext().sessions.get(sessionId)
  76       const inputLocalUrl = session.inputOriginLocalUrl + streamPath
  77       const inputPublicUrl = session.inputOriginPublicUrl + streamPath
  78
  79       this.handleSession({ sessionId, inputPublicUrl, inputLocalUrl, streamKey: splittedPath[2] })
  80         .catch(err => logger.error('Cannot handle sessions.', { err, ...lTags(sessionId) }))
  81     })
  82
  83     events.on('donePublish', sessionId => {
  84       logger.info('Live session ended.', { sessionId, ...lTags(sessionId) })
  85
  86       // Force session aborting, so we kill ffmpeg even if it still has data to process (slow CPU)
  87       setTimeout(() => this.abortSession(sessionId), 2000)
  88     })
  89
  90     registerConfigChangedHandler(() => {
  91       if (!this.running && CONFIG.LIVE.ENABLED === true) {
  92         this.run().catch(err => logger.error('Cannot run live server.', { err }))
  93         return
  94       }
  95
  96       if (this.running && CONFIG.LIVE.ENABLED === false) {
  97         this.stop()
  98       }
  99     })
 100
 101     // Cleanup broken lives, that were terminated by a server restart for example
 102     this.handleBrokenLives()
 103       .catch(err => logger.error('Cannot handle broken lives.', { err, ...lTags() }))
 104   }
 105
 106   async run () {
 107     this.running = true
 108
 109     if (CONFIG.LIVE.RTMP.ENABLED) {
 110       logger.info('Running RTMP server on port %d', CONFIG.LIVE.RTMP.PORT, lTags())
 111
 112       this.rtmpServer = createServer(socket => {
 113         const session = new NodeRtmpSession(config, socket)
 114
 115         session.inputOriginLocalUrl = 'rtmp://127.0.0.1:' + CONFIG.LIVE.RTMP.PORT
 116         session.inputOriginPublicUrl = WEBSERVER.RTMP_URL
 117         session.run()
 118       })
 119
 120       this.rtmpServer.on('error', err => {
 121         logger.error('Cannot run RTMP server.', { err, ...lTags() })
 122       })
 123
 124       this.rtmpServer.listen(CONFIG.LIVE.RTMP.PORT, CONFIG.LIVE.RTMP.HOSTNAME)
 125     }
 126
 127     if (CONFIG.LIVE.RTMPS.ENABLED) {
 128       logger.info('Running RTMPS server on port %d', CONFIG.LIVE.RTMPS.PORT, lTags())
 129
 130       const [ key, cert ] = await Promise.all([
 131         readFile(CONFIG.LIVE.RTMPS.KEY_FILE),
 132         readFile(CONFIG.LIVE.RTMPS.CERT_FILE)
 133       ])
 134       const serverOptions = { key, cert }
 135
 136       this.rtmpsServer = createServerTLS(serverOptions, socket => {
 137         const session = new NodeRtmpSession(config, socket)
 138
 139         session.inputOriginLocalUrl = 'rtmps://127.0.0.1:' + CONFIG.LIVE.RTMPS.PORT
 140         session.inputOriginPublicUrl = WEBSERVER.RTMPS_URL
 141         session.run()
 142       })
 143
 144       this.rtmpsServer.on('error', err => {
 145         logger.error('Cannot run RTMPS server.', { err, ...lTags() })
 146       })
 147
 148       this.rtmpsServer.listen(CONFIG.LIVE.RTMPS.PORT, CONFIG.LIVE.RTMPS.HOSTNAME)
 149     }
 150   }
 151
 152   stop () {
 153     this.running = false
 154
 155     if (this.rtmpServer) {
 156       logger.info('Stopping RTMP server.', lTags())
 157
 158       this.rtmpServer.close()
 159       this.rtmpServer = undefined
 160     }
 161
 162     if (this.rtmpsServer) {
 163       logger.info('Stopping RTMPS server.', lTags())
 164
 165       this.rtmpsServer.close()
 166       this.rtmpsServer = undefined
 167     }
 168
 169     // Sessions is an object
 170     this.getContext().sessions.forEach((session: any) => {
 171       if (session instanceof NodeRtmpSession) {
 172         session.stop()
 173       }
 174     })
 175   }
 176
 177   isRunning () {
 178     return !!this.rtmpServer
 179   }
 180
 181   stopSessionOf (videoUUID: string, error: LiveVideoError | null) {
 182     const sessionId = this.videoSessions.get(videoUUID)
 183     if (!sessionId) {
 184       logger.debug('No live session to stop for video %s', videoUUID, lTags(sessionId, videoUUID))
 185       return
 186     }
 187
 188     logger.info('Stopping live session of video %s', videoUUID, { error, ...lTags(sessionId, videoUUID) })
 189
 190     this.saveEndingSession(videoUUID, error)
 191       .catch(err => logger.error('Cannot save ending session.', { err, ...lTags(sessionId, videoUUID) }))
 192
 193     this.videoSessions.delete(videoUUID)
 194     this.abortSession(sessionId)
 195   }
 196
 197   private getContext () {
 198     return context
 199   }
 200
 201   private abortSession (sessionId: string) {
 202     const session = this.getContext().sessions.get(sessionId)
 203     if (session) {
 204       session.stop()
 205       this.getContext().sessions.delete(sessionId)
 206     }
 207
 208     const muxingSession = this.muxingSessions.get(sessionId)
 209     if (muxingSession) {
 210       // Muxing session will fire and event so we correctly cleanup the session
 211       muxingSession.abort()
 212
 213       this.muxingSessions.delete(sessionId)
 214     }
 215   }
 216
 217   private async handleSession (options: {
 218     sessionId: string
 219     inputLocalUrl: string
 220     inputPublicUrl: string
 221     streamKey: string
 222   }) {
 223     const { inputLocalUrl, inputPublicUrl, sessionId, streamKey } = options
 224
 225     const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
 226     if (!videoLive) {
 227       logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId))
 228       return this.abortSession(sessionId)
 229     }
 230
 231     const video = videoLive.Video
 232     if (video.isBlacklisted()) {
 233       logger.warn('Video is blacklisted. Refusing stream %s.', streamKey, lTags(sessionId, video.uuid))
 234       return this.abortSession(sessionId)
 235     }
 236
 237     if (this.videoSessions.has(video.uuid)) {
 238       logger.warn('Video %s has already a live session. Refusing stream %s.', video.uuid, streamKey, lTags(sessionId, video.uuid))
 239       return this.abortSession(sessionId)
 240     }
 241
 242     // Cleanup old potential live (could happen with a permanent live)
 243     const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
 244     if (oldStreamingPlaylist) {
 245       if (!videoLive.permanentLive) throw new Error('Found previous session in a non permanent live: ' + video.uuid)
 246
 247       await cleanupAndDestroyPermanentLive(video, oldStreamingPlaylist)
 248     }
 249
 250     this.videoSessions.set(video.uuid, sessionId)
 251
 252     const now = Date.now()
 253     const probe = await ffprobePromise(inputLocalUrl)
 254
 255     const [ { resolution, ratio }, fps, bitrate, hasAudio ] = await Promise.all([
 256       getVideoStreamDimensionsInfo(inputLocalUrl, probe),
 257       getVideoStreamFPS(inputLocalUrl, probe),
 258       getVideoStreamBitrate(inputLocalUrl, probe),
 259       hasAudioStream(inputLocalUrl, probe)
 260     ])
 261
 262     logger.info(
 263       '%s probing took %d ms (bitrate: %d, fps: %d, resolution: %d)',
 264       inputLocalUrl, Date.now() - now, bitrate, fps, resolution, lTags(sessionId, video.uuid)
 265     )
 266
 267     const allResolutions = await Hooks.wrapObject(
 268       this.buildAllResolutionsToTranscode(resolution, hasAudio),
 269       'filter:transcoding.auto.resolutions-to-transcode.result',
 270       { video }
 271     )
 272
 273     logger.info(
 274       'Handling live video of original resolution %d.', resolution,
 275       { allResolutions, ...lTags(sessionId, video.uuid) }
 276     )
 277
 278     return this.runMuxingSession({
 279       sessionId,
 280       videoLive,
 281
 282       inputLocalUrl,
 283       inputPublicUrl,
 284       fps,
 285       bitrate,
 286       ratio,
 287       allResolutions,
 288       hasAudio
 289     })
 290   }
 291
 292   private async runMuxingSession (options: {
 293     sessionId: string
 294     videoLive: MVideoLiveVideoWithSetting
 295
 296     inputLocalUrl: string
 297     inputPublicUrl: string
 298
 299     fps: number
 300     bitrate: number
 301     ratio: number
 302     allResolutions: number[]
 303     hasAudio: boolean
 304   }) {
 305     const { sessionId, videoLive } = options
 306     const videoUUID = videoLive.Video.uuid
 307     const localLTags = lTags(sessionId, videoUUID)
 308
 309     const liveSession = await this.saveStartingSession(videoLive)
 310
 311     const user = await UserModel.loadByLiveId(videoLive.id)
 312     LiveQuotaStore.Instance.addNewLive(user.id, videoLive.id)
 313
 314     const muxingSession = new MuxingSession({
 315       context: this.getContext(),
 316       sessionId,
 317       videoLive,
 318       user,
 319
 320       ...pick(options, [ 'inputLocalUrl', 'inputPublicUrl', 'bitrate', 'ratio', 'fps', 'allResolutions', 'hasAudio' ])
 321     })
 322
 323     muxingSession.on('live-ready', () => this.publishAndFederateLive(videoLive, localLTags))
 324
 325     muxingSession.on('bad-socket-health', ({ videoUUID }) => {
 326       logger.error(
 327         'Too much data in client socket stream (ffmpeg is too slow to transcode the video).' +
 328         ' Stopping session of video %s.', videoUUID,
 329         localLTags
 330       )
 331
 332       this.stopSessionOf(videoUUID, LiveVideoError.BAD_SOCKET_HEALTH)
 333     })
 334
 335     muxingSession.on('duration-exceeded', ({ videoUUID }) => {
 336       logger.info('Stopping session of %s: max duration exceeded.', videoUUID, localLTags)
 337
 338       this.stopSessionOf(videoUUID, LiveVideoError.DURATION_EXCEEDED)
 339     })
 340
 341     muxingSession.on('quota-exceeded', ({ videoUUID }) => {
 342       logger.info('Stopping session of %s: user quota exceeded.', videoUUID, localLTags)
 343
 344       this.stopSessionOf(videoUUID, LiveVideoError.QUOTA_EXCEEDED)
 345     })
 346
 347     muxingSession.on('transcoding-error', ({ videoUUID }) => {
 348       this.stopSessionOf(videoUUID, LiveVideoError.FFMPEG_ERROR)
 349     })
 350
 351     muxingSession.on('transcoding-end', ({ videoUUID }) => {
 352       this.onMuxingFFmpegEnd(videoUUID, sessionId)
 353     })
 354
 355     muxingSession.on('after-cleanup', ({ videoUUID }) => {
 356       this.muxingSessions.delete(sessionId)
 357
 358       LiveQuotaStore.Instance.removeLive(user.id, videoLive.id)
 359
 360       muxingSession.destroy()
 361
 362       return this.onAfterMuxingCleanup({ videoUUID, liveSession })
 363         .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags }))
 364     })
 365
 366     this.muxingSessions.set(sessionId, muxingSession)
 367
 368     muxingSession.runMuxing()
 369       .catch(err => {
 370         logger.error('Cannot run muxing.', { err, ...localLTags })
 371         this.abortSession(sessionId)
 372       })
 373   }
 374
 375   private async publishAndFederateLive (live: MVideoLiveVideo, localLTags: { tags: string[] }) {
 376     const videoId = live.videoId
 377
 378     try {
 379       const video = await VideoModel.loadFull(videoId)
 380
 381       logger.info('Will publish and federate live %s.', video.url, localLTags)
 382
 383       video.state = VideoState.PUBLISHED
 384       video.publishedAt = new Date()
 385       await video.save()
 386
 387       live.Video = video
 388
 389       await wait(getLiveSegmentTime(live.latencyMode) * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION)
 390
 391       try {
 392         await federateVideoIfNeeded(video, false)
 393       } catch (err) {
 394         logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags })
 395       }
 396
 397       PeerTubeSocket.Instance.sendVideoLiveNewState(video)
 398     } catch (err) {
 399       logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags })
 400     }
 401   }
 402
 403   private onMuxingFFmpegEnd (videoUUID: string, sessionId: string) {
 404     // Session already cleaned up
 405     if (!this.videoSessions.has(videoUUID)) return
 406
 407     this.videoSessions.delete(videoUUID)
 408
 409     this.saveEndingSession(videoUUID, null)
 410       .catch(err => logger.error('Cannot save ending session.', { err, ...lTags(sessionId) }))
 411   }
 412
 413   private async onAfterMuxingCleanup (options: {
 414     videoUUID: string
 415     liveSession?: MVideoLiveSession
 416     cleanupNow?: boolean // Default false
 417   }) {
 418     const { videoUUID, liveSession: liveSessionArg, cleanupNow = false } = options
 419
 420     logger.debug('Live of video %s has been cleaned up. Moving to its next state.', videoUUID, lTags(videoUUID))
 421
 422     try {
 423       const fullVideo = await VideoModel.loadFull(videoUUID)
 424       if (!fullVideo) return
 425
 426       const live = await VideoLiveModel.loadByVideoId(fullVideo.id)
 427
 428       const liveSession = liveSessionArg ?? await VideoLiveSessionModel.findLatestSessionOf(fullVideo.id)
 429
 430       // On server restart during a live
 431       if (!liveSession.endDate) {
 432         liveSession.endDate = new Date()
 433         await liveSession.save()
 434       }
 435
 436       JobQueue.Instance.createJobAsync({
 437         type: 'video-live-ending',
 438         payload: {
 439           videoId: fullVideo.id,
 440
 441           replayDirectory: live.saveReplay
 442             ? await this.findReplayDirectory(fullVideo)
 443             : undefined,
 444
 445           liveSessionId: liveSession.id,
 446           streamingPlaylistId: fullVideo.getHLSPlaylist()?.id,
 447
 448           publishedAt: fullVideo.publishedAt.toISOString()
 449         },
 450
 451         delay: cleanupNow
 452           ? 0
 453           : VIDEO_LIVE.CLEANUP_DELAY
 454       })
 455
 456       fullVideo.state = live.permanentLive
 457         ? VideoState.WAITING_FOR_LIVE
 458         : VideoState.LIVE_ENDED
 459
 460       await fullVideo.save()
 461
 462       PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
 463
 464       await federateVideoIfNeeded(fullVideo, false)
 465     } catch (err) {
 466       logger.error('Cannot save/federate new video state of live streaming of video %s.', videoUUID, { err, ...lTags(videoUUID) })
 467     }
 468   }
 469
 470   private async handleBrokenLives () {
 471     await RunnerJobModel.cancelAllJobs({ type: 'live-rtmp-hls-transcoding' })
 472
 473     const videoUUIDs = await VideoModel.listPublishedLiveUUIDs()
 474
 475     for (const uuid of videoUUIDs) {
 476       await this.onAfterMuxingCleanup({ videoUUID: uuid, cleanupNow: true })
 477     }
 478   }
 479
 480   private async findReplayDirectory (video: MVideo) {
 481     const directory = getLiveReplayBaseDirectory(video)
 482     const files = await readdir(directory)
 483
 484     if (files.length === 0) return undefined
 485
 486     return join(directory, files.sort().reverse()[0])
 487   }
 488
 489   private buildAllResolutionsToTranscode (originResolution: number, hasAudio: boolean) {
 490     const includeInput = CONFIG.LIVE.TRANSCODING.ALWAYS_TRANSCODE_ORIGINAL_RESOLUTION
 491
 492     const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED
 493       ? computeResolutionsToTranscode({ input: originResolution, type: 'live', includeInput, strictLower: false, hasAudio })
 494       : []
 495
 496     if (resolutionsEnabled.length === 0) {
 497       return [ originResolution ]
 498     }
 499
 500     return resolutionsEnabled
 501   }
 502
 503   private async saveStartingSession (videoLive: MVideoLiveVideoWithSetting) {
 504     const replaySettings = videoLive.saveReplay
 505       ? new VideoLiveReplaySettingModel({
 506         privacy: videoLive.ReplaySetting.privacy
 507       })
 508       : null
 509
 510     return sequelizeTypescript.transaction(async t => {
 511       if (videoLive.saveReplay) {
 512         await replaySettings.save({ transaction: t })
 513       }
 514
 515       return VideoLiveSessionModel.create({
 516         startDate: new Date(),
 517         liveVideoId: videoLive.videoId,
 518         saveReplay: videoLive.saveReplay,
 519         replaySettingId: videoLive.saveReplay ? replaySettings.id : null,
 520         endingProcessed: false
 521       }, { transaction: t })
 522     })
 523   }
 524
 525   private async saveEndingSession (videoUUID: string, error: LiveVideoError | null) {
 526     const liveSession = await VideoLiveSessionModel.findCurrentSessionOf(videoUUID)
 527     if (!liveSession) return
 528
 529     liveSession.endDate = new Date()
 530     liveSession.error = error
 531
 532     return liveSession.save()
 533   }
 534
 535   static get Instance () {
 536     return this.instance || (this.instance = new this())
 537   }
 538 }
 539
 540 // ---------------------------------------------------------------------------
 541
 542 export {
 543   LiveManager
 544 }