server/lib/live/live-manager.ts

   1
   2 import { createServer, Server } from 'net'
   3 import { isTestInstance } from '@server/helpers/core-utils'
   4 import {
   5   computeResolutionsToTranscode,
   6   ffprobePromise,
   7   getVideoFileBitrate,
   8   getVideoFileFPS,
   9   getVideoFileResolution
  10 } from '@server/helpers/ffprobe-utils'
  11 import { logger, loggerTagsFactory } from '@server/helpers/logger'
  12 import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
  13 import { P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE, VIEW_LIFETIME } from '@server/initializers/constants'
  14 import { UserModel } from '@server/models/user/user'
  15 import { VideoModel } from '@server/models/video/video'
  16 import { VideoLiveModel } from '@server/models/video/video-live'
  17 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
  18 import { MStreamingPlaylistVideo, MVideo, MVideoLiveVideo } from '@server/types/models'
  19 import { VideoState, VideoStreamingPlaylistType } from '@shared/models'
  20 import { federateVideoIfNeeded } from '../activitypub/videos'
  21 import { JobQueue } from '../job-queue'
  22 import { PeerTubeSocket } from '../peertube-socket'
  23 import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename } from '../video-paths'
  24 import { LiveQuotaStore } from './live-quota-store'
  25 import { LiveSegmentShaStore } from './live-segment-sha-store'
  26 import { cleanupLive } from './live-utils'
  27 import { MuxingSession } from './shared'
  28
  29 const NodeRtmpSession = require('node-media-server/src/node_rtmp_session')
  30 const context = require('node-media-server/src/node_core_ctx')
  31 const nodeMediaServerLogger = require('node-media-server/src/node_core_logger')
  32
  33 // Disable node media server logs
  34 nodeMediaServerLogger.setLogType(0)
  35
  36 const config = {
  37   rtmp: {
  38     port: CONFIG.LIVE.RTMP.PORT,
  39     chunk_size: VIDEO_LIVE.RTMP.CHUNK_SIZE,
  40     gop_cache: VIDEO_LIVE.RTMP.GOP_CACHE,
  41     ping: VIDEO_LIVE.RTMP.PING,
  42     ping_timeout: VIDEO_LIVE.RTMP.PING_TIMEOUT
  43   },
  44   transcoding: {
  45     ffmpeg: 'ffmpeg'
  46   }
  47 }
  48
  49 const lTags = loggerTagsFactory('live')
  50
  51 class LiveManager {
  52
  53   private static instance: LiveManager
  54
  55   private readonly muxingSessions = new Map<string, MuxingSession>()
  56   private readonly videoSessions = new Map<number, string>()
  57   // Values are Date().getTime()
  58   private readonly watchersPerVideo = new Map<number, number[]>()
  59
  60   private rtmpServer: Server
  61
  62   private constructor () {
  63   }
  64
  65   init () {
  66     const events = this.getContext().nodeEvent
  67     events.on('postPublish', (sessionId: string, streamPath: string) => {
  68       logger.debug('RTMP received stream', { id: sessionId, streamPath, ...lTags(sessionId) })
  69
  70       const splittedPath = streamPath.split('/')
  71       if (splittedPath.length !== 3 || splittedPath[1] !== VIDEO_LIVE.RTMP.BASE_PATH) {
  72         logger.warn('Live path is incorrect.', { streamPath, ...lTags(sessionId) })
  73         return this.abortSession(sessionId)
  74       }
  75
  76       this.handleSession(sessionId, streamPath, splittedPath[2])
  77         .catch(err => logger.error('Cannot handle sessions.', { err, ...lTags(sessionId) }))
  78     })
  79
  80     events.on('donePublish', sessionId => {
  81       logger.info('Live session ended.', { sessionId, ...lTags(sessionId) })
  82     })
  83
  84     registerConfigChangedHandler(() => {
  85       if (!this.rtmpServer && CONFIG.LIVE.ENABLED === true) {
  86         this.run()
  87         return
  88       }
  89
  90       if (this.rtmpServer && CONFIG.LIVE.ENABLED === false) {
  91         this.stop()
  92       }
  93     })
  94
  95     // Cleanup broken lives, that were terminated by a server restart for example
  96     this.handleBrokenLives()
  97       .catch(err => logger.error('Cannot handle broken lives.', { err, ...lTags() }))
  98
  99     setInterval(() => this.updateLiveViews(), VIEW_LIFETIME.LIVE)
 100   }
 101
 102   run () {
 103     logger.info('Running RTMP server on port %d', config.rtmp.port, lTags())
 104
 105     this.rtmpServer = createServer(socket => {
 106       const session = new NodeRtmpSession(config, socket)
 107
 108       session.run()
 109     })
 110
 111     this.rtmpServer.on('error', err => {
 112       logger.error('Cannot run RTMP server.', { err, ...lTags() })
 113     })
 114
 115     this.rtmpServer.listen(CONFIG.LIVE.RTMP.PORT)
 116   }
 117
 118   stop () {
 119     logger.info('Stopping RTMP server.', lTags())
 120
 121     this.rtmpServer.close()
 122     this.rtmpServer = undefined
 123
 124     // Sessions is an object
 125     this.getContext().sessions.forEach((session: any) => {
 126       if (session instanceof NodeRtmpSession) {
 127         session.stop()
 128       }
 129     })
 130   }
 131
 132   isRunning () {
 133     return !!this.rtmpServer
 134   }
 135
 136   stopSessionOf (videoId: number) {
 137     const sessionId = this.videoSessions.get(videoId)
 138     if (!sessionId) return
 139
 140     this.videoSessions.delete(videoId)
 141     this.abortSession(sessionId)
 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   private getContext () {
 158     return context
 159   }
 160
 161   private abortSession (sessionId: string) {
 162     const session = this.getContext().sessions.get(sessionId)
 163     if (session) {
 164       session.stop()
 165       this.getContext().sessions.delete(sessionId)
 166     }
 167
 168     const muxingSession = this.muxingSessions.get(sessionId)
 169     if (muxingSession) {
 170       // Muxing session will fire and event so we correctly cleanup the session
 171       muxingSession.abort()
 172
 173       this.muxingSessions.delete(sessionId)
 174     }
 175   }
 176
 177   private async handleSession (sessionId: string, streamPath: string, streamKey: string) {
 178     const videoLive = await VideoLiveModel.loadByStreamKey(streamKey)
 179     if (!videoLive) {
 180       logger.warn('Unknown live video with stream key %s.', streamKey, lTags(sessionId))
 181       return this.abortSession(sessionId)
 182     }
 183
 184     const video = videoLive.Video
 185     if (video.isBlacklisted()) {
 186       logger.warn('Video is blacklisted. Refusing stream %s.', streamKey, lTags(sessionId, video.uuid))
 187       return this.abortSession(sessionId)
 188     }
 189
 190     // Cleanup old potential live files (could happen with a permanent live)
 191     LiveSegmentShaStore.Instance.cleanupShaSegments(video.uuid)
 192
 193     const oldStreamingPlaylist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
 194     if (oldStreamingPlaylist) {
 195       await cleanupLive(video, oldStreamingPlaylist)
 196     }
 197
 198     this.videoSessions.set(video.id, sessionId)
 199
 200     const rtmpUrl = 'rtmp://127.0.0.1:' + config.rtmp.port + streamPath
 201
 202     const now = Date.now()
 203     const probe = await ffprobePromise(rtmpUrl)
 204
 205     const [ { resolution, ratio }, fps, bitrate ] = await Promise.all([
 206       getVideoFileResolution(rtmpUrl, probe),
 207       getVideoFileFPS(rtmpUrl, probe),
 208       getVideoFileBitrate(rtmpUrl, probe)
 209     ])
 210
 211     logger.info(
 212       '%s probing took %d ms (bitrate: %d, fps: %d, resolution: %d)',
 213       rtmpUrl, Date.now() - now, bitrate, fps, resolution, lTags(sessionId, video.uuid)
 214     )
 215
 216     const allResolutions = this.buildAllResolutionsToTranscode(resolution)
 217
 218     logger.info(
 219       'Will mux/transcode live video of original resolution %d.', resolution,
 220       { allResolutions, ...lTags(sessionId, video.uuid) }
 221     )
 222
 223     const streamingPlaylist = await this.createLivePlaylist(video, allResolutions)
 224
 225     return this.runMuxingSession({
 226       sessionId,
 227       videoLive,
 228       streamingPlaylist,
 229       rtmpUrl,
 230       fps,
 231       bitrate,
 232       ratio,
 233       allResolutions
 234     })
 235   }
 236
 237   private async runMuxingSession (options: {
 238     sessionId: string
 239     videoLive: MVideoLiveVideo
 240     streamingPlaylist: MStreamingPlaylistVideo
 241     rtmpUrl: string
 242     fps: number
 243     bitrate: number
 244     ratio: number
 245     allResolutions: number[]
 246   }) {
 247     const { sessionId, videoLive, streamingPlaylist, allResolutions, fps, bitrate, ratio, rtmpUrl } = options
 248     const videoUUID = videoLive.Video.uuid
 249     const localLTags = lTags(sessionId, videoUUID)
 250
 251     const user = await UserModel.loadByLiveId(videoLive.id)
 252     LiveQuotaStore.Instance.addNewLive(user.id, videoLive.id)
 253
 254     const muxingSession = new MuxingSession({
 255       context: this.getContext(),
 256       user,
 257       sessionId,
 258       videoLive,
 259       streamingPlaylist,
 260       rtmpUrl,
 261       bitrate,
 262       ratio,
 263       fps,
 264       allResolutions
 265     })
 266
 267     muxingSession.on('master-playlist-created', () => this.publishAndFederateLive(videoLive, localLTags))
 268
 269     muxingSession.on('bad-socket-health', ({ videoId }) => {
 270       logger.error(
 271         'Too much data in client socket stream (ffmpeg is too slow to transcode the video).' +
 272         ' Stopping session of video %s.', videoUUID,
 273         localLTags
 274       )
 275
 276       this.stopSessionOf(videoId)
 277     })
 278
 279     muxingSession.on('duration-exceeded', ({ videoId }) => {
 280       logger.info('Stopping session of %s: max duration exceeded.', videoUUID, localLTags)
 281
 282       this.stopSessionOf(videoId)
 283     })
 284
 285     muxingSession.on('quota-exceeded', ({ videoId }) => {
 286       logger.info('Stopping session of %s: user quota exceeded.', videoUUID, localLTags)
 287
 288       this.stopSessionOf(videoId)
 289     })
 290
 291     muxingSession.on('ffmpeg-error', ({ sessionId }) => this.abortSession(sessionId))
 292     muxingSession.on('ffmpeg-end', ({ videoId }) => {
 293       this.onMuxingFFmpegEnd(videoId)
 294     })
 295
 296     muxingSession.on('after-cleanup', ({ videoId }) => {
 297       this.muxingSessions.delete(sessionId)
 298
 299       muxingSession.destroy()
 300
 301       return this.onAfterMuxingCleanup(videoId)
 302         .catch(err => logger.error('Error in end transmuxing.', { err, ...localLTags }))
 303     })
 304
 305     this.muxingSessions.set(sessionId, muxingSession)
 306
 307     muxingSession.runMuxing()
 308       .catch(err => {
 309         logger.error('Cannot run muxing.', { err, ...localLTags })
 310         this.abortSession(sessionId)
 311       })
 312   }
 313
 314   private async publishAndFederateLive (live: MVideoLiveVideo, localLTags: { tags: string[] }) {
 315     const videoId = live.videoId
 316
 317     try {
 318       const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
 319
 320       logger.info('Will publish and federate live %s.', video.url, localLTags)
 321
 322       video.state = VideoState.PUBLISHED
 323       await video.save()
 324
 325       live.Video = video
 326
 327       setTimeout(() => {
 328         federateVideoIfNeeded(video, false)
 329           .catch(err => logger.error('Cannot federate live video %s.', video.url, { err, ...localLTags }))
 330
 331         PeerTubeSocket.Instance.sendVideoLiveNewState(video)
 332       }, VIDEO_LIVE.SEGMENT_TIME_SECONDS * 1000 * VIDEO_LIVE.EDGE_LIVE_DELAY_SEGMENTS_NOTIFICATION)
 333     } catch (err) {
 334       logger.error('Cannot save/federate live video %d.', videoId, { err, ...localLTags })
 335     }
 336   }
 337
 338   private onMuxingFFmpegEnd (videoId: number) {
 339     this.watchersPerVideo.delete(videoId)
 340     this.videoSessions.delete(videoId)
 341   }
 342
 343   private async onAfterMuxingCleanup (videoUUID: string, cleanupNow = false) {
 344     try {
 345       const fullVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoUUID)
 346       if (!fullVideo) return
 347
 348       const live = await VideoLiveModel.loadByVideoId(fullVideo.id)
 349
 350       if (!live.permanentLive) {
 351         JobQueue.Instance.createJob({
 352           type: 'video-live-ending',
 353           payload: {
 354             videoId: fullVideo.id
 355           }
 356         }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY })
 357
 358         fullVideo.state = VideoState.LIVE_ENDED
 359       } else {
 360         fullVideo.state = VideoState.WAITING_FOR_LIVE
 361       }
 362
 363       await fullVideo.save()
 364
 365       PeerTubeSocket.Instance.sendVideoLiveNewState(fullVideo)
 366
 367       await federateVideoIfNeeded(fullVideo, false)
 368     } catch (err) {
 369       logger.error('Cannot save/federate new video state of live streaming of video %d.', videoUUID, { err, ...lTags(videoUUID) })
 370     }
 371   }
 372
 373   private async updateLiveViews () {
 374     if (!this.isRunning()) return
 375
 376     if (!isTestInstance()) logger.info('Updating live video views.', lTags())
 377
 378     for (const videoId of this.watchersPerVideo.keys()) {
 379       const notBefore = new Date().getTime() - VIEW_LIFETIME.LIVE
 380
 381       const watchers = this.watchersPerVideo.get(videoId)
 382
 383       const numWatchers = watchers.length
 384
 385       const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
 386       video.views = numWatchers
 387       await video.save()
 388
 389       await federateVideoIfNeeded(video, false)
 390
 391       PeerTubeSocket.Instance.sendVideoViewsUpdate(video)
 392
 393       // Only keep not expired watchers
 394       const newWatchers = watchers.filter(w => w > notBefore)
 395       this.watchersPerVideo.set(videoId, newWatchers)
 396
 397       logger.debug('New live video views for %s is %d.', video.url, numWatchers, lTags())
 398     }
 399   }
 400
 401   private async handleBrokenLives () {
 402     const videoUUIDs = await VideoModel.listPublishedLiveUUIDs()
 403
 404     for (const uuid of videoUUIDs) {
 405       await this.onAfterMuxingCleanup(uuid, true)
 406     }
 407   }
 408
 409   private buildAllResolutionsToTranscode (originResolution: number) {
 410     const resolutionsEnabled = CONFIG.LIVE.TRANSCODING.ENABLED
 411       ? computeResolutionsToTranscode(originResolution, 'live')
 412       : []
 413
 414     return resolutionsEnabled.concat([ originResolution ])
 415   }
 416
 417   private async createLivePlaylist (video: MVideo, allResolutions: number[]): Promise<MStreamingPlaylistVideo> {
 418     const playlist = await VideoStreamingPlaylistModel.loadOrGenerate(video)
 419
 420     playlist.playlistFilename = generateHLSMasterPlaylistFilename(true)
 421     playlist.segmentsSha256Filename = generateHlsSha256SegmentsFilename(true)
 422
 423     playlist.p2pMediaLoaderPeerVersion = P2P_MEDIA_LOADER_PEER_VERSION
 424     playlist.type = VideoStreamingPlaylistType.HLS
 425
 426     playlist.assignP2PMediaLoaderInfoHashes(video, allResolutions)
 427
 428     return playlist.save()
 429   }
 430
 431   static get Instance () {
 432     return this.instance || (this.instance = new this())
 433   }
 434 }
 435
 436 // ---------------------------------------------------------------------------
 437
 438 export {
 439   LiveManager
 440 }