Merge pull request #1676 from felixhjh/pipeline_encrypt_model

add pipeline encrypt demo

examples/Pipeline/PaddleClas/DarkNet53-encryption/README.md

+# Imagenet Pipeline WebService
+
+This document will takes Imagenet service as an example to introduce how to use Pipeline WebService.
+
+## Get model
+```
+sh get_model.sh
+python encrypt.py
+```
+
+## Start server
+
+```
+python -m paddle_serving_server.serve --model encrypt_server/ --port 9400  --encryption_rpc_port 9401  --use_encryption_model &
+python web_service.py &>log.txt &
+```
+
+## client test
+```
+python http_client.py
+```
+
+if you configure the api gateway, you can use `https_client.py`

examples/Pipeline/PaddleClas/DarkNet53-encryption/README_CN.md

+# Imagenet Pipeline WebService
+
+这里以 Imagenet 服务为例来介绍 Pipeline WebService 的使用。
+
+## 获取模型
+```
+sh get_model.sh
+python encrypt.py
+```
+
+## 启动服务
+
+```
+python -m paddle_serving_server.serve --model encrypt_server/ --port 9400  --encryption_rpc_port 9401  --use_encryption_model &
+python web_service.py &>log.txt &
+```
+
+## 测试
+```
+python http_client.py
+```
+如果您已经配置好了api gateway， 您可以使用 `https_client.py`
+
+~

examples/Pipeline/PaddleClas/DarkNet53-encryption/config.yml

+#worker_num, 最大并发数。当build_dag_each_worker=True时, 框架会创建worker_num个进程，每个进程内构建grpcSever和DAG
+##当build_dag_each_worker=False时，框架会设置主线程grpc线程池的max_workers=worker_num
+worker_num: 1
+
+#http端口, rpc_port和http_port不允许同时为空。当rpc_port可用且http_port为空时，不自动生成http_port
+http_port: 18080
+rpc_port: 9993
+
+dag:
+    #op资源类型, True, 为线程模型；False，为进程模型
+    is_thread_op: False
+op:
+    imagenet:
+        #并发数，is_thread_op=True时，为线程并发；否则为进程并发
+        concurrency: 1
+        client_type: brpc
+        retry: 1
+        timeout: 3000
+        server_endpoints: ["127.0.0.1:9400"]
+        client_config: "encrypt_client"
+        fetch_list: ["save_infer_model/scale_0.tmp_0"]
+        batch_size: 1
+        auto_batching_timeout: 2000
+        use_encryption_model: True
+        encryption_key: "./key" 

examples/Pipeline/PaddleClas/DarkNet53-encryption/encrypt.py

+from paddle_serving_client.io import inference_model_to_serving
+
+
+def serving_encryption():
+    inference_model_to_serving(
+        dirname="./DarkNet53/ppcls_model/",
+        model_filename="__model__",
+        params_filename="./__params__",
+        serving_server="encrypt_server",
+        serving_client="encrypt_client",
+        encryption=True)
+
+
+if __name__ == "__main__":
+    serving_encryption()

examples/Pipeline/PaddleClas/DarkNet53-encryption/get_model.sh

+wget --no-check-certificate https://paddle-serving.bj.bcebos.com/model/DarkNet53.tar
+tar -xf DarkNet53.tar
+
+wget --no-check-certificate https://paddle-serving.bj.bcebos.com/imagenet-example/image_data.tar.gz
+tar -xzvf image_data.tar.gz

examples/Pipeline/PaddleClas/DarkNet53-encryption/http_client.py

+import numpy as np
+import requests
+import json
+import cv2
+import base64
+import os
+
+def cv2_to_base64(image):
+    return base64.b64encode(image).decode('utf8')
+
+if __name__ == "__main__":
+    url = "http://127.0.0.1:18080/imagenet/prediction"
+    with open(os.path.join(".", "daisy.jpg"), 'rb') as file:
+        image_data1 = file.read()
+    image = cv2_to_base64(image_data1)
+    data = {"key": ["image"], "value": [image]}
+    for i in range(1):
+        r = requests.post(url=url, data=json.dumps(data))
+        print(r.json())

examples/Pipeline/PaddleClas/DarkNet53-encryption/https_client.py

+import numpy as np
+import requests
+import json
+import cv2
+import base64
+import os
+
+def cv2_to_base64(image):
+    return base64.b64encode(image).decode('utf8')
+
+if __name__ == "__main__":
+    url = "https://10.21.8.132:8443/image-clas/imagenet/prediction"
+    with open(os.path.join(".", "daisy.jpg"), 'rb') as file:
+        image_data1 = file.read()
+    image = cv2_to_base64(image_data1)
+    headers = {"Content-Type":"application/json", "apikey":"BlfvO08Z9mQpFjcMagl2dxOIA8h2UVdp", "X-INSTANCE-ID" : "kong_ins10"}
+    data = {"key": ["image"], "value": [image]}
+    for i in range(1):
+        r = requests.post(url=url, headers=headers, data=json.dumps(data),verify=False)
+        print(r.json())

examples/Pipeline/PaddleClas/DarkNet53-encryption/web_service.py

+# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved.
+#
+# 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.
+import sys
+from paddle_serving_app.reader import Sequential, URL2Image, Resize, CenterCrop, RGB2BGR, Transpose, Div, Normalize, Base64ToImage
+from paddle_serving_server.web_service import WebService, Op
+import logging
+import numpy as np
+import base64, cv2
+
+
+class ImagenetOp(Op):
+    def init_op(self):
+        self.seq = Sequential([
+            Resize(256), CenterCrop(224), RGB2BGR(), Transpose((2, 0, 1)),
+            Div(255), Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225],
+                                True)
+        ])
+        self.label_dict = {}
+        label_idx = 0
+        with open("imagenet.label") as fin:
+            for line in fin:
+                self.label_dict[label_idx] = line.strip()
+                label_idx += 1
+
+    def preprocess(self, input_dicts, data_id, log_id):
+        (_, input_dict), = input_dicts.items()
+        batch_size = len(input_dict.keys())
+        imgs = []
+        for key in input_dict.keys():
+            data = base64.b64decode(input_dict[key].encode('utf8'))
+            data = np.fromstring(data, np.uint8)
+            im = cv2.imdecode(data, cv2.IMREAD_COLOR)
+            img = self.seq(im)
+            imgs.append(img[np.newaxis, :].copy())
+        input_imgs = np.concatenate(imgs, axis=0)
+        return {"image": input_imgs}, False, None, ""
+
+    def postprocess(self, input_dicts, fetch_dict, data_id=0, log_id=0):
+        score_list = fetch_dict["save_infer_model/scale_0.tmp_0"]
+        result = {"label": [], "prob": []}
+        for score in score_list:
+            score = score.tolist()
+            max_score = max(score)
+            result["label"].append(self.label_dict[score.index(max_score)]
+                                   .strip().replace(",", ""))
+            result["prob"].append(max_score)
+        result["label"] = str(result["label"])
+        result["prob"] = str(result["prob"])
+        return result, None, ""
+
+
+class ImageService(WebService):
+    def get_pipeline_response(self, read_op):
+        image_op = ImagenetOp(name="imagenet", input_ops=[read_op])
+        return image_op
+
+
+uci_service = ImageService(name="imagenet")
+uci_service.prepare_pipeline_config("config.yml")
+uci_service.run_service()

python/paddle_serving_server/serve.py

@@ -184,6 +184,12 @@ def serve_args():
         default=False,
         action="store_true",
         help="Use encryption model")
+    parser.add_argument(
+        "--encryption_rpc_port", 
+        type=int, 
+        required=False, 
+        default=12000, 
+        help="Port of encryption model, only valid for arg.use_encryption_model")
     parser.add_argument(
         "--use_trt", default=False, action="store_true", help="Use TensorRT")
     parser.add_argument(
@@ -352,8 +358,11 @@ def start_multi_card(args, serving_port=None):  # pylint: disable=doc-string-mis
 
 
 class MainService(BaseHTTPRequestHandler):
+    #def __init__(self):
+    #    print("MainService ___init________\n")
     def get_available_port(self):
-        default_port = 12000
+        global encryption_rpc_port
+        default_port = encryption_rpc_port
         for i in range(1000):
             if port_is_available(default_port + i):
                 return default_port + i
@@ -553,7 +562,8 @@ if __name__ == "__main__":
         p_flag = False
         p = None
         serving_port = 0
-        server = HTTPServer(('0.0.0.0', int(args.port)), MainService)
+        encryption_rpc_port = args.encryption_rpc_port
+        server = HTTPServer(('localhost', int(args.port)), MainService)
         print(
             'Starting encryption server, waiting for key from client, use <Ctrl-C> to stop'
         )

python/pipeline/operator.py

@@ -102,6 +102,8 @@ class Op(object):
         self._retry = max(1, retry)
         self._batch_size = batch_size
         self._auto_batching_timeout = auto_batching_timeout
+        self._use_encryption_model = None
+        self._encryption_key = ""
 
         self._input = None
         self._outputs = []
@@ -161,6 +163,11 @@ class Op(object):
             self._fetch_names = conf.get("fetch_list")
         if self._client_config is None:
             self._client_config = conf.get("client_config")
+        if self._use_encryption_model is None:
+            print ("config use_encryption model here", conf.get("use_encryption_model"))
+            self._use_encryption_model = conf.get("use_encryption_model")
+            if self._encryption_key is None or self._encryption_key=="":
+                self._encryption_key = conf.get("encryption_key")
         if self._timeout is None:
             self._timeout = conf["timeout"]
         if self._timeout > 0:
@@ -409,7 +416,12 @@ class Op(object):
             self._fetch_names = client.fetch_names_
             _LOGGER.info("Op({}) has no fetch name set. So fetch all vars")
         if self.client_type != "local_predictor":
-            client.connect(server_endpoints)
+            if self._use_encryption_model is None or self._use_encryption_model is False:
+               client.connect(server_endpoints)
+            else:
+               print("connect to encryption rpc client")
+               client.use_key(self._encryption_key)
+               client.connect(server_endpoints, encryption=True)
         _LOGGER.info("init_client, feed_list:{}, fetch_list: {}".format(self.right_feed_names, self.right_fetch_names))
         return client