The commit consists of:
1) Support metadata (e.g. get-depth) for rpc request.
2) Support parameters in rpc requests
3) Generate the code for netconf rpc to Voltha grpc mapping
4) Initial Support custom rpc requests (those defined in the voltha YANG schemas).
Change-Id: I24dc7fd75b5f71d0d8591637579672b25fda57ec
diff --git a/netconf/grpc_client/grpc_client.py b/netconf/grpc_client/grpc_client.py
index 159abc2..9dd7f87 100644
--- a/netconf/grpc_client/grpc_client.py
+++ b/netconf/grpc_client/grpc_client.py
@@ -36,10 +36,11 @@
from netconf.protos.schema_pb2 import SchemaServiceStub
from google.protobuf.empty_pb2 import Empty
from common.utils.consulhelpers import get_endpoint_from_consul
-from netconf.protos.voltha_pb2 import VolthaLocalServiceStub, \
- VolthaGlobalServiceStub
-from google.protobuf import empty_pb2
-from google.protobuf.json_format import MessageToDict, ParseDict
+# from netconf.protos.voltha_pb2 import VolthaLocalServiceStub, \
+# VolthaGlobalServiceStub
+# from google.protobuf import empty_pb2
+# from google.protobuf.json_format import MessageToDict, ParseDict
+from nc_rpc_mapper import get_nc_rpc_mapper_instance
from google.protobuf import descriptor
import base64
import math
@@ -154,9 +155,6 @@
if self.reconnect_callback is not None:
reactor.callLater(0, self.reconnect_callback)
- # self.local_stub = voltha_pb2.VolthaLocalServiceStub(self.channel)
- # self.global_stub = voltha_pb2.VolthaGlobalServiceStub(self.channel)
-
return
except _Rendezvous, e:
@@ -257,6 +255,8 @@
'-I%s '
'--python_out=. '
'--grpc_python_out=. '
+ '--plugin=protoc-gen-gw=%s/rpc_gw_gen.py '
+ '--gw_out=. '
'--plugin=protoc-gen-custom=%s/proto2yang.py '
'%s'
'%s' % (
@@ -265,6 +265,7 @@
':'.join([google_api_dir, netconf_base_dir]),
google_api_dir,
self.plugin_dir,
+ self.plugin_dir,
'--custom_out=. ' if need_yang else '',
fname)
)
@@ -272,17 +273,9 @@
os.system(cmd)
log.info('compiled', file=fname)
- # # test-load each _pb2 file to see all is right
- # if self.work_dir not in sys.path:
- # sys.path.insert(0, self.work_dir)
- #
- # for fname in [f for f in os.listdir(self.work_dir)
- # if f.endswith('_pb2.py')]:
- # modname = fname[:-len('.py')]
- # log.debug('test-import', modname=modname)
- # _ = __import__(modname)
-
- # TODO: find a different way to test the generated yang files
+ # Load the generated modules
+ mapper = get_nc_rpc_mapper_instance(self.work_dir, self)
+ mapper.load_modules()
def _set_yang_schemas(self):
if self.work_dir not in sys.path:
@@ -298,35 +291,27 @@
self.yang_schemas.add(fname[:-len('.yang')])
log.info('yang-schemas', schemas=self.yang_schemas)
- # TODO: should be generated code
- # Focus for now is issuing a GET request for VolthaGlobalService or VolthaLocalService
@inlineCallbacks
- def invoke_voltha_api(self, key):
- # TODO: This should be part of a parameter request
- depth = [('get-depth', '-1')]
+ def invoke_voltha_rpc(self, service, method, params, metadata=None):
try:
- data = {}
- req = ParseDict(data, empty_pb2.Empty())
- service_method = key.split('-')
- service = service_method[0]
- method = service_method[1]
- if service == 'VolthaGlobalService':
- stub = VolthaGlobalServiceStub
- elif service == 'VolthaLocalService':
- stub = VolthaLocalServiceStub
- else:
- raise # Exception
+ mapper = get_nc_rpc_mapper_instance()
- log.info('voltha-rpc', service=service, method=method, req=req,
- depth=depth)
+ # Get the mapping function using the service and method name
+ func = mapper.get_function(service, method)
+ if func is None:
+ log.info('unsupported-rpc', service=service, method=method)
+ return
- res, metadata = yield self.invoke(stub, method, req, depth)
+ response = yield func(self, params, metadata)
- # returnValue(MessageToDict(res, True, True))
- returnValue(self.convertToDict(res))
+ log.info('rpc-result', service=service, method=method,
+ response=response)
+
+ returnValue(response)
except Exception, e:
- log.error('failure', exception=repr(e))
+ log.exception('rpc-failure', service=service, method=method,
+ params=params, e=e)
@inlineCallbacks
def invoke(self, stub, method_name, request, metadata, retry=1):
diff --git a/netconf/grpc_client/nc_rpc_mapper.py b/netconf/grpc_client/nc_rpc_mapper.py
new file mode 100644
index 0000000..dacfc2b
--- /dev/null
+++ b/netconf/grpc_client/nc_rpc_mapper.py
@@ -0,0 +1,84 @@
+#
+# Copyright 2017 the original author or 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.
+#
+import os
+import sys
+import inspect
+from structlog import get_logger
+
+log = get_logger()
+
+
+class NetconfRPCMapper:
+ # Keeps the mapping between a Netconf RPC request and a voltha GPRC
+ # request. Singleton class.
+
+
+ instance = None
+
+ def __init__(self, work_dir, grpc_client):
+ self.work_dir = work_dir
+ self.grpc_client = grpc_client
+ self.rpc_map = {}
+
+ def _add_rpc_map(self, func_name, func_ref):
+ if not self.rpc_map.has_key(func_name):
+ log.debug('adding-function', name=func_name, ref=func_ref)
+ self.rpc_map[func_name] = func_ref
+
+ def _add_module_rpc(self, mod):
+ for name, ref in self.list_functions(mod):
+ self._add_rpc_map(name, ref)
+
+ def is_mod_function(self, mod, func):
+ return inspect.isfunction(func) and inspect.getmodule(func) == mod
+
+ def list_functions(self, mod):
+ return [(func.__name__, func) for func in mod.__dict__.itervalues()
+ if self.is_mod_function(mod, func)]
+
+ def load_modules(self):
+ if self.work_dir not in sys.path:
+ sys.path.insert(0, self.work_dir)
+
+ for fname in [f for f in os.listdir(self.work_dir)
+ if f.endswith('_rpc_gw.py')]:
+ modname = fname[:-len('.py')]
+ log.debug('load-modules', modname=modname)
+ try:
+ m = __import__(modname)
+ self._add_module_rpc(m)
+ except Exception, e:
+ log.exception('loading-module-exception', modname=modname, e=e)
+
+ def get_function(self, service, method):
+ if service:
+ func_name = ''.join([service, '_', method])
+ else:
+ func_name = method
+
+ if self.rpc_map.has_key(func_name):
+ return self.rpc_map[func_name]
+ else:
+ return None
+
+ def is_rpc_exist(self, rpc_name):
+ return self.rpc_map.has_key(rpc_name)
+
+
+def get_nc_rpc_mapper_instance(work_dir=None, grpc_client=None):
+ if NetconfRPCMapper.instance == None:
+ NetconfRPCMapper.instance = NetconfRPCMapper(work_dir, grpc_client)
+ return NetconfRPCMapper.instance