Add consul as a config backend
Two backends available, none and consul. Can be set through the
BACKEND environment variable or the --backend flag.
Currently access to consul blocks the reactor. Requests to consul
are cached. In my conversations with Zsolt, we concluded that
blocking the reactor is fine for now to prevent multiple tasks
from modifying the config at the same time. An easier first step
to making it all twisted is to add a write queue. That comes with its
own challenges though. A longer goal would be making the entire config
access inlineCallbacks, possibly with a locking mechanism to prevent
multiple tasks from modifying the config at once.
Change-Id: Ia3132a455ca03123dbb9b2e56c9f4df1202347cd
diff --git a/voltha/core/config/config_backend.py b/voltha/core/config/config_backend.py
new file mode 100644
index 0000000..21ef5ec
--- /dev/null
+++ b/voltha/core/config/config_backend.py
@@ -0,0 +1,82 @@
+# 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.
+#
+from consul import Consul
+
+
+class ConsulStore(object):
+ """ Config kv store for consul with a cache for quicker subsequent reads
+
+ TODO: This will block the reactor. Should either change
+ whole call stack to yield or put the put/delete transactions into a
+ queue to write later with twisted. Will need a transaction
+ log to ensure we don't lose anything.
+ Making the whole callstack yield is troublesome because other tasks can
+ come in on the side and start modifying things which could be bad.
+ """
+
+ def __init__(self, host, port, path_prefix):
+ self._consul = Consul(host=host, port=port)
+ self._path_prefix = path_prefix
+ self._cache = {}
+
+ def make_path(self, key):
+ return '{}/{}'.format(self._path_prefix, key)
+
+ def __getitem__(self, key):
+ if key in self._cache:
+ return self._cache[key]
+ index, value = self._consul.kv.get(self.make_path(key))
+ if value is not None:
+ # consul turns empty strings to None, so we do the reverse here
+ self._cache[key] = value['Value'] or ''
+ return value['Value'] or ''
+ else:
+ raise KeyError(key)
+
+ def __contains__(self, key):
+ if key in self._cache:
+ return True
+ index, value = self._consul.kv.get(self.make_path(key))
+ if value is not None:
+ self._cache[key] = value['Value']
+ return True
+ else:
+ return False
+
+ def __setitem__(self, key, value):
+ assert isinstance(value, basestring)
+ self._cache[key] = value
+ self._consul.kv.put(self.make_path(key), value)
+
+ def __delitem__(self, key):
+ self._cache.pop(key, None)
+ self._consul.kv.delete(self.make_path(key))
+
+
+def load_backend(args):
+ """ Return the kv store backend based on the command line arguments
+ """
+ # TODO: Make this more dynamic
+
+ def load_consul_store():
+ host, port = args.consul.split(':', 1)
+ return ConsulStore(host, int(port), 'service/voltha/config_data')
+
+ loaders = {
+ 'none': lambda: None,
+ 'consul': load_consul_store
+ }
+
+ return loaders[args.backend]()
diff --git a/voltha/core/core.py b/voltha/core/core.py
index 34dc0a0..3b28cbd 100644
--- a/voltha/core/core.py
+++ b/voltha/core/core.py
@@ -62,11 +62,11 @@
self.change_event_queue = Queue()
@inlineCallbacks
- def start(self):
+ def start(self, config_backend=None):
log.debug('starting')
yield self.dispatcher.start()
yield self.global_handler.start()
- yield self.local_handler.start()
+ yield self.local_handler.start(config_backend=config_backend)
self.local_root_proxy = self.get_proxy('/')
self.local_root_proxy.register_callback(
CallbackType.POST_ADD, self._post_add_callback)
diff --git a/voltha/core/local_handler.py b/voltha/core/local_handler.py
index 61a6e2c..c0309eb 100644
--- a/voltha/core/local_handler.py
+++ b/voltha/core/local_handler.py
@@ -21,6 +21,7 @@
from common.utils.grpc_utils import twisted_async
from voltha.core.config.config_root import ConfigRoot
+from voltha.core.config.config_backend import ConsulStore
from voltha.protos.openflow_13_pb2 import PacketIn, Flows, FlowGroups, \
ofp_port_status
from voltha.protos.voltha_pb2 import \
@@ -41,9 +42,19 @@
self.root = None
self.stopped = False
- def start(self):
+ def start(self, config_backend=None):
log.debug('starting')
- self.root = ConfigRoot(VolthaInstance(**self.init_kw))
+ if config_backend:
+ if 'root' in config_backend:
+ # This is going to block the entire reactor until loading is completed
+ log.info('loading config from persisted backend')
+ self.root = ConfigRoot.load(VolthaInstance, kv_store=config_backend)
+ else:
+ log.info('initializing new config')
+ self.root = ConfigRoot(VolthaInstance(**self.init_kw), kv_store=config_backend)
+ else:
+ self.root = ConfigRoot(VolthaInstance(**self.init_kw))
+
registry('grpc_server').register(
add_VolthaLocalServiceServicer_to_server, self)
log.info('started')
diff --git a/voltha/main.py b/voltha/main.py
index 9653674..5a31f42 100755
--- a/voltha/main.py
+++ b/voltha/main.py
@@ -37,6 +37,7 @@
from voltha.adapters.loader import AdapterLoader
from voltha.coordinator import Coordinator
from voltha.core.core import VolthaCore
+from voltha.core.config.config_backend import load_backend
from voltha.northbound.diagnostics import Diagnostics
from voltha.northbound.grpc.grpc_server import VolthaGrpcServer
from voltha.northbound.kafka.kafka_proxy import KafkaProxy, get_kafka_proxy
@@ -62,6 +63,7 @@
rest_port=os.environ.get('REST_PORT', 8880),
kafka=os.environ.get('KAFKA', 'localhost:9092'),
manhole_port=os.environ.get('MANHOLE_PORT', 12222),
+ backend=os.environ.get('BACKEND', 'none'),
)
@@ -192,6 +194,12 @@
default=defs['kafka'],
help=_help)
+ _help = 'backend to use for config persitence'
+ parser.add_argument('-b', '--backend',
+ default=defs['backend'],
+ choices=['none', 'consul'],
+ help=_help)
+
args = parser.parse_args()
# post-processing
@@ -305,7 +313,7 @@
version=VERSION,
log_level=LogLevel.INFO
)
- ).start()
+ ).start(config_backend=load_backend(self.args))
yield registry.register(
'frameio',