feat(rpc): federation service (bytedance#78)

web_console/constants/index.js

@@ -1,3 +1,5 @@
+const path = require('path');
+
 const DEFAULT_SERVER_CONFIG = {
   SERVER_CIPHER: 'leader',
   SERVER_DECIPHER: 'follower',
@@ -8,6 +10,12 @@ const DEFAULT_SERVER_CONFIG = {
   NAMESPACE: 'default',
   ES_HOST: 'fedlearner-stack-elasticsearch-client',
   ES_PORT: 9200,
+  GRPC_HOST: 'localhost',
+  GRPC_PORT: 50051,
+  GRPC_AUTHORITY: 'FL',
+  GRPC_CA: path.resolve(__dirname, 'tests', 'fixtures', 'ca.pem'),
+  GRPC_KEY: path.resolve(__dirname, 'tests', 'fixtures', 'server.key'),
+  GRPC_CERT: path.resolve(__dirname, 'tests', 'fixtures', 'server.pem'),
 };
 
 module.exports = {

web_console/models/federation.js

@@ -40,17 +40,11 @@ module.exports = (sequelize, DataTypes) => {
       default: null,
       comment: 'the web console endpoint',
     },
-    token: {
-      type: DataTypes.STRING(16),
+    fingerprint: {
+      type: DataTypes.STRING(64),
       allowNull: true,
       default: null,
-      comment: 'client-side certificate',
-    },
-    cipher: {
-      type: DataTypes.STRING(200),
-      allowNull: true,
-      default: null,
-      comment: 'used for authorization. null stands for a passive pair, others stands for initiative pair',
+      comment: 'fingerprint generated from RSA public key',
     },
     k8s_settings: {
       type: DataTypes.TEXT('long'),

web_console/package.json

@@ -82,6 +82,8 @@
     "url": "https://github.com/bytedance/fedlearner.git"
   },
   "dependencies": {
+    "@grpc/grpc-js": "1.1.1",
+    "@grpc/proto-loader": "0.5.4",
     "@koa/router": "8.0.8",
     "@zeit-ui/react": "1.6.2",
     "@zeit-ui/react-icons": "1.2.2",

web_console/rpc/client.js

@@ -0,0 +1,74 @@
+/**
+ * for debugging with this client, just set environment variables by:
+ * ```
+ * export GRPC_TRACE=all
+ * export GRPC_VERBOSITY=DEBUG
+ * ```
+ */
+const path = require('path');
+const grpc = require('@grpc/grpc-js');
+const protoLoader = require('@grpc/proto-loader');
+// const { readFileSync } = require('../utils');
+const getConfig = require('../utils/get_confg');
+
+const config = getConfig({
+  GRPC_CA: process.env.GRPC_CA,
+  GRPC_KEY: process.env.GRPC_KEY,
+  GRPC_CERT: process.env.GRPC_CERT,
+  GRPC_AUTHORITY: process.env.GRPC_AUTHORITY,
+  GRPC_HOST: process.env.GRPC_HOST,
+  GRPC_PORT: process.env.GRPC_PORT,
+});
+// const ca = readFileSync(config.GRPC_CA);
+
+const packageDefinition = protoLoader.loadSync(
+  path.resolve(__dirname, 'meta.proto'),
+  {
+    keepCase: true,
+    longs: String,
+    enums: String,
+    defaults: true,
+    oneofs: true,
+  });
+const pkg = grpc.loadPackageDefinition(packageDefinition);
+
+class FederationClient {
+  constructor(host, federation, fingerprint) {
+    // TODO: use federation signature for production @peng09
+    // this.channelCredentials = grpc.credentials.createSsl(ca);
+    this.metadata = new grpc.Metadata();
+    this.metadata.set('authority', config.GRPC_AUTHORITY);
+    this.metadata.set('x-host', host);
+    this.metadata.set('x-federation', federation);
+    this.metadata.set('x-fingerprint', fingerprint);
+    // this.callCredentials = grpc.credentials.createFromMetadataGenerator((options, callback) => {
+    //   callback(null, this.metadata);
+    // });
+    this._client = new pkg.federation.Federation(
+      `${config.GRPC_HOST}:${config.GRPC_PORT}`,
+      // grpc.credentials.combineChannelCredentials(this.channelCredentials, this.callCredentials),
+      grpc.credentials.createInsecure(),
+    );
+    this._request = (method, params) => new Promise((resolve, reject) => {
+      this._client[method](params, this.metadata, (err, data) => {
+        if (err) reject(err);
+        resolve(data);
+      });
+    });
+  }
+
+  // Protobuf limit: `job_type` must be a string
+  getTickets(params = { job_type: '', role: 'leader', offset: 0, limit: 10 }) {
+    return this._request('getTickets', params);
+  }
+
+  createJob(params) {
+    return this._request('createJob', params);
+  }
+
+  deleteJob(params) {
+    return this._request('deleteJob', params);
+  }
+}
+
+module.exports = FederationClient;

web_console/rpc/meta.proto

@@ -0,0 +1,68 @@
+// Federation Service
+// a service used for cross-platform communication
+
+syntax = "proto3";
+
+package federation;
+
+message Ticket {
+  string name = 1;
+  string job_type = 2;
+  string role = 3;
+  string sdk_version = 4;
+  string comment = 5;
+  string params = 6;
+}
+
+message Job {
+  string name = 1;
+  string job_type = 2;
+  string client_ticket_name = 3;
+  string server_ticket_name = 4;
+  string server_params = 5;
+  string client_params = 6;
+}
+
+message GetTicketsRequest {
+  string job_type = 1;
+  string role = 2;
+  int32 offset = 3;
+  int32 limit = 4;
+}
+
+message GetTicketsResponse {
+  repeated Ticket data = 1;
+  int32 total = 2;
+}
+
+message CreateJobRequest {
+  string name = 1;
+  string job_type = 2;
+  string client_ticket_name = 3;
+  string server_ticket_name = 4;
+  string server_params = 5;
+  string client_params = 6;
+}
+
+message CreateJobResponse {
+  Job data = 1;
+}
+
+message DeleteJobRequest {
+  string name = 1;
+}
+
+message DeleteJobResponse {
+  string message = 1;
+}
+
+service Federation {
+  // Obtain available tickets
+  rpc GetTickets(GetTicketsRequest) returns (GetTicketsResponse) {}
+
+  // Obtain token from preset cipher
+  rpc CreateJob(CreateJobRequest) returns (CreateJobResponse) {}
+
+  // Obtain token from preset cipher
+  rpc DeleteJob(DeleteJobRequest) returns (DeleteJobResponse) {}
+}