Flow decomposition and miscellenous improvements
Specifically:
The biggest addition is an initial flow decomposition
implementation that splits flows and flow groups
defined over the logical device into per physical
device flows, based on a very crude and heuristic
approach. We expect this part to be much improved
later on, both in term of genericness as well as
speed.
The flow decomposition is triggered by any flow
or group mods applied to a logical device, and it
consequently touches up the affected device tables.
This uses the POST_UPDATE (post-commit) mechanism
of core.
There is also an initial arhcitecture diagram added
under docs.
Additional improvements:
* Implemented metadata passing across the gRPC
link, both in Voltha and in Chameleon. This paves
the road to pass query args as metadata, and also
to pass HTTP header fields back and forth across
the gRPC API. This is alrady used to pass in depth
for GET /api/v1/local, and it will be used to
allow working with transactions and specific config
revs.
* Improved automatic reload and reconnect of chameleon
after Voltha is restarted.
* Improved error handling in gRPC hanlers, especially
for the "resource not found (404)", and bad argument
(400) type errors. This makes gRPC Rendezvous errors
a bit cleaner, and also allows Chameleon to map these
errors into 404/400 codes.
* Better error logging in generic errors in gRPC handlers.
* Many new test-cases
* Initial skeleton and first many steps implemented for
the automated testing for the cold PON activation
sequence.
* Convenience functions for working with flows (exemplified
by the test-cases)
* Fixed bug in config engine that dropped changes that
were made in a POST_* callback, such as the ones used
to propagae the logical flow tables into the device
tables. The fix was to defer the callbacks till the
initial changes are complete and then execute all
callbacks in sequence.
* Adapter proxy with well defined API that can be
used by the adapters to communicate back to Core.
* Extended simulated_olt and simulated_onu adapters to
both demonstrate discovery-style and provisioned
activation style use cases.
* Adapter-, device-, and logical device agents to provide
the active business logic associated with these
entities.
* Fixed 64-bit value passing across the stack. There was
an issue due to inconsistent use of two JSON<-->Proto
librarier, one of which did not adhere to the Google
specs which recommend passing 64-bit integer values as
strings.
* Annotation added for all gRPC methods.
All Voltha test-cases are passing.
Change-Id: Id949e8d1b76276741471bedf9901ac33bfad9ec6
diff --git a/voltha/core/global_handler.py b/voltha/core/global_handler.py
new file mode 100644
index 0000000..2624ed6
--- /dev/null
+++ b/voltha/core/global_handler.py
@@ -0,0 +1,388 @@
+# Copyright 2016 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 structlog
+from grpc import StatusCode
+from twisted.internet.defer import inlineCallbacks
+from twisted.internet.defer import returnValue
+
+from common.utils.grpc_utils import twisted_async
+from voltha.core.config.config_root import ConfigRoot
+from voltha.protos.voltha_pb2 import \
+ add_VolthaGlobalServiceServicer_to_server, VolthaLocalServiceStub, \
+ VolthaGlobalServiceServicer, Voltha, VolthaInstances, VolthaInstance, \
+ LogicalDevice, Ports, Flows, FlowGroups, Device
+from voltha.registry import registry
+from google.protobuf.empty_pb2 import Empty
+
+log = structlog.get_logger()
+
+
+class GlobalHandler(VolthaGlobalServiceServicer):
+
+ def __init__(self, dispatcher, instance_id, **init_kw):
+ self.dispatcher = dispatcher
+ self.instance_id = instance_id
+ self.init_kw = init_kw
+ self.root = None
+ self.stopped = False
+
+ def start(self):
+ log.debug('starting')
+ self.root = ConfigRoot(Voltha(**self.init_kw))
+ registry('grpc_server').register(
+ add_VolthaGlobalServiceServicer_to_server, self)
+ log.info('started')
+ return self
+
+ def stop(self):
+ log.debug('stopping')
+ self.stopped = True
+ log.info('stopped')
+
+ # gRPC service method implementations. BE CAREFUL; THESE ARE CALLED ON
+ # the gRPC threadpool threads.
+
+ @twisted_async
+ def GetVoltha(self, request, context):
+ log.info('grpc-request', request=request)
+ return self.root.get('/', depth=1)
+
+ @twisted_async
+ @inlineCallbacks
+ def ListVolthaInstances(self, request, context):
+ log.info('grpc-request', request=request)
+ items = yield registry('coordinator').get_members()
+ returnValue(VolthaInstances(items=items))
+
+ @twisted_async
+ def GetVolthaInstance(self, request, context):
+ log.info('grpc-request', request=request)
+ instance_id = request.id
+ try:
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'GetVolthaInstance',
+ Empty(),
+ context)
+ except KeyError:
+ context.set_details(
+ 'Voltha instance \'{}\' not found'.format(instance_id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return VolthaInstance()
+
+ @twisted_async
+ def ListLogicalDevices(self, request, context):
+ log.warning('temp-limited-implementation')
+ # TODO dispatching to local instead of collecting all
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'ListLogicalDevices',
+ Empty(),
+ context)
+
+ @twisted_async
+ def GetLogicalDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_logical_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Logical device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return LogicalDevice()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'GetLogicalDevice',
+ request,
+ context)
+
+ @twisted_async
+ def ListLogicalDevicePorts(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_logical_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Logical device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Ports()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'ListLogicalDevicePorts',
+ request,
+ context)
+
+ @twisted_async
+ def ListLogicalDeviceFlows(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_logical_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Logical device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Flows()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'ListLogicalDeviceFlows',
+ request,
+ context)
+
+ @twisted_async
+ def UpdateLogicalDeviceFlowTable(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_logical_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Logical device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Empty()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'UpdateLogicalDeviceFlowTable',
+ request,
+ context)
+
+ @twisted_async
+ def ListLogicalDeviceFlowGroups(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_logical_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Logical device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return FlowGroups()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'ListLogicalDeviceFlowGroups',
+ request,
+ context)
+
+ @twisted_async
+ def UpdateLogicalDeviceFlowGroupTable(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_logical_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Logical device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Empty()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'UpdateLogicalDeviceFlowGroupTable',
+ request,
+ context)
+
+ @twisted_async
+ def ListDevices(self, request, context):
+ log.warning('temp-limited-implementation')
+ # TODO dispatching to local instead of collecting all
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'ListDevices',
+ request,
+ context)
+
+ @twisted_async
+ def GetDevice(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Device()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'GetDevice',
+ request,
+ context)
+
+ @twisted_async
+ def CreateDevice(self, request, context):
+ log.info('grpc-request', request=request)
+ # TODO dispatching to local instead of passing it to leader
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'CreateDevice',
+ request,
+ context)
+
+ @twisted_async
+ def ActivateDevice(self, request, context):
+ log.info('grpc-request', request=request)
+ # TODO dispatching to local instead of passing it to leader
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'ActivateDevice',
+ request,
+ context)
+
+ @twisted_async
+ def ListDevicePorts(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Ports()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'ListDevicePorts',
+ request,
+ context)
+
+ @twisted_async
+ def ListDeviceFlows(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return Flows()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'ListDeviceFlows',
+ request,
+ context)
+
+ @twisted_async
+ def ListDeviceFlowGroups(self, request, context):
+ log.info('grpc-request', request=request)
+
+ try:
+ instance_id = self.dispatcher.instance_id_by_device_id(
+ request.id
+ )
+ except KeyError:
+ context.set_details(
+ 'Device \'{}\' not found'.format(request.id))
+ context.set_code(StatusCode.NOT_FOUND)
+ return FlowGroups()
+
+ return self.dispatcher.dispatch(
+ instance_id,
+ VolthaLocalServiceStub,
+ 'ListDeviceFlowGroups',
+ request,
+ context)
+
+ @twisted_async
+ def ListDeviceTypes(self, request, context):
+ log.info('grpc-request', request=request)
+ # we always deflect this to the local instance, as we assume
+ # they all loaded the same adapters, supporting the same device
+ # types
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'ListDeviceTypes',
+ request,
+ context)
+
+ @twisted_async
+ def GetDeviceType(self, request, context):
+ log.info('grpc-request', request=request)
+ # we always deflect this to the local instance, as we assume
+ # they all loaded the same adapters, supporting the same device
+ # types
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'GetDeviceType',
+ request,
+ context)
+
+ @twisted_async
+ def ListDeviceGroups(self, request, context):
+ log.warning('temp-limited-implementation')
+ # TODO dispatching to local instead of collecting all
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'ListDeviceGroups',
+ Empty(),
+ context)
+
+ @twisted_async
+ def GetDeviceGroup(self, request, context):
+ log.warning('temp-limited-implementation')
+ # TODO dispatching to local instead of collecting all
+ return self.dispatcher.dispatch(
+ self.instance_id,
+ VolthaLocalServiceStub,
+ 'GetDeviceGroup',
+ request,
+ context)
+
+