GRPC implementation

wrappers/s2i/nodejs/Dockerfile

@@ -9,6 +9,10 @@ COPY microservice.js /microservice
 
 COPY package.json /microservice
 
+COPY prediction_grpc_pb.js /microservice
+
+COPY prediction_pb.js /microservice
+
 RUN npm install
 
 COPY ./s2i/bin/ /s2i/bin

wrappers/s2i/nodejs/README.md

@@ -57,7 +57,7 @@ This builds the base wrapper image needed for any nodejs model to be deployed on
 ### Building s2i nodejs model Image
 
 ```
-s2i build test/model-template-app seldonio/seldon-core-s2i-nodejs seldon-core-template-model
+s2i build -E ./test/model-template-app/.s2i/environment test/model-template-app seldonio/seldon-core-s2i-nodejs seldon-core-template-model
 ```
 
 This creates the actual nodejs model image as a seldon component
@@ -77,3 +77,22 @@ Make sure the current user can run npm commands.
 ```
 make test
 ```
+
+### GRPC code-generated Proto JS Files
+
+This is the code pre-generated using protoc and the Node gRPC protoc plugin, and the generated code can be found in various `*_pb.js` files.
+The creation of the grpc srevice assumes these files to be present.
+
+```
+cd ../../../proto/
+npm install -g grpc-tools
+grpc_tools_node_protoc --js_out=import_style=commonjs,binary:../wrappers/s2i/nodejs/ --grpc_out=../wrappers/s2i/nodejs --plugin=protoc-gen-grpc=`which grpc_tools_node_protoc_plugin` prediction.proto
+cd ../wrappers/s2i/nodejs/
+```
+
+### Test using GRPC client
+
+```
+npm i
+node grpc_client.js
+```

wrappers/s2i/nodejs/grpc_client.js

@@ -0,0 +1,61 @@
+/*
+ *
+ * Copyright 2015 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+var messages = require("./prediction_pb");
+var services = require("./prediction_grpc_pb");
+
+var grpc = require("grpc");
+
+function main() {
+  var client = new services.ModelClient(
+    "localhost:5000",
+    grpc.credentials.createInsecure()
+  );
+  var tensorData = new messages.Tensor();
+  tensorData.setShapeList([1, 10]);
+  tensorData.setValuesList([0, 0, 1, 1, 5, 6, 7, 8, 4, 3]);
+
+  var defdata = new messages.DefaultData();
+  defdata.setTensor(tensorData);
+  defdata.setNamesList([]);
+
+  var request = new messages.SeldonMessage();
+  request.setData(defdata);
+  client.predict(request, function(err, response) {
+    if (err) {
+      console.log(err);
+    } else {
+      console.log(
+        "Seldon Message => \n\nNames: ",
+        response.getData().getNamesList(),
+        "\n\nShape: ",
+        response
+          .getData()
+          .getTensor()
+          .getShapeList(),
+        "\n\nValues: ",
+        response
+          .getData()
+          .getTensor()
+          .getValuesList()
+      );
+    }
+  });
+}
+
+main();

wrappers/s2i/nodejs/microservice.js

@@ -3,9 +3,13 @@ const express = require("express");
 const app = express();
 const bodyParser = require("body-parser");
 const nj = require("numjs");
-app.use(bodyParser.urlencoded({ extended: true }));
+const grpc = require("grpc");
+const grpc_messages = require("./prediction_pb");
+const grpc_services = require("./prediction_grpc_pb");
+const process = require("process");
 let predict = null;
 const port = process.env.PREDICTIVE_UNIT_SERVICE_PORT || 5000;
+
 const loadModel = async function(model) {
   model = "./model/" + model;
   try {
@@ -58,6 +62,26 @@ const array_to_rest_data = function(array, original_datadef) {
   return data;
 };
 
+const array_to_grpc_data = function(array, original_datadef) {
+  array = nj.array(array);
+
+  var defdata = new grpc_messages.DefaultData();
+  defdata.setNamesList(get_predict_classNames(array.shape[1]));
+  if (original_datadef["tensor"]) {
+    var tensorData = new grpc_messages.Tensor();
+    tensorData.setShapeList(array.shape.length > 1 ? array.shape : []);
+    tensorData.setValuesList(array.flatten().tolist());
+    defdata.setTensor(tensorData);
+  } else if (original_datadef["ndarray"]) {
+    datadef.setNdarray(array.tolist());
+  } else {
+    datadef.setNdarray(array.tolist());
+  }
+  var data = new grpc_messages.SeldonMessage();
+  data.setData(defdata);
+  return data;
+};
+
 const parser = new argparse.ArgumentParser({
   description: "Seldon-core nodejs microservice builder",
   addHelp: true
@@ -81,34 +105,70 @@ parser.addArgument("--persistence", {
 });
 const args = parser.parseArgs();
 
-console.log(args.model, args.api, args.service, args.persistence);
-
-if (args.service === "MODEL" && args.api === "REST") {
-  app.post("/predict", async (req, res) => {
-    try {
-      body = JSON.parse(req.body.json);
-      body = body.data;
-    } catch (msg) {
-      console.log(msg);
-      res.status(500).send("Cannot parse predict input json " + req.body);
-    }
-    if (predict && typeof predict === "function") {
-      result = predict(rest_data_to_array(body), body.names);
-      result = { data: array_to_rest_data(result, body) };
-      res.status(200).send(result);
-    } else {
-      console.log("Predict function not Found");
-      res.status(500).send(predict);
-    }
-  });
-}
-
-app.listen(port, async () => {
+const getPredictFunction = async () => {
   predict = await loadModel(
     args.model,
     args.api,
     args.service,
     args.persistence
   );
-  console.log(`NodeJs Microservice listening on port ${port}!`);
-});
+  if (predict && typeof predict === "function") {
+    console.log("Predict function loaded successfully");
+    createServer();
+  } else {
+    console.log("Predict function not Found ", predict);
+    process.exit(1);
+  }
+};
+console.log(args.model, args.api, args.service, args.persistence);
+getPredictFunction();
+
+const createServer = () => {
+  if (args.service === "MODEL" && args.api === "REST") {
+    app.use(bodyParser.urlencoded({ extended: true }));
+    app.post("/predict", (req, res) => {
+      try {
+        body = JSON.parse(req.body.json);
+        body = body.data;
+      } catch (msg) {
+        console.log(msg);
+        res.status(500).send("Cannot parse predict input json " + req.body);
+      }
+      if (predict && typeof predict === "function") {
+        result = predict(rest_data_to_array(body), body.names);
+        result = { data: array_to_rest_data(result, body) };
+        res.status(200).send(result);
+      } else {
+        console.log("Predict function not Found");
+        res.status(500).send(predict);
+      }
+    });
+    app.listen(port, () => {
+      console.log(`NodeJs REST Microservice listening on port ${port}!`);
+    });
+  }
+
+  if (args.service === "MODEL" && args.api === "GRPC") {
+    function predictEndpoint(call, callback) {
+      let data = call.request.getData();
+      let body = { names: data.getNamesList() };
+
+      if (data.hasTensor()) {
+        data = data.getTensor();
+        body["tensor"] = {
+          shape: data.getShapeList(),
+          values: data.getValuesList()
+        };
+      } else {
+        body["ndarray"] = data.getNdarray();
+      }
+      result = predict(rest_data_to_array(body), body.names);
+      callback(null, array_to_grpc_data(result, body));
+    }
+    var server = new grpc.Server();
+    server.addService(grpc_services.ModelService, { predict: predictEndpoint });
+    server.bind("0.0.0.0:" + port, grpc.ServerCredentials.createInsecure());
+    server.start();
+    console.log(`NodeJs GRPC Microservice listening on port ${port}!`);
+  }
+};

wrappers/s2i/nodejs/package.json

@@ -15,6 +15,8 @@
     "argparse": "1.0.10",
     "body-parser": "1.18.3",
     "express": "4.16.3",
+    "google-protobuf": "3.6.1",
+    "grpc": "1.15.1",
     "numjs": "0.16.0"
   }
 }