Merge pull request #1275 from starwarfan/mst-p-grpc

Enable gRPC option for internal service

Author: starwarfan

doc/servermd/EnableGRPC.md

@@ -0,0 +1,92 @@
+##### Table of Contents
+* 1.[Introduction](#introduction)
+* 2.[Enable gRPC for internal RPC](#dependencies)
+  * 2.1[Configuration for gRPC](#dependencies1)
+  * 2.2[Enable TLS for gRPC](#dependencies2)
+
+# 1 Introduction
+
+By default, OWT[Open WebRTC Toolkit](https://github.com/open-webrtc-toolkit/owt-server) builds internal RPC utility based on message queue middleware(RabbitMQ). In this document we introduce the gRPC alternative for RabbitMQ, and a step by step guide to setup gRPC as OWT server's RPC framework.
+
+# 2 Enable gRPC for internal RPC
+
+In OWT server package, you can turn on gRPC option by updating configuration files. This section introduces the configuration settings for internal gRPC.
+
+# 2.1 Configuration for gRPC
+
+To enable basic gRPC functionality for OWT server, edit following configuration files in server package:
+
+    cluster_manager/cluster_manager.toml
+    management_api/management_api
+    portal/portal.toml
+    webrtc_agent/agent.toml
+    audio_agent/agent.toml
+    video_agent/agent.toml
+    conference_agent/agent.toml
+    analytics_agent/agent.toml
+    recording_agent/agent.toml
+    streaming_agent/agent.toml
+    quic_agent/agent.toml
+    sip_agent/agent.toml
+    sip_portal/sip_portal.toml
+
+Set the item `enable_grpc` to `true` in these configuration files.
+
+    enable_grpc = true
+
+In `cluster_manager/cluster_manager.toml`, set `grpc_host` of section `[manager]` with `IP|hostname:port` of your own(when using TLS, only hostname is allowed).
+
+    [manager]
+    grpc_host = "localhost:10080"
+
+In other modules' toml files, edit `host` of section `[cluster]` to the value of `cluster_manager`.
+
+    [cluster]
+    host = "localhost:10080"
+
+And to specify hostname of other modules' gRPC service, add following section in toml files.
+
+    [cluster.worker]
+    ip="localhost" # You could also set hostname here. E.g, ip="myhost.com"
+
+If you want to enable HTTP proxy for gRPC, set environment variable `GRPC_ARG_HTTP_PROXY` to `1`, then system proxy will be used.
+
+Restart OWT service after above configuration files being updated.
+Note that RabbitMQ based RPC server will be disabled if gRPC is used.
+
+# 2.2 Enable TLS for gRPC
+
+To enable TLS for gRPC of OWT server, you need prepare SSL certificates for internal gRPC services.
+Both server certificate and client certificate are needed.
+
+You could use following openssl commands to generate self-signed server and client certificates.
+
+    # Generate CA key
+    openssl genrsa -des3 -out ca.key 4096
+    # Generate CA certificate
+    openssl req -new -x509 -days 365 -key ca.key -out ca.crt -subj "/C=XX/ST=XX/L=XX/O=XX/OU=OWT/CN=ca"
+    # Generate server key
+    openssl genrsa -des3 -out server.key 4096
+    # Generate server CSR
+    openssl req -new -key server.key -out server.csr -subj "/C=XX/ST=XX/L=XX/O=XX/OU=OWT/CN=localhost"
+    # Generate server certificate
+    openssl x509 -req -days 365 -in server.csr -CA ca.crt -CAkey ca.key -set_serial 01 -out server.crt
+    # Generate client key
+    openssl genrsa -des3 -out client.key 4096
+    # Generate client CSR
+    openssl req -new -key client.key -out client.csr -subj "/C=XX/ST=XX/L=XX/O=XX/OU=OWT/CN=localhost"
+    # Generate client certificate
+    openssl x509 -req -days 365 -in client.csr -CA ca.crt -CAkey ca.key -set_serial 01 -out client.crt
+
+During creating private keys, you need to enter passphrase for them. To save these passphrases in OWT's credential store, run `(cluster_manager|management_api|portal|webrtc_agent|{MODULE_NAME})/initauth.js --grpc` and save gRPC client/server key passphrase.
+
+For each OWT server component, place root, server and client certificates in the same machine/instance. Make sure OWT processes have pemission to access those certificates.
+Set following environment variables for SSL certificates before starting OWT server:
+
+    OWT_GRPC_ROOT_CERT: The root certificate for server and client certificates.
+    OWT_GRPC_SERVER_CERT: The server certificate.
+    OWT_GRPC_SERVER_KEY: The encrypted server key.
+    OWT_GRPC_CLIENT_CERT: The client certificate.
+    OWT_GRPC_CLIENT_KEY: The encrypted client key.
+
+Once you have updated these settings correctly, TLS for gRPC will be enabled  after restart OWT service.

scripts/release/initauth.js

@@ -122,6 +122,30 @@ const updateInternal = () => new Promise((resolve, reject) => {
     });
 });
 
+const updateGKeyPass = () => new Promise((resolve, reject) => {
+  question('Update passphrase for gRPC TLS key?[yes/no]')
+    .then((answer) => {
+      answer = answer.toLowerCase();
+      if (answer !== 'y' && answer !== 'yes') {
+        resolve();
+        return;
+      }
+      question(`(${authBase}) Enter passphrase of server key: `)
+        .then((serverPass) => {
+          mutableStdout.muted = false;
+          question(`(${authBase}) Enter passphrase of client key: `)
+            .then((clientPass) => {
+              mutableStdout.muted = false;
+              saveAuth({ grpc: { serverPass, clientPass } }, authStore)
+                .then(resolve)
+                .catch(reject);
+            });
+          mutableStdout.muted = true;
+        });
+      mutableStdout.muted = true;
+    });
+});
+
 const options = {};
 const parseArgs = () => {
   if (process.argv.includes('--rabbitmq')) {
@@ -133,6 +157,9 @@ const parseArgs = () => {
   if (process.argv.includes('--internal')) {
     options.internal = true;
   }
+  if (process.argv.includes('--grpc')) {
+    options.grpc = true;
+  }
   if (Object.keys(options).length === 0) {
     options.rabbit = true;
     options.mongo = true;
@@ -156,4 +183,9 @@ parseArgs()
       return updateInternal();
     }
   })
+  .then(() => {
+    if (options.grpc) {
+      return updateGKeyPass();
+    }
+  })
   .finally(() => readline.close());

source/agent/analytics/agent.toml

@@ -5,6 +5,8 @@ maxProcesses = 13 #default: 13
 #Number of precesses that agent runs when it starts. 1 <= prerunProcesses <= maxProcesses.
 prerunProcesses = 2 #default: 2
 
+# Setup as GRPC server
+#enable_grpc = true
 
 [cluster]
 name = "owt-cluster"

source/agent/analytics/dist.json

@@ -23,13 +23,20 @@
             "../../common/makeRPC.js",
             "../../common/rpcChannel.js",
             "../../common/mediaUtil.js",
+            "../../common/grpcTools.js",
+            "../../protos/protoConfig.json",
+            "../../protos/*.proto",
             "../../../scripts/release/initauth.js",
             "../../../scripts/release/initcert.js",
             "../../../scripts/detectOS.sh"
         ],
         "folders": {
             "analytics": [
                 "index.js",
+                "grpcAdapter.js",
+                "../../common/grpcTools.js",
+                "../../protos/protoConfig.json",
+                "../../protos/*.proto",
                 "../connections.js",
                 "../internalConnectionRouter.js"
             ],

source/agent/analytics/grpcAdapter.js

@@ -0,0 +1,113 @@
+// Copyright (C) <2022> Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+
+'use strict';
+
+const unpackOption = require('./grpcTools').unpackOption;
+const packNotification = require('./grpcTools').packNotification;
+
+// Create GRPC interface for analytics agent
+function createGrpcInterface(controller, streamingEmitter) {
+  const that = {};
+
+  // GRPC export
+  that.grpcInterface = {
+    publish: function (call, callback) {
+      const req = call.request;
+      const option = unpackOption(req.type, req.option);
+      controller.publish(req.id, req.type, option, (n, code, data) => {
+        if (code === 'error') {
+          callback(new Error(data), null);
+        } else {
+          callback(null, {id: req.id});
+        }
+      });
+    },
+    unpublish: function (call, callback) {
+      controller.unpublish(call.request.id, (n, code, data) => {
+        if (code === 'error') {
+          callback(new Error(data), null);
+        } else {
+          callback(null, {});
+        }
+      });
+    },
+    subscribe: function (call, callback) {
+      const req = call.request;
+      const option = unpackOption(req.type, req.option);
+      controller.subscribe(req.id, req.type, option, (n, code, data) => {
+        if (code === 'error') {
+          callback(new Error(data), null);
+        } else {
+          callback(null, {id: req.id});
+        }
+      });
+    },
+    unsubscribe: function (call, callback) {
+      controller.unsubscribe(call.request.id, (n, code, data) => {
+        if (code === 'error') {
+          callback(new Error(data), null);
+        } else {
+          callback(null, {});
+        }
+      });
+    },
+    linkup: function (call, callback) {
+      const req = call.request;
+      controller.linkup(
+        req.id, req.from,
+        (n, code, data) => {
+          if (code === 'error') {
+            callback(new Error(data), null);
+          } else {
+            callback(null, {message: data});
+          }
+        });
+    },
+    cutoff: function (call, callback) {
+      controller.cutoff(call.request.id, (n, code, data) => {
+        if (code === 'error') {
+          callback(new Error(data), null);
+        } else {
+          callback(null, {});
+        }
+      });
+    },
+    listenToNotifications: function (call, callback) {
+      const writeNotification = (notification) => {
+        const progress = packNotification({
+          type: 'analytics',
+          name: notification.name,
+          data: notification.data,
+        });
+        call.write(progress);
+      };
+      const endCall = () => {
+        call.end();
+      };
+      streamingEmitter.on('notification', writeNotification);
+      streamingEmitter.on('close', endCall);
+      call.on('cancelled', () => {
+        call.end();
+      });
+      call.on('close', () => {
+        streamingEmitter.off('notification', writeNotification);
+        streamingEmitter.off('close', endCall);
+      });
+    },
+    getInternalAddress: function (call, callback) {
+      controller.getInternalAddress((n, code, data) => {
+          if (code === 'error') {
+            callback(new Error(data), null);
+          } else {
+            callback(null, code);
+          }
+        });
+    },
+  };
+
+  return that;
+}
+
+exports.createGrpcInterface = createGrpcInterface;

source/agent/analytics/index.js

@@ -11,6 +11,12 @@ const VideoAnalyzer = require('../videoGstPipeline/build/Release/videoAnalyzer-p
 
 var {InternalConnectionRouter} = require('./internalConnectionRouter');
 
+// Setup GRPC server
+var createGrpcInterface = require('./grpcAdapter').createGrpcInterface;
+var enableGRPC = global.config.agent.enable_grpc || false;
+
+var EventEmitter = require('events').EventEmitter;
+
 function doPublish(rpcClient, conference, user, streamId, streamInfo) {
   return new Promise(function(resolve, reject) {
     makeRPC(
@@ -50,6 +56,8 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
   var outputs = {};
   var engine = new VideoAnalyzer();
   var controller;
+  // For GRPC notifications
+  var streamingEmitter = new EventEmitter();
 
   try {
     algorithms = toml.parse(fs.readFileSync('./plugin.cfg'));
@@ -59,6 +67,16 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
 
   const notifyStatus = (controller, sessionId, direction, status) => {
     rpcClient.remoteCast(controller, 'onSessionProgress', [sessionId, direction, status]);
+    // Emit GRPC notifications
+    const notification = {
+      name: 'onSessionProgress',
+      data: {
+          id: sessionId,
+          status,
+          direction
+      }
+    };
+    streamingEmitter.emit('notification', notification);
   };
 
   // for algorithms that generate new stream
@@ -71,6 +89,16 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
       .catch((err) => {
         log.debug('Generated stream:', streamId, 'publish failed', err);
       });
+      // Emit GRPC notifications
+      const notification = {
+        name: 'onStreamAdded',
+        data: {
+            id: streamId,
+            owner: 'admin',
+            info: streamInfo
+        }
+      };
+      streamingEmitter.emit('notification', notification);
   };
   // destroy generated stream
   const destroyStream = (controller, streamId) => {
@@ -82,6 +110,15 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
       .catch((err) => {
         log.debug('Destroyed stream:', streamId, 'unpublish failed', err);
       });
+      // Emit GRPC notifications
+      const notification = {
+        name: 'onStreamRemoved',
+        data: {
+            id: streamId,
+            owner: 'admin'
+        }
+      };
+      streamingEmitter.emit('notification', notification);
   };
 
   // RPC callback
@@ -141,7 +178,7 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
       return callback('callback', {type: 'failed', reason: 'invalid connctionType'+connectionType});
     }
 
-      // check options
+    // check options
     if (!algorithms[options.connection.algorithm]) {
       return callback('callback', {type: 'failed', reason: 'Not valid algorithm:'+options.connection.algorithm});
     }
@@ -289,5 +326,10 @@ module.exports = function (rpcClient, rpcId, agentId, clusterIp) {
     }
   };
 
+  if (enableGRPC) {
+    // Export GRPC interface.
+    return createGrpcInterface(that, streamingEmitter);
+  }
+
   return that;
 };

source/agent/analytics/package.json

@@ -1,11 +1,13 @@
 {
-  "name": "OWT-Server-Agent",
+  "name": "OWT-Server-Analytics-Agent",
   "version":"5.0.0",
   "dependencies": {
     "amqplib": "^0.7.0",
     "log4js": "^6.4.0",
     "node-getopt": "*",
-    "toml": "*"
+    "toml": "*",
+    "@grpc/proto-loader": "^0.5.0",
+    "@grpc/grpc-js": "^1.1.0"
   },
   "devDependencies": {
     "node-gyp": "*"