feat(proto): versioning protobuf and validating in recv_message

gnes/__init__.py

@@ -14,7 +14,10 @@
 #  limitations under the License.
 
 
-# do not change this line
-# this is managed by git tag and replaced on every release
+# do not change this line manually
+# this is managed by git tag and updated on every release
 __version__ = '0.0.38'
-__proto_version__ = '20190905'
+
+# do not change this line manually
+# this is managed by shell/make-proto.sh and updated on every execution
+__proto_version__ = '0.0.4'

gnes/cli/parser.py

@@ -151,6 +151,9 @@ def set_service_parser(parser=None):
     parser.add_argument('--parallel_type', type=ParallelType.from_string, choices=list(ParallelType),
                         default=ParallelType.PUSH_NONBLOCK,
                         help='parallel type of the concurrent services')
+    parser.add_argument('--check_version', action='store_true', default=False,
+                        help='comparing the GNES and proto version of incoming message with local setup, '
+                             'mismatch raise an exception')
     return parser
 
 

gnes/client/base.py

@@ -55,6 +55,6 @@ def send_message(self, message: "gnes_pb2.Message", timeout: int = -1):
         send_message(self.sender, message, timeout=timeout)
 
     def recv_message(self, timeout: int = -1) -> gnes_pb2.Message:
-        r = recv_message(self.receiver, timeout=timeout)
+        r = recv_message(self.receiver, timeout=timeout, check_version=self.args.check_version)
         self.logger.info('recv a message: %s' % r.envelope)
         return r

gnes/proto/__init__.py

@@ -133,7 +133,7 @@ def send_message(sock: 'zmq.Socket', msg: 'gnes_pb2.Message', timeout: int = -1)
         sock.setsockopt(zmq.SNDTIMEO, -1)
 
 
-def recv_message(sock: 'zmq.Socket', timeout: int = -1) -> Optional['gnes_pb2.Message']:
+def recv_message(sock: 'zmq.Socket', timeout: int = -1, check_version: bool = False) -> Optional['gnes_pb2.Message']:
     response = []
     try:
         if timeout > 0:
@@ -144,6 +144,21 @@ def recv_message(sock: 'zmq.Socket', timeout: int = -1) -> Optional['gnes_pb2.Me
         _, msg_data = sock.recv_multipart()
         msg = gnes_pb2.Message()
         msg.ParseFromString(msg_data)
+
+        if check_version and msg.envelope:
+            from .. import __version__, __proto_version__
+            if hasattr(msg.envelope, 'gnes_version') and __version__ != msg.envelope.gnes_version:
+                raise AttributeError('mismatched GNES version! '
+                                     'incoming message has GNES version %s, whereas local GNES version %s' % (
+                                         msg.envelope.gnes_version, __version__))
+            if hasattr(msg.envelope, 'proto_version') and __proto_version__ != msg.envelope.proto_version:
+                raise AttributeError('mismatched protobuf version! '
+                                     'incoming message has protobuf version %s, whereas local protobuf version %s' % (
+                                         msg.envelope.proto_version, __proto_version__))
+            if not hasattr(msg.envelope, 'proto_version') and not hasattr(msg.envelope, 'gnes_version'):
+                raise AttributeError('version_check=True locally, '
+                                     'but incoming message contains no version info in its envelope. '
+                                     'the message is probably sent from an outdated GNES service')
         return msg
 
     except ValueError:

gnes/proto/gnes.proto

@@ -92,6 +92,10 @@ message Envelope {
         google.protobuf.Timestamp timestamp = 2;
     }
     repeated route routes = 6;
+
+    string gnes_version = 7;
+
+    string proto_version = 8;
 }
 
 message Message {

gnes/service/base.py

@@ -394,7 +394,7 @@ def _run(self, ctx):
                         self.is_handler_done.clear()
 
                         # receive message
-                        msg = recv_message(pull_sock)
+                        msg = recv_message(pull_sock, check_version=self.args.check_version)
 
                         # choose output sock
                         if msg.request and msg.request.WhichOneof('body') and \

shell/make-proto.sh

@@ -1,7 +1,10 @@
 #!/usr/bin/env bash
 
+set -e
+
 SRC_NAME=gnes.proto
 SRC_DIR=../gnes/proto/
+VER_FILE=../gnes/__init__.py
 
 # generating test proto
 
@@ -11,7 +14,21 @@ SRC_DIR=../gnes/proto/
 PLUGIN_PATH=/Volumes/TOSHIBA-4T/Documents/grpc/bins/opt/grpc_python_plugin
 #PLUGIN_PATH=/user/local/grpc/bins/opt/grpc_python_plugin
 
+printf "\e[1;33mgenerating protobuf and grpc python interface\e[0m\n"
+
 protoc -I ${SRC_DIR} --python_out=${SRC_DIR} --grpc_python_out=${SRC_DIR} --plugin=protoc-gen-grpc_python=${PLUGIN_PATH} ${SRC_DIR}${SRC_NAME}
 
+printf "\e[1;33mfixing grpc import\e[0m\n"
 # fix import bug in google generator
 sed -i '' -e '4s/.*/from\ \.\ import\ gnes_pb2\ as\ gnes__pb2/' ${SRC_DIR}gnes_pb2_grpc.py
+
+# update protobuf version in gnes/__init__.py
+
+OLDVER=$(sed -n 's/^__proto_version__ = '\''\(.*\)'\''$/\1/p' $VER_FILE)
+printf "current proto version:\t\e[1;33m$OLDVER\e[0m\n"
+
+NEWVER=$(echo $OLDVER | awk -F. -v OFS=. 'NF==1{print ++$NF}; NF>1{$NF=sprintf("%0*d", length($NF), ($NF+1)); print}')
+printf "bump proto version to:\t\e[1;32m$NEWVER\e[0m\n"
+
+sed -i '' -e 's/^__proto_version__ = '\''\(.*\)'\''/__proto_version__ = '\'"$NEWVER"\''/' $VER_FILE
+printf "\e[1;32mAll done!\e[0m\n"