Add region based schedule for video/audio/streaming agent (#612)

qwu16 · web-flow · commit 406950d420cc · 2020-09-07T11:14:33.000+08:00
* Add region based schedule for video/audio/streaming agent
diff --git a/source/agent/audio/agent.toml b/source/agent/audio/agent.toml
@@ -23,6 +23,14 @@ max_load = 0.85 #default: 0.85
 host = "localhost" #default: "localhost"
 port = 5672 #default: 5672
 
+[capacity]
+#The ISP list this agent is able to handle.
+#If the ISP list is set to be non-empty, only the creating token requests with preference.ips being matched with one element of this list will be scheduled to this agent.
+isps = [] #default: [], which means all ISPs.
+#The region list this agent prefers to handle
+#If the region list is set to be non-empty, the creating token requests with preference.region being matched with one element of this list will be priorly scheduled to this agent.
+regions = [] #default: [], which means all regions.
+
 [internal]
 #The IP address used for internal-cluster media spreading. Will use the IP got from the 'network_interface' item if 'ip_address' is not specified or equal to "".
 ip_address = "" #default: ""
diff --git a/source/agent/conference/conference.js b/source/agent/conference/conference.js
@@ -340,7 +340,7 @@ var Conference = function (rpcClient, selfRpcId) {
     }
   };
 
-  var initRoom = function(roomId) {
+  var initRoom = function(roomId, origin) {
     if (is_initializing) {
       return new Promise(function(resolve, reject) {
         var interval = setInterval(function() {
@@ -376,6 +376,7 @@ var Conference = function (rpcClient, selfRpcId) {
                   rpcClient: rpcClient,
                   room: roomId,
                   config: room_config,
+                  origin: origin,
                   selfRpcId: selfRpcId
                 },
                 function onOk(rmController) {
@@ -396,6 +397,7 @@ var Conference = function (rpcClient, selfRpcId) {
                     var mixed_stream_info = Stream.createMixStream(room_id, viewSettings, room_config.mediaOut, av_capability);
 
                     streams[mixed_stream_id] = mixed_stream_info;
+                    streams[mixed_stream_id].info.origin = origin;
                     log.debug('Mixed stream info:', mixed_stream_info);
                     room_config.notifying.streamChange &&
                       sendMsg('room', 'all', 'stream', {id: mixed_stream_id, status: 'add', data: Stream.toPortalFormat(mixed_stream_info)});
@@ -406,7 +408,7 @@ var Conference = function (rpcClient, selfRpcId) {
                                                        user: 'admin',
                                                        role: 'admin',
                                                        portal: undefined,
-                                                       origin: {isp: 'isp', region: 'region'},//FIXME: hard coded.
+                                                       origin: origin,
                                                        permission: {
                                                         subscribe: {audio: true, video: true},
                                                         publish: {audio: true, video: true}
@@ -598,9 +600,10 @@ var Conference = function (rpcClient, selfRpcId) {
     }
 
     var isReadded = !!(streams[id] && !streams[id].isInConnecting);
-
+    var origin = streams[id].info.origin;
+    info.origin = origin;
     return new Promise((resolve, reject) => {
-      roomController && roomController.publish(info.owner, id, locality, media, info.type, function() {
+      roomController && roomController.publish(info.owner, id, locality, media, info.type, origin, function() {
         if (participants[info.owner]) {
           var st = Stream.createForwardStream(id, media, info, room_config);
           st.info.inViews = [];
@@ -733,6 +736,7 @@ var Conference = function (rpcClient, selfRpcId) {
           };
         }
       }
+      media.origin = streams[media.video.from].info.origin;
     }
 
     if (media.audio) {
@@ -745,6 +749,7 @@ var Conference = function (rpcClient, selfRpcId) {
       media.audio.format = (media.audio.format || source.format);
       mediaSpec.audio.format = media.audio.format;
       mediaSpec.audio.status = (media.audio.status || 'active');
+      media.origin = streams[media.audio.from].info.origin;
     }
 
     const isAudioPubPermitted = !!participants[info.owner].isPublishPermitted('audio'); 
@@ -851,7 +856,7 @@ var Conference = function (rpcClient, selfRpcId) {
   that.join = function(roomId, participantInfo, callback) {
     log.debug('participant:', participantInfo, 'join room:', roomId);
     var permission;
-    return initRoom(roomId)
+    return initRoom(roomId, participantInfo.origin)
       .then(function() {
         log.debug('room_config.participantLimit:', room_config.participantLimit, 'current participants count:', Object.keys(participants).length);
         if (room_config.participantLimit > 0 && (Object.keys(participants).length >= room_config.participantLimit + 1)) {
@@ -1006,7 +1011,8 @@ var Conference = function (rpcClient, selfRpcId) {
         }
       }
 
-      initiateStream(streamId, {owner: participantId, type: pubInfo.type});
+      var origin = participants[participantId].getOrigin();
+      initiateStream(streamId, {owner: participantId, type: pubInfo.type, origin: origin});
       return accessController.initiate(participantId, streamId, 'in', participants[participantId].getOrigin(), pubInfo, format_preference)
       .then((result) => {
         callback('callback', result);
@@ -1796,8 +1802,12 @@ var Conference = function (rpcClient, selfRpcId) {
     log.debug('addStreamingIn, roomId:', roomId, 'pubInfo:', JSON.stringify(pubInfo));
 
     if (pubInfo.type === 'streaming') {
+      var origin = { isp:'isp', region:'region'};
+      if(pubInfo.connection.origin != null) {
+          origin = pubInfo.connection.origin;
+      }
       var stream_id = Math.round(Math.random() * 1000000000000000000) + '';
-      return initRoom(roomId)
+      return initRoom(roomId, origin)
       .then(() => {
         if (room_config.inputLimit >= 0 && (room_config.inputLimit <= currentInputCount())) {
           return Promise.reject('Too many inputs');
@@ -1811,8 +1821,8 @@ var Conference = function (rpcClient, selfRpcId) {
           return Promise.reject('Video is forbiden');
         }
 
-        initiateStream(stream_id, {owner: 'admin', type: pubInfo.type});
-        return accessController.initiate('admin', stream_id, 'in', participants['admin'].getOrigin(), pubInfo);
+        initiateStream(stream_id, {owner: 'admin', type: pubInfo.type, origin: origin});
+        return accessController.initiate('admin', stream_id, 'in', origin, pubInfo);
       }).then((result) => {
         return 'ok';
       }).then(() => {
@@ -2082,7 +2092,8 @@ var Conference = function (rpcClient, selfRpcId) {
 
     if (subDesc.type === 'streaming' || subDesc.type === 'recording' || subDesc.type === 'analytics') {
       var subscription_id = Math.round(Math.random() * 1000000000000000000) + '';
-      return initRoom(roomId)
+      var origin = {isp: 'isp', region: 'region'};
+      return initRoom(roomId, origin)
       .then(() => {
         if (subDesc.media.audio && !room_config.mediaOut.audio.length) {
           return Promise.reject('Audio is forbiden');
@@ -2180,7 +2191,16 @@ var Conference = function (rpcClient, selfRpcId) {
         }
 
         // Schedule preference for worker node
-        var accessPreference = Object.assign({}, participants['admin'].getOrigin());
+        var streamFrom = undefined;
+        if(subDesc.media.video && subDesc.media.video.from) {
+          streamFrom = subDesc.media.video.from;
+        } else if(subDesc.media.audio && subDesc.media.audio.from) {
+          streamFrom = subDesc.media.audio.from;
+        } else {
+          return Promise.reject('No video or audio source to process');
+        }
+
+        var accessPreference = Object.assign({}, streams[streamFrom].info.origin);
         if (subDesc.type === 'analytics') {
           // Schedule analytics agent according to the algorithm
           accessPreference.algorithm = subDesc.connection.algorithm;
diff --git a/source/agent/conference/roomController.js b/source/agent/conference/roomController.js
@@ -57,6 +57,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         rpcClient = spec.rpcClient,
         config = spec.config,
         room_id = spec.room,
+	origin = spec.origin,
         selfRpcId = spec.selfRpcId,
         enable_audio_transcoding = config.transcoding && !!config.transcoding.audio,
         enable_video_transcoding = config.transcoding && !!config.transcoding.video,
@@ -214,7 +215,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         // Media mixer initializer
         var mixStreamId = getMixStreamOfView(view);
         var initMixer = (mixerId, type, mixConfig) => new Promise(function(resolve, reject) {
-            newTerminal(mixerId, type, mixStreamId, false,
+            newTerminal(mixerId, type, mixStreamId, false, origin,
                 function onTerminalReady() {
                     log.debug('new terminal ok. terminal_id', mixerId, 'type:', type, 'view:', view, 'mixConfig:', mixConfig);
                     initMediaProcessor(mixerId, ['mixing', mixConfig, room_id, selfRpcId, view])
@@ -350,19 +351,20 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         streams = {};
     };
 
-    var newTerminal = function (terminal_id, terminal_type, owner, preAssignedNode, on_ok, on_error) {
-        log.debug('newTerminal:', terminal_id, 'terminal_type:', terminal_type, 'owner:', owner);
+    var newTerminal = function (terminal_id, terminal_type, owner, preAssignedNode, origin, on_ok, on_error) {
+        log.debug('newTerminal:', terminal_id, 'terminal_type:', terminal_type, 'owner:', owner, " origin:", origin);
         if (terminals[terminal_id] === undefined) {
                 var purpose = (terminal_type === 'vmixer' || terminal_type === 'vxcoder') ? 'video'
                               : ((terminal_type === 'amixer' || terminal_type === 'axcoder') ? 'audio' : 'unknown');
-
+                mediaPreference.origin = origin;
                 var nodeLocality = (preAssignedNode ? Promise.resolve(preAssignedNode)
                                                : rpcReq.getWorkerNode(cluster, purpose, {room: room_id, task: terminal_id}, mediaPreference));
 
                 return nodeLocality
                     .then(function(locality) {
                         terminals[terminal_id] = {
                             owner: owner,
+                            origin: origin,
                             type: terminal_type,
                             locality: locality,
                             published: [],
@@ -903,7 +905,8 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
     var getAudioTranscoder = function (stream_id, on_ok, on_error) {
         findExistingAudioTranscoder(stream_id, on_ok, function () {
             var axcoder = randomId();
-            newTerminal(axcoder, 'axcoder', streams[stream_id].owner, false, function () {
+            var terminalId = streams[stream_id].owner;
+            newTerminal(axcoder, 'axcoder', streams[stream_id].owner, false, terminals[terminalId].origin, function () {
                 var on_failed = function (reason) {
                     makeRPC(
                         rpcClient,
@@ -1016,7 +1019,8 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
     var getVideoTranscoder = function (stream_id, on_ok, on_error) {
         findExistingVideoTranscoder(stream_id, on_ok, function () {
             var vxcoder = randomId();
-            newTerminal(vxcoder, 'vxcoder', streams[stream_id].owner, false, function () {
+            var terminalId = streams[stream_id].owner;
+            newTerminal(vxcoder, 'vxcoder', streams[stream_id].owner, false, terminals[terminalId].origin, function () {
                 var on_failed = function (error_reason) {
                     makeRPC(
                         rpcClient,
@@ -1422,12 +1426,12 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         });
     };
 
-    that.publish = function (participantId, streamId, accessNode, streamInfo, streamType, on_ok, on_error) {
-        log.debug('publish, participantId: ', participantId, 'streamId:', streamId, 'accessNode:', accessNode.node, 'streamInfo:', JSON.stringify(streamInfo));
+    that.publish = function (participantId, streamId, accessNode, streamInfo, streamType, origin, on_ok, on_error) {
+        log.debug('publish, participantId: ', participantId, 'streamId:', streamId, 'accessNode:', accessNode.node, 'streamInfo:', JSON.stringify(streamInfo), ' origin is:', origin);
         if (streams[streamId] === undefined) {
             var terminal_id = pubTermId(participantId, streamId);
             var terminal_owner = (streamType === 'webrtc' || streamType === 'sip') ? participantId : room_id + '-' + randomId();
-            newTerminal(terminal_id, streamType, terminal_owner, accessNode, function () {
+            newTerminal(terminal_id, streamType, terminal_owner, accessNode, origin, function () {
                 streams[streamId] = {owner: terminal_id,
                                      audio: streamInfo.audio ? {format: formatStr(streamInfo.audio),
                                                                 subscribers: [],
@@ -1680,7 +1684,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
             };
 
             var terminal_owner = (((subType === 'webrtc' || subType === 'sip') && isAudioPubPermitted) ? participantId : room_id);
-            newTerminal(terminal_id, subType, terminal_owner, accessNode, function () {
+            newTerminal(terminal_id, subType, terminal_owner, accessNode, subInfo.origin, function () {
                 doSubscribe();
             }, on_error);
         } else {
@@ -1913,6 +1917,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         }
 
         log.debug('rebuildVideoMixer, vmixerId:', vmixerId, 'view:', view);
+        var origin = terminals[vmixerId].origin;
         for (var sub_id in terminals[vmixerId].subscribed) {
             var vst_id = terminals[vmixerId].subscribed[sub_id].video;
             inputs.push(vst_id);
@@ -1936,7 +1941,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
             }
         });
         terminals[vmixerId].published = [];
-
+        mediaPreference.origin = origin;
         return rpcReq.getWorkerNode(cluster, 'video', {room: room_id, task: vmixerId}, mediaPreference)
             .then(function (locality) {
                 log.debug('Got new video mixer node:', locality);
@@ -2013,6 +2018,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         var input, outputs = [];
 
         log.debug('rebuildVideoTranscoder, vxcoderId:', vxcoderId);
+        var origin = terminals[vxcoderId].origin;
         for (var sub_id in terminals[vxcoderId].subscribed) {
             var vst_id = terminals[vxcoderId].subscribed[sub_id].video;
             input = vst_id;
@@ -2129,6 +2135,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
             }
         }
 
+        var origin = terminals[amixerId].origin;
         for (var sub_id in terminals[amixerId].subscribed) {
             var ast_id = terminals[amixerId].subscribed[sub_id].audio;
             inputs.push(ast_id);
@@ -2154,6 +2161,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         });
         terminals[amixerId].published = [];
 
+        mediaPreference.origin = origin;
         return rpcReq.getWorkerNode(cluster, 'audio', {room: room_id, task: amixerId}, mediaPreference)
             .then(function (locality) {
                 log.debug('Got new audio mixer node:', locality);
@@ -2228,6 +2236,7 @@ module.exports.create = function (spec, on_init_ok, on_init_failed) {
         var old_locality = terminals[axcoderId].locality;
         var input, outputs = [];
 
+        var origin = terminals[axcoderId].origin;
         for (var sub_id in terminals[axcoderId].subscribed) {
             var vst_id = terminals[axcoderId].subscribed[sub_id].audio;
             input = vst_id;
diff --git a/source/agent/conference/stream.js b/source/agent/conference/stream.js
@@ -418,6 +418,9 @@ function toPortalFormat (internalStream) {
       delete videoInfo.alternative;
     }
   }
+  if(stream && stream.info && stream.info.origin) {
+    delete stream.info.origin;
+  }
   return stream;
 };
 
diff --git a/source/agent/streaming/agent.toml b/source/agent/streaming/agent.toml
@@ -21,6 +21,13 @@ max_load = 0.85 #default: 0.85
 #The bandwidth of network-interface used for RTSP/RTMP connections. 
 network_max_scale = 1000 #unit: Mbps
 
+[capacity]
+#The ISP list this agent is able to handle.
+#If the ISP list is set to be non-empty, only the creating token requests with preference.ips being matched with one element of this list will be scheduled to this agent.
+isps = [] #default: [], which means all ISPs.
+#The region list this agent prefers to handle
+#If the region list is set to be non-empty, the creating token requests with preference.region being matched with one element of this list will be priorly scheduled to this agent.
+regions = [] #default: [], which means all regions.
 
 [rabbit]
 host = "localhost" #default: "localhost"
diff --git a/source/agent/video/agent.toml b/source/agent/video/agent.toml
@@ -18,6 +18,13 @@ report_load_interval = 1000 #default: 1000, unit: millisecond
 #The max CPU/GPU load under which this worker can take new tasks.
 max_load = 0.85 #default: 0.85
 
+[capacity]
+#The ISP list this agent is able to handle.
+#If the ISP list is set to be non-empty, only the creating token requests with preference.ips being matched with one element of this list will be scheduled to this agent.
+isps = [] #default: [], which means all ISPs.
+#The region list this agent prefers to handle
+#If the region list is set to be non-empty, the creating token requests with preference.region being matched with one element of this list will be priorly scheduled to this agent.
+regions = [] #default: [], which means all regions.
 
 [rabbit]
 host = "localhost" #default: "localhost"
diff --git a/source/cluster_manager/matcher.js b/source/cluster_manager/matcher.js
diff --git a/source/management_api/restReqValidator.js b/source/management_api/restReqValidator.js

Original file line number	Diff line number	Diff line change
`@@ -418,6 +418,9 @@ function toPortalFormat (internalStream) {`
`418`	`418`	`delete videoInfo.alternative;`
`419`	`419`	`}`
`420`	`420`	`}`
	`421`	`+ if(stream && stream.info && stream.info.origin) {`
	`422`	`+ delete stream.info.origin;`
	`423`	`+ }`
`421`	`424`	`return stream;`
`422`	`425`	`};`
`423`	`426`