Add scheduler for stream service

Author: starwarfan

source/agent/customized/mycontroller.js

@@ -0,0 +1,41 @@
+// Copyright (C) <2022> Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+
+'use strict';
+
+const log = require('./logger').logger.getLogger('MyController');
+const {DomainHandler, RemoteDomainHandler} = require('./domainHandler');
+
+const STREAM_ENGINE_ID = 'stream-service';
+
+class MyController {
+
+  constructor(rpcId, rpcClient) {
+    this.rpcId = rpcId;
+    this.domain = null;
+    // id => JoinData
+    this.clients = new Map();
+    // id => count
+    this.portals = new Map();
+    // id => stream
+    this.streams = new Map();
+  }
+}
+
+function RPCInterface(rpcClient, rpcID) {
+  const controller = new DomainHandler();
+  const API = {};
+  for (const method of RemoteDomainHandler.supportedMethods) {
+    API[method] = function (req, callback) {
+      controller[method](req).then((ret) => {
+        callback('callback', ret ? ret : 'ok');
+      }).catch((e) => {
+        callback('callback', 'error', e && e.message);
+      });
+    }
+  }
+  return API;
+}
+
+module.exports = RPCInterface;

source/portal/portal.js

@@ -238,13 +238,11 @@ var Portal = function(spec, rpcReq) {
 
   that.onSessionSignaling = function(participantId, sessionId, signaling) {
     log.debug('onSessionSignaling, participantId:', participantId, 'sessionId:', sessionId, 'signaling:', signaling);
-
-    var participant = participants[participantId];
     if (participants[participantId] === undefined) {
       return Promise.reject('Participant has NOT joined');
     }
 
-    return rpcReq.onSessionSignaling(participants[participantId].controller, sessionId, signaling);
+    return rpcReq.onSessionSignaling(participants[participantId].controller, sessionId, signaling, participantId);
   };
 
   that.text = function(participantId, to, msg) {

source/portal/rpcRequest.js

@@ -119,9 +119,6 @@ var RpcRequest = function(rpcChannel) {
   };
 
   that.join = function(controller, roomId, participant) {
-    if (STREAM_ENGINE) {
-      participant.domain = roomId;
-    }
     if (enableGrpc) {
       startConferenceClientIfNeeded(controller);
       const req = {roomId, participant};
@@ -135,6 +132,11 @@ var RpcRequest = function(rpcChannel) {
         });
       });
     }
+    if (STREAM_ENGINE) {
+      participant.domain = roomId;
+      participant.participant = participant.id;
+      return rpcChannel.makeRPC(controller, 'join', [participant], 6000);
+    }
     return rpcChannel.makeRPC(controller, 'join', [roomId, participant], 6000);
   };
 
@@ -153,7 +155,7 @@ var RpcRequest = function(rpcChannel) {
       });
     }
     if (STREAM_ENGINE) {
-      const req = {id: participantId};
+      const req = {id: participantId, participant: participantId};
       return rpcChannel.makeRPC(controller, 'leave', [req]);
     }
     return rpcChannel.makeRPC(controller, 'leave', [participantId]);
@@ -362,6 +364,14 @@ var RpcRequest = function(rpcChannel) {
         });
       });
     }
+    if (STREAM_ENGINE) {
+      const req = {
+        id: sessionId,
+        signaling,
+        participant: participantId
+      };
+      return rpcChannel.makeRPC(controller, 'onSessionSignaling', [req]);
+    }
     return rpcChannel.makeRPC(controller, 'onSessionSignaling', [sessionId, signaling]);
   };
 

source/stream_service/dist.json

@@ -18,6 +18,7 @@
             "domainHandler.js",
             "stateStores.js",
             "stateTypes.js",
+            "scheduler.js",
             "controllers.json",
             "../agent/conference/rpcRequest.js",
             "../common/amqpClient.js",

source/stream_service/scheduler.js

@@ -0,0 +1,116 @@
+// Copyright (C) <2022> Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+
+'use strict';
+
+const log = require('./logger').logger.getLogger('Scheduler');
+
+const HASH_SIZE = 512;
+function stringHash(str) {
+  let hash = 0;
+  if (str.length === 0) return hash;
+  for (let i = 0; i < str.length; i++) {
+    const ch = str.charCodeAt(i);
+    hash = ((hash << 5) - hash) + ch;
+    hash |= 0;
+  }
+  return hash % HASH_SIZE + HASH_SIZE;
+}
+
+const CHECK_INTERVAL = 60 * 1000; // 1 min
+
+class ServiceScheduler {
+  static supportedMethods = [
+    'join',
+    'leave',
+    'publish',
+    'unpublish',
+    'getPublications',
+    'subscribe',
+    'unsubscribe',
+    'getSubscriptions',
+    'streamControl',
+    'subscriptionControl',
+    'onStatus',
+    'addProcessor',
+    'removeProcessor',
+    'getProcessors',
+    'onSessionSignaling',
+  ];
+  constructor(rpcChannel, stateStores) {
+    this.rpcChannel = rpcChannel;
+    this.stateStores = stateStores;
+    this.checkAliveTimer = setInterval(() => {
+      this.checkService();
+    }, CHECK_INTERVAL);
+  }
+
+  async scheduleService(req) {
+    const key = (req.participant || '') + (req.domain || '');
+    const hash = stringHash(key);
+    log.debug('hash:', key, hash);
+    const ret = await this.stateStores.readMany('streamEngineNodes', {});
+    const serviceNodes = ret.data || [];
+    if (serviceNodes.length === 0) {
+      throw new Error('No available stream engine nodes');
+    }
+    const scheduled = await this.stateStores.read('scheduleMaps', {_id: hash});
+    if (scheduled) {
+      log.debug('scheduled:', scheduled);
+      return scheduled.node;
+    } else {
+      const node = serviceNodes[hash % serviceNodes.length].id;
+      log.debug('node:', node);
+      try {
+        const map = {_id: hash, node};
+        await this.stateStores.create('scheduleMaps', map);
+      } catch (e) {
+        log.debug('Failed to update schedule map:', e?.message);
+      }
+      return node;
+    }
+  }
+
+  // Check service availability
+  checkService() {
+    log.debug('Check service availability');
+    this.stateStores.readMany('streamEngineNodes', {})
+      .then(async (ret) => {
+        const serviceNodes = ret.data || [];
+        const req = {id: 'non-existent'};
+        for (const service of serviceNodes) {
+          try {
+            await this.rpcChannel.makeRPC(
+              service.id, 'getNodes', [req]);
+          } catch (e) {
+            log.warn('Failed to call service node:', service.id);
+          }
+        }
+      }).catch((e) => {
+        log.debug('Read service error:', e?.message);
+      });
+  }
+
+  getAPI() {
+    const api = {};
+    const self = this;
+    for (const method of ServiceScheduler.supportedMethods) {
+      api[method] = async function (req, callback) {
+        try {
+          const serviceNode = await self.scheduleService(req);
+          log.debug('Schedule req:', req, serviceNode);
+          const ret = await self.rpcChannel.makeRPC(
+            serviceNode, method, [req]);
+          log.debug('Schedule ret:', ret, serviceNode);
+          callback('callback', ret);
+        } catch (e) {
+          callback('callback', 'error', e?.message || e);
+        }
+      };
+    }
+    return api;
+  }
+}
+
+module.exports.ServiceScheduler = ServiceScheduler;

source/stream_service/service.toml

@@ -1,10 +1,18 @@
 [cluster]
 name = "owt-cluster"
 
+[service]
+#name = "stream-service"
+control_agent = "mycontroller"
+
+#Use scheduler for service requests
+[scheduler]
+name = "stream-service"
+start = true #Start a scheduler with server
+
 [rabbit]
 host = "localhost" #default: "localhost"
 port = 5672 #default: 5672
 
-[service]
-name = "stream-service"
-control_agent = "mycontroller"
+[mongo]
+dataBaseURL = "localhost/owtstreams" #default: "localhost/owtstreams"

source/stream_service/stateStores.js

@@ -65,7 +65,7 @@ class StateStores {
   // CRUD operations
   async create(type, obj) {
     const db = await this.prepare;
-    obj._id = (obj.id || obj._id);
+    obj.id && (obj._id = obj.id);
     return new Promise((resolve, reject) => {
       db.collection(type).insertOne(obj, (err, result) => {
         if (err) {
@@ -78,7 +78,7 @@ class StateStores {
   }
   async read(type, query) {
     const db = await this.prepare;
-    query._id = (query.id || query._id);
+    query.id && (query._id = query.id);
     return new Promise((resolve, reject) => {
       db.collection(type).findOne(query, (err, result) => {
         if (err) {
@@ -109,7 +109,7 @@ class StateStores {
   // update: {jsonPath: value}
   async update(type, filter, updates) {
     const db = await this.prepare;
-    filter._id = (filter.id || filter._id);
+    filter.id && (filter._id = filter.id);
     const mongoUpdate = {'$set': updates};
     for (const key in updates) {
       if (updates[key] === null) {
@@ -130,9 +130,9 @@ class StateStores {
   }
   async delete(type, filter) {
     const db = await this.prepare;
-    filter._id = (filter.id || filter._id);
+    filter.id && (filter._id = filter.id);
     return new Promise((resolve, reject) => {
-      db.collection(type).deleteOne(filter, (err, result) => {
+      db.collection(type).deleteMany(filter, (err, result) => {
         if (err) {
           reject(err);
         } else {
@@ -144,7 +144,7 @@ class StateStores {
   // Add value in set
   async setAdd(type, filter, updates) {
     const db = await this.prepare;
-    filter._id = (filter.id || filter._id);
+    filter.id && (filter._id = filter.id);
     const mongoUpdate = {'$addToSet': updates};
     return new Promise((resolve, reject) => {
       db.collection(type).updateOne(filter, mongoUpdate, (err, result) => {
@@ -159,7 +159,7 @@ class StateStores {
   // Remove value in set
   async setRemove(type, filter, updates) {
     const db = await this.prepare;
-    filter._id = (filter.id || filter._id);
+    filter.id && (filter._id = filter.id);
     const mongoUpdate = {'$pull': updates};
     return new Promise((resolve, reject) => {
       db.collection(type).updateOne(filter, mongoUpdate, (err, result) => {

source/stream_service/stateTypes.js

@@ -242,4 +242,3 @@ exports.Publication = Publication;
 exports.Subscription = Subscription;
 exports.Processor = Processor;
 exports.SourceTrack = SourceTrack;
-