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   hasSession (sessionId: string) {
 182     return this.getContext().sessions.has(sessionId)
 183   }
 184
 185   stopSessionOf (videoUUID: string, error: LiveVideoError | null) {
 186     const sessionId = this.videoSessions.get(videoUUID)
 187     if (!sessionId) {
 188       logger.debug('No live session to stop for video %s', videoUUID, lTags(sessionId, videoUUID))
 189       return
 190     }
 191
 192     logger.info('Stopping live session of video %s', videoUUID, { error, ...lTags(sessionId, videoUUID) })
 193
 194     this.saveEndingSession(videoUUID, error)
 195       .catch(err => logger.error('Cannot save ending session.', { err, ...lTags(sessionId, videoUUID) }))
 196
 197     this.videoSessions.delete(videoUUID)
 198     this.abortSession(sessionId)
 199   }
 200
 201   private getContext () {
 202     return context
 203   }
 204
 205   private abortSession (sessionId: string) {
 206     const session = this.getContext().sessions.get(sessionId)
 207     if (session) {
 208       session.stop()
 209       this.getContext().sessions.delete(sessionId)
 210     }
 211
 212     const muxingSession = this.muxingSessions.get(sessionId)
 213     if (muxingSession) {
 214       // Muxing session will fire and event so we correctly cleanup the session
 215       muxingSession.abort()
 216
 217       this.muxingSessions.delete(sessionId)
 218     }
 219   }
 220
 221   private async handleSession (options: {
 222     sessionId: string
 223     inputLocalUrl: string
 224     inputPublicUrl: string
 225     streamKey: string
 226   }) {
 227     const { inputLocalUrl, inputPublicUrl, sessionId, streamKey } = options
 228
 229     const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
 230     if (!videoLive) {
 231       logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId))
 232       return this.abortSession(sessionId)
 233     }
 234
 235     const video = videoLive.Video
 236     if (video.isBlacklisted()) {
 237       logger.warn('Video is blacklisted. Refusing stream %s.', streamKey, lTags(sessionId, video.uuid))
 238       return this.abortSession(sessionId)
 239     }
 240
 241     if (this.videoSessions.has(video.uuid)) {
 242       logger.warn('Video %s has already a live session. Refusing stream %s.', video.uuid, streamKey, lTags(sessionId, video.uuid))
 243       return this.abortSession(sessionId)
 244     }
 245
 246     // Cleanup old potential live (could happen with a permanent live)
 247     const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
 248     if (oldStreamingPlaylist) {
 249       if (!videoLive.permanentLive) throw new Error('Found previous session in a non permanent live: ' + video.uuid)
 250
 251       await cleanupAndDestroyPermanentLive(video, oldStreamingPlaylist)
 252     }
 253
 254     this.videoSessions.set(video.uuid, sessionId)
 255
 256     const now = Date.now()
 257     const probe = await ffprobePromise(inputLocalUrl)
 258
 259     const [ { resolution, ratio }, fps, bitrate, hasAudio ] = await Promise.all([
 260       getVideoStreamDimensionsInfo(inputLocalUrl, probe),
 261       getVideoStreamFPS(inputLocalUrl, probe),
 262       getVideoStreamBitrate(inputLocalUrl, probe),
 263       hasAudioStream(inputLocalUrl, probe)
 264     ])
 265
 266     logger.info(
 267       '%s probing took %d ms (bitrate: %d, fps: %d, resolution: %d)',
 268       inputLocalUrl, Date.now() - now, bitrate, fps, resolution, lTags(sessionId, video.uuid)
 269     )
 270
 271     const allResolutions = await Hooks.wrapObject(
 272       this.buildAllResolutionsToTranscode(resolution, hasAudio),
 273       'filter:transcoding.auto.resolutions-to-transcode.result',
 274       { video }
 275     )
 276
 277     logger.info(
 278       'Handling live video of original resolution %d.', resolution,
 279       { allResolutions, ...lTags(sessionId, video.uuid) }
 280     )
 281
 282     return this.runMuxingSession({
 283       sessionId,
 284       videoLive,
 285
 286       inputLocalUrl,
 287       inputPublicUrl,
 288       fps,
 289       bitrate,
 290       ratio,
 291       allResolutions,
 292       hasAudio
 293     })
 294   }
 295
 296   private async runMuxingSession (options: {
 297     sessionId: string
 298     videoLive: MVideoLiveVideoWithSetting
 299
 300     inputLocalUrl: string
 301     inputPublicUrl: string
 302
 303     fps: number
 304     bitrate: number
 305     ratio: number
 306     allResolutions: number[]
 307     hasAudio: boolean
 308   }) {
 309     const { sessionId, videoLive } = options
 310     const videoUUID = videoLive.Video.uuid
 311     const localLTags = lTags(sessionId, videoUUID)
 312
 313     const liveSession = await this.saveStartingSession(videoLive)
 314
 315     const user = await UserModel.loadByLiveId(videoLive.id)
 316     LiveQuotaStore.Instance.addNewLive(user.id, sessionId)
 317
 318     const muxingSession = new MuxingSession({
 319       context: this.getContext(),
 320       sessionId,
 321       videoLive,
 322       user,
 323
 324       ...pick(options, [ 'inputLocalUrl', 'inputPublicUrl', 'bitrate', 'ratio', 'fps', 'allResolutions', 'hasAudio' ])
 325     })
 326
 327     muxingSession.on('live-ready', () => this.publishAndFederateLive(videoLive, localLTags))
 328
 329     muxingSession.on('bad-socket-health', ({ videoUUID }) => {
 330       logger.error(
 331         'Too much data in client socket stream (ffmpeg is too slow to transcode the video).' +
 332         ' Stopping session of video %s.', videoUUID,
 333         localLTags
 334       )
 335
 336       this.stopSessionOf(videoUUID, LiveVideoError.BAD_SOCKET_HEALTH)
 337     })
 338
 339     muxingSession.on('duration-exceeded', ({ videoUUID }) => {
 340       logger.info('Stopping session of %s: max duration exceeded.', videoUUID, localLTags)
 341
 342       this.stopSessionOf(videoUUID, LiveVideoError.DURATION_EXCEEDED)
 343     })
 344
 345     muxingSession.on('quota-exceeded', ({ videoUUID }) => {
 346       logger.info('Stopping session of %s: user quota exceeded.', videoUUID, localLTags)
 347
 348       this.stopSessionOf(videoUUID, LiveVideoError.QUOTA_EXCEEDED)
 349     })
 350
 351     muxingSession.on('transcoding-error', ({ videoUUID }) => {
 352       this.stopSessionOf(videoUUID, LiveVideoError.FFMPEG_ERROR)
 353     })
 354
 355     muxingSession.on('transcoding-end', ({ videoUUID }) => {
 356       this.onMuxingFFmpegEnd(videoUUID, sessionId)
 357     })
 358
 359     muxingSession.on('after-cleanup', ({ videoUUID }) => {
 360       this.muxingSessions.delete(sessionId)
 361
 362       LiveQuotaStore.Instance.removeLive(user.id, sessionId)
 363
 364       muxingSession.destroy()
 365
 366       return this.onAfterMuxingCleanup({ videoUUID, liveSession })
 367         .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags }))
 368     })
 369
 370     this.muxingSessions.set(sessionId, muxingSession)
 371
 372     muxingSession.runMuxing()
 373       .catch(err => {
 374         logger.error('Cannot run muxing.', { err, ...localLTags })
 375         this.abortSession(sessionId)
 376       })
 377   }
 378
 379   private async publishAndFederateLive (live: MVideoLiveVideo, localLTags: { tags: string[] }) {
 380     const videoId = live.videoId
 381
 382     try {
 383       const video = await VideoModel.loadFull(videoId)
 384
 385       logger.info('Will publish and federate live %s.', video.url, localLTags)
 386
 387       video.state = VideoState.PUBLISHED
 388       video.publishedAt = new Date()
 389       await video.save()
 390
 391       live.Video = video
 392
 393       await wait(getLiveSegmentTime(live.latencyMode) * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION)
 394
 395       try {
 396         await federateVideoIfNeeded(video, false)
 397       } catch (err) {
 398         logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags })
 399       }
 400
 401       PeerTubeSocket.Instance.sendVideoLiveNewState(video)
 402
 403       Hooks.runAction('action:live.video.state.updated', { video })
 404     } catch (err) {
 405       logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags })
 406     }
 407   }
 408
 409   private onMuxingFFmpegEnd (videoUUID: string, sessionId: string) {
 410     // Session already cleaned up
 411     if (!this.videoSessions.has(videoUUID)) return
 412
 413     this.videoSessions.delete(videoUUID)
 414
 415     this.saveEndingSession(videoUUID, null)
 416       .catch(err => logger.error('Cannot save ending session.', { err, ...lTags(sessionId) }))
 417   }
 418
 419   private async onAfterMuxingCleanup (options: {
 420     videoUUID: string
 421     liveSession?: MVideoLiveSession
 422     cleanupNow?: boolean // Default false
 423   }) {
 424     const { videoUUID, liveSession: liveSessionArg, cleanupNow = false } = options
 425
 426     logger.debug('Live of video %s has been cleaned up. Moving to its next state.', videoUUID, lTags(videoUUID))
 427
 428     try {
 429       const fullVideo = await VideoModel.loadFull(videoUUID)
 430       if (!fullVideo) return
 431
 432       const live = await VideoLiveModel.loadByVideoId(fullVideo.id)
 433
 434       const liveSession = liveSessionArg ?? await VideoLiveSessionModel.findLatestSessionOf(fullVideo.id)
 435
 436       // On server restart during a live
 437       if (!liveSession.endDate) {
 438         liveSession.endDate = new Date()
 439         await liveSession.save()
 440       }
 441
 442       JobQueue.Instance.createJobAsync({
 443         type: 'video-live-ending',
 444         payload: {
 445           videoId: fullVideo.id,
 446
 447           replayDirectory: live.saveReplay
 448             ? await this.findReplayDirectory(fullVideo)
 449             : undefined,
 450
 451           liveSessionId: liveSession.id,
 452           streamingPlaylistId: fullVideo.getHLSPlaylist()?.id,
 453
 454           publishedAt: fullVideo.publishedAt.toISOString()
 455         },
 456
 457         delay: cleanupNow
 458           ? 0
 459           : VIDEO_LIVE.CLEANUP_DELAY
 460       })
 461
 462       fullVideo.state = live.permanentLive
 463         ? VideoState.WAITING_FOR_LIVE
 464         : VideoState.LIVE_ENDED
 465
 466       await fullVideo.save()
 467
 468       PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
 469
 470       await federateVideoIfNeeded(fullVideo, false)
 471
 472       Hooks.runAction('action:live.video.state.updated', { video: fullVideo })
 473     } catch (err) {
 474       logger.error('Cannot save/federate new video state of live streaming of video %s.', videoUUID, { err, ...lTags(videoUUID) })
 475     }
 476   }
 477
 478   private async handleBrokenLives () {
 479     await RunnerJobModel.cancelAllJobs({ type: 'live-rtmp-hls-transcoding' })
 480
 481     const videoUUIDs = await VideoModel.listPublishedLiveUUIDs()
 482
 483     for (const uuid of videoUUIDs) {
 484       await this.onAfterMuxingCleanup({ videoUUID: uuid, cleanupNow: true })
 485     }
 486   }
 487
 488   private async findReplayDirectory (video: MVideo) {
 489     const directory = getLiveReplayBaseDirectory(video)
 490     const files = await readdir(directory)
 491
 492     if (files.length === 0) return undefined
 493
 494     return join(directory, files.sort().reverse()[0])
 495   }
 496
 497   private buildAllResolutionsToTranscode (originResolution: number, hasAudio: boolean) {
 498     const includeInput = CONFIG.LIVE.TRANSCODING.ALWAYS_TRANSCODE_ORIGINAL_RESOLUTION
 499
 500     const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED
 501       ? computeResolutionsToTranscode({ input: originResolution, type: 'live', includeInput, strictLower: false, hasAudio })
 502       : []
 503
 504     if (resolutionsEnabled.length === 0) {
 505       return [ originResolution ]
 506     }
 507
 508     return resolutionsEnabled
 509   }
 510
 511   private async saveStartingSession (videoLive: MVideoLiveVideoWithSetting) {
 512     const replaySettings = videoLive.saveReplay
 513       ? new VideoLiveReplaySettingModel({
 514         privacy: videoLive.ReplaySetting.privacy
 515       })
 516       : null
 517
 518     return sequelizeTypescript.transaction(async t => {
 519       if (videoLive.saveReplay) {
 520         await replaySettings.save({ transaction: t })
 521       }
 522
 523       return VideoLiveSessionModel.create({
 524         startDate: new Date(),
 525         liveVideoId: videoLive.videoId,
 526         saveReplay: videoLive.saveReplay,
 527         replaySettingId: videoLive.saveReplay ? replaySettings.id : null,
 528         endingProcessed: false
 529       }, { transaction: t })
 530     })
 531   }
 532
 533   private async saveEndingSession (videoUUID: string, error: LiveVideoError | null) {
 534     const liveSession = await VideoLiveSessionModel.findCurrentSessionOf(videoUUID)
 535     if (!liveSession) return
 536
 537     liveSession.endDate = new Date()
 538     liveSession.error = error
 539
 540     return liveSession.save()
 541   }
 542
 543   static get Instance () {
 544     return this.instance || (this.instance = new this())
 545   }
 546 }
 547
 548 // ---------------------------------------------------------------------------
 549
 550 export {
 551   LiveManager
 552 }