VOL-724 VOL-2106 Return of MIB external storage
- New lazy writer in-memory storage class. Lazy writes mib_db_dict to etcd if dirty. No need to write for every set()
This is the preferred storage class and enables adapter restart/reconciliation
- New Twisted etcd storage class. defers to threads given etcd3 blocks. should improve performance.
- Totally refactored mib_ext_db and mib_alarm_ext_db.
Before they were totally unusable, but now _ext_db can be used if chosen. Warning, they do not use the Twisted etcd storage given
the ripple effect change through most of openomci needed. So they use the blocking etcd3. With 100s of onu this
will cause timeouts.
Given every set() is a write to etcd use of these classes should be discouraged for performance.
- Modifed mib template task to used async twisted etcd class. Also update db interface api to include new template functions
- No longer de-jsonify certain complex attributes in in-memory dict_db.
they must be escaped json string in the field in order for mib storage/recovery to work
Change-Id: I1b77e48ca4185542fa19016200a6cdd5daca0764
diff --git a/Makefile b/Makefile
index 43bab9e..5ddfc8f 100644
--- a/Makefile
+++ b/Makefile
@@ -54,6 +54,7 @@
clean:
find . -name '*.pyc' | xargs rm -f
+ find . -name '__pycache__' | xargs rm -rf
rm -rf \
.tox \
.coverage \
diff --git a/VERSION b/VERSION
index 56f55dc..3d55c7a 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-2.3.24
+2.3.25
diff --git a/pyvoltha/adapters/common/kvstore/etcd_store.py b/pyvoltha/adapters/common/kvstore/etcd_store.py
new file mode 100644
index 0000000..abae7e4
--- /dev/null
+++ b/pyvoltha/adapters/common/kvstore/etcd_store.py
@@ -0,0 +1,40 @@
+# Copyright 2020 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 __future__ import absolute_import
+
+import etcd3
+
+
+class EtcdStore(object):
+
+ def __init__(self, host, port, path_prefix):
+ self._etcd = etcd3.client(host=host, port=port)
+ self.host = host
+ self.port = port
+ self._path_prefix = path_prefix
+
+ def make_path(self, key):
+ return '{}/{}'.format(self._path_prefix, key)
+
+ def get(self, key):
+ (value, meta) = self._etcd.get(self.make_path(key))
+ return value
+
+ def set(self, key, value):
+ self._etcd.put(self.make_path(key), value)
+
+ def delete(self, key):
+ success = self._etcd.delete(self.make_path(key))
+ return success
diff --git a/pyvoltha/adapters/common/kvstore/twisted_etcd_store.py b/pyvoltha/adapters/common/kvstore/twisted_etcd_store.py
new file mode 100644
index 0000000..7003a74
--- /dev/null
+++ b/pyvoltha/adapters/common/kvstore/twisted_etcd_store.py
@@ -0,0 +1,76 @@
+# Copyright 2020 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 __future__ import absolute_import
+from twisted.internet import threads
+
+import etcd3
+
+
+class TwistedEtcdStore(object):
+
+ def __init__(self, host, port, path_prefix):
+ self._etcd = etcd3.client(host=host, port=port)
+ self.host = host
+ self.port = port
+ self._path_prefix = path_prefix
+
+ def make_path(self, key):
+ return '{}/{}'.format(self._path_prefix, key)
+
+ def get(self, key):
+
+ def success(results):
+ (value, meta) = results
+ return value
+
+ def failure(exception):
+ raise exception
+
+ deferred = threads.deferToThread(self._etcd.get, self.make_path(key))
+ deferred.addCallback(success)
+ deferred.addErrback(failure)
+ return deferred
+
+ def set(self, key, value):
+
+ def success(results):
+ if results:
+ return results
+ else:
+ return False
+
+ def failure(exception):
+ raise exception
+
+ deferred = threads.deferToThread(self._etcd.put, self.make_path(key), value)
+ deferred.addCallback(success)
+ deferred.addErrback(failure)
+ return deferred
+
+ def delete(self, key):
+
+ def success(results):
+ if results:
+ return results
+ else:
+ return False
+
+ def failure(exception):
+ raise exception
+
+ deferred = threads.deferToThread(self._etcd.delete, self.make_path(key))
+ deferred.addCallback(success)
+ deferred.addErrback(failure)
+ return deferred
diff --git a/pyvoltha/adapters/extensions/omci/database/alarm_db_ext.py b/pyvoltha/adapters/extensions/omci/database/alarm_db_ext.py
index 6159c21..8df372d 100644
--- a/pyvoltha/adapters/extensions/omci/database/alarm_db_ext.py
+++ b/pyvoltha/adapters/extensions/omci/database/alarm_db_ext.py
@@ -13,16 +13,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from __future__ import absolute_import
+from __future__ import absolute_import, division
from .mib_db_api import *
-from voltha_protos.omci_alarm_db_pb2 import AlarmInstanceData, AlarmClassData, \
- AlarmDeviceData, AlarmAttributeData
-from pyvoltha.common.config.config_backend import EtcdStore
+from pyvoltha.adapters.extensions.omci.database.mib_db_ext import MibDbExternal, MibDbStatistic
+from pyvoltha.adapters.common.kvstore.etcd_store import EtcdStore
from pyvoltha.common.utils.registry import registry
-import six
-from six.moves import range
-class AlarmDbExternal(MibDbApi):
+
+class AlarmDbExternal(MibDbExternal):
"""
A persistent external OpenOMCI Alarm Database
"""
@@ -31,13 +29,9 @@
_TIME_FORMAT = '%Y%m%d-%H%M%S.%f'
- # Paths from root proxy
- ALARM_PATH = '/omci_alarms'
- DEVICE_PATH = '{}' # .format(device_id)
-
- # Classes, Instances, and Attributes as lists from root proxy
- CLASS_PATH = DEVICE_PATH + '/classes/{}' # .format(device_id, class_id)
- INSTANCE_PATH = DEVICE_PATH + '/classes/{}/instances/{}' # .format(device_id, class_id, instance_id)
+ ALARM_PATH = 'service/voltha/omci_alarms'
+ DEVICE_PATH = '{}' # .format(device_id)
+ CLASS_PATH = DEVICE_PATH + '/classes/{}' # .format(device_id, class_id)
def __init__(self, omci_agent):
"""
@@ -45,131 +39,18 @@
:param omci_agent: (OpenOMCIAgent) OpenOMCI Agent
"""
super(AlarmDbExternal, self).__init__(omci_agent)
-
+ self._core = omci_agent.core_proxy
+ # Some statistics to help with debug/tuning/...
+ self._statistics = {
+ 'get': MibDbStatistic('get'),
+ 'set': MibDbStatistic('set'),
+ 'create': MibDbStatistic('create'),
+ 'delete': MibDbStatistic('delete')
+ }
self.args = registry('main').get_args()
host, port = self.args.etcd.split(':', 1)
self._kv_store = EtcdStore(host, port, AlarmDbExternal.ALARM_PATH)
- def start(self):
- """
- Start up/restore the database
- """
- self.log.debug('start')
-
- if not self._started:
- super(AlarmDbExternal, self).start()
-
- def stop(self):
- """
- Start up the database
- """
- self.log.debug('stop')
-
- if self._started:
- super(AlarmDbExternal, self).stop()
- # TODO: Delete this method if nothing else is done except calling the base class
-
- def _time_to_string(self, time):
- return time.strftime(AlarmDbExternal._TIME_FORMAT) if time is not None else ''
-
- def _string_to_time(self, time):
- return datetime.strptime(time, AlarmDbExternal._TIME_FORMAT) if len(time) else None
-
- def _attribute_to_string(self, value):
- """
- Convert an ME's attribute value to string representation
-
- :param value: (long) Alarm bitmaps are always a Long
- :return: (str) String representation of the value
- """
- return str(value)
-
- def _string_to_attribute(self, str_value):
- """
- Convert an ME's attribute value-string to its Scapy decode equivalent
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) Class ID
- :param attr_name: (str) Attribute Name (see EntityClasses)
- :param str_value: (str) Attribute Value in string form
-
- :return: (various) String representation of the value
- :raises KeyError: Device, Class ID, or Attribute does not exist
- """
- # Alarms are always a bitmap which is a long
- return int(str_value) if len(str_value) else 0
-
- def add(self, device_id, overwrite=False):
- """
- Add a new ONU to database
-
- :param device_id: (str) Device ID of ONU to add
- :param overwrite: (bool) Overwrite existing entry if found.
-
- :raises KeyError: If device already exists and 'overwrite' is False
- """
- self.log.debug('add-device', device_id=device_id, overwrite=overwrite)
-
- now = datetime.utcnow()
- found = False
-
- data = AlarmDeviceData(device_id=device_id,
- created=self._time_to_string(now),
- version=AlarmDbExternal.CURRENT_VERSION,
- last_alarm_sequence=0)
- path = self._get_device_path(device_id)
- self.log.debug("add-device-path", device_id=device_id, path=path)
- try:
- #raises KeyError if not found
- device = self._kv_store[path]
-
- #device is found at this point
- found = True
-
- if not overwrite:
- # Device already exists
- raise KeyError('Device with ID {} already exists in Alarm database'.
- format(device_id))
-
- # Overwrite with new data
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
-
- except KeyError:
- if found:
- raise
- # Did not exist, add it now
- self._kv_store[path] = data.SerializeToString()
- self._created = now
- self._modified = now
-
- def remove(self, device_id):
- """
- Remove an ONU from the database
-
- :param device_id: (str) Device ID of ONU to remove from database
- """
- self.log.debug('remove-device', device_id=device_id)
-
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID should be an string')
-
- try:
- path = self._get_device_path(device_id)
- del self._kv_store[path]
- self._modified = datetime.utcnow()
-
- except KeyError:
- # Did not exists, which is not a failure
- pass
-
- except Exception as e:
- self.log.exception('remove-exception', device_id=device_id, e=e)
- raise
-
def _get_device_path(self, device_id):
return AlarmDbExternal.DEVICE_PATH.format(device_id)
@@ -183,447 +64,40 @@
fmt = AlarmDbExternal.CLASS_PATH
return fmt.format(device_id, class_id)
+ def _time_to_string(self, time):
+ return time.strftime(AlarmDbExternal._TIME_FORMAT) if time is not None else ''
- def _get_instance_path(self, device_id, class_id, instance_id):
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
+ def _string_to_time(self, time):
+ return datetime.strptime(time, AlarmDbExternal._TIME_FORMAT) if len(time) else None
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID is a string')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError('class-id is 0..0xFFFF')
-
- if not 0 <= instance_id <= 0xFFFF:
- raise ValueError('instance-id is 0..0xFFFF')
-
- fmt = AlarmDbExternal.INSTANCE_PATH
- return fmt.format(device_id, class_id, instance_id)
-
- def save_last_sync_time(self, device_id, value):
+ def _attribute_to_string(self, device_id, class_id, attr_name, value, old_value=None):
"""
- Save the Last Sync time to the database in an easy location to access
+ Convert an ME's attribute value to string representation
+
+ :param value: (long) Alarm bitmaps are always a Long
+ :return: (str) String representation of the value
+ """
+ self.log.debug("attribute-to-string", device_id=device_id, class_id=class_id,
+ attr_name=attr_name, value=value)
+ return str(value)
+
+ def _string_to_attribute(self, device_id, class_id, attr_name, str_value):
+ """
+ Convert an Alarm ME's attribute value-string to its Scapy decode equivalent
:param device_id: (str) ONU Device ID
- :param value: (DateTime) Value to save
+ :param class_id: (int) Class ID
+ :param attr_name: (str) Attribute Name (at this point only alarm_bit_map)
+ :param str_value: (str) Attribute Value in string form
+
+ :return: (int) Long integer representation of the value
"""
- self.log.debug('save-last-sync-time', device_id=device_id, time=str(value))
+ # Alarms are always a bitmap which is a long
+ if attr_name == AlarmDbExternal.ALARM_BITMAP_KEY:
+ value = int(str_value) if len(str_value) else 0
+ else:
+ value = 0
- try:
- if not isinstance(value, datetime):
- raise TypeError('Expected a datetime object, got {}'.
- format(type(datetime)))
-
- data = AlarmDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[path])
-
- now = datetime.utcnow()
- data.last_sync_time = self._time_to_string(value)
-
- # Update
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
- self.log.debug('save-sync-time-complete', device_id=device_id)
-
- except Exception as e:
- self.log.exception('save-last-sync-exception', device_id=device_id, e=e)
- raise
-
- def get_last_sync_time(self, device_id):
- """
- Get the Last Sync Time saved to the database for a device
-
- :param device_id: (str) ONU Device ID
- :return: (int) The Value or None if not found
- """
- self.log.debug('get-last-sync-time', device_id=device_id)
-
- try:
- data = AlarmDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[path])
- return self._string_to_time(data.last_sync_time)
-
- except KeyError:
- return None # OMCI MIB_DB entry has not yet been created
-
- except Exception as e:
- self.log.exception('get-last-sync-time-exception', e=e)
- raise
-
- def save_alarm_last_sync(self, device_id, value):
- """
- Save the Last Alarm Sequence value to the database in an easy location to access
-
- :param device_id: (str) ONU Device ID
- :param value: (int) Value to save
- """
- self.log.debug('save-last-sync', device_id=device_id, seq=str(value))
-
- try:
- if not isinstance(value, int):
- raise TypeError('Expected a integer, got {}'.format(type(value)))
-
- data = AlarmDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[path])
-
- now = datetime.utcnow()
- data.last_alarm_sequence = int(value)
-
- # Update
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
- self.log.debug('save-sequence-complete', device_id=device_id)
-
- except Exception as e:
- self.log.exception('save-last-sync-exception', device_id=device_id, e=e)
- raise
-
- def get_alarm_last_sync(self, device_id):
- """
- Get the Last Sync Time saved to the database for a device
-
- :param device_id: (str) ONU Device ID
- :return: (int) The Value or None if not found
- """
- self.log.debug('get-last-sync', device_id=device_id)
-
- try:
- data = AlarmDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self.kv_store[path])
- return int(data.last_alarm_sequence)
-
- except KeyError:
- return None # OMCI ALARM_DB entry has not yet been created
-
- except Exception as e:
- self.log.exception('get-last-alarm-exception', e=e)
- raise
-
- def _add_new_class_and_instance(self, device_id, class_id, instance_id, attributes):
- """
- Create an entry for a new class in the external database
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param instance_id: (int) ME Entity ID
- :param attributes: (dict) Attribute dictionary
-
- :returns: (bool) True if the value was saved to the database. False if the
- value was identical to the current instance
- """
- self.log.debug('add-new-class-and-instance', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
-
- class_data = AlarmClassData(class_id=class_id)
-
- class_path = AlarmDbExternal.CLASS_PATH.format(device_id, class_id)
- self._kv_store[class_path] = class_data.SerializeToString()
-
- inst_data = self._create_new_instance(device_id, class_id, instance_id, attributes)
- inst_path = AlarmDbExternal.INSTANCE_PATH.format(device_id, class_id, instance_id)
- self._kv_store[inst_path] = inst_data.SerializeToString()
-
- self.log.debug('set-complete', device_id=device_id, class_id=class_id,
- entity_id=instance_id, attributes=attributes)
- return True
-
- def _create_new_instance(self, device_id, class_id, instance_id, attributes):
- """
- Create an entry for a instance of an existing class in the external database
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param instance_id: (int) ME Entity ID
- :param attributes: (dict) Attribute dictionary
-
- :returns: (AlarmInstanceData) The new instance object
- """
- self.log.debug('create-new-instance', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
-
- now = self._time_to_string(datetime.utcnow())
- attrs = [AlarmAttributeData(name=k,
- value=self._attribute_to_string(v)) for k, v in attributes.items()]
- instance_data = AlarmInstanceData(instance_id=instance_id,
- created=now,
- modified=now,
- attributes=attrs)
-
- self.log.debug('create-new-instance-complete', device_id=device_id, class_id=class_id,
- entity_id=instance_id, attributes=attributes)
- return instance_data
-
- def set(self, device_id, class_id, instance_id, attributes):
- """
- Set a database value. This should only be called by the Alarm synchronizer
- and its related tasks
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param instance_id: (int) ME Entity ID
- :param attributes: (dict) Attribute dictionary
-
- :returns: (bool) True if the value was saved to the database. False if the
- value was identical to the current instance
-
- :raises KeyError: If device does not exist
- :raises DatabaseStateError: If the database is not enabled
- """
- self.log.debug('set', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
- try:
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID should be a string')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError("Invalid Class ID: {}, should be 0..65535".format(class_id))
-
- if not 0 <= instance_id <= 0xFFFF:
- raise ValueError("Invalid Instance ID: {}, should be 0..65535".format(instance_id))
-
- if not isinstance(attributes, dict):
- raise TypeError("Attributes should be a dictionary")
-
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- try:
- class_data = AlarmClassData()
- class_path = AlarmDbExternal.CLASS_PATH.format(device_id, class_id)
- class_data.ParseFromString(self._kv_store[class_path])
-
- modified = False
- new_data = None
- try:
- inst_data = AlarmInstanceData()
- inst_path = AlarmDbExternal.INSTANCE_PATH.format(device_id, class_id, instance_id)
- inst_data.ParseFromString(self._kv_store[inst_path])
-
- exist_attr_indexes = dict()
- attr_len = len(inst_data.attributes)
-
- for index in range(0, attr_len):
- exist_attr_indexes[inst_data.attributes[index].name] = index
-
- str_value = ''
- new_attributes = []
-
- for k, v in attributes.items():
- try:
- str_value = self._attribute_to_string(v)
- new_attributes.append(AlarmAttributeData(name=k, value=str_value))
-
- except Exception as e:
- self.log.exception('save-error', e=e, class_id=class_id,
- attr=k, value_type=type(v))
-
- if k not in exist_attr_indexes or \
- inst_data.attributes[exist_attr_indexes[k]].value != str_value:
- modified = True
- self.log.debug('different-attributes', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=inst_data.attributes, new_attributes=new_attributes)
-
- if modified:
- now = datetime.utcnow()
- new_data = AlarmInstanceData(instance_id=instance_id,
- created=inst_data.created,
- modified=self._time_to_string(now),
- attributes=new_attributes)
-
- except KeyError:
- # Here if the instance_id does not yet exist in the database
- new_data = self._create_new_instance(device_id, class_id, instance_id, attributes)
- modified = True
-
- if modified:
- del self._kv_store[inst_path]
- self._kv_store[inst_path] = new_data.SerializeToString()
- return modified
-
- except KeyError:
- # Here if the class-id does not yet exist in the database
- return self._add_new_class_and_instance(device_id, class_id, instance_id,
- attributes)
- except Exception as e:
- self.log.exception('set-exception', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes, e=e)
- raise
-
- def delete(self, device_id, class_id, entity_id):
- """
- Delete an entity from the database if it exists. If all instances
- of a class are deleted, the class is deleted as well.
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param entity_id: (int) ME Entity ID
-
- :returns: (bool) True if the instance was found and deleted. False
- if it did not exist.
-
- :raises KeyError: If device does not exist
- :raises DatabaseStateError: If the database is not enabled
- """
- self.log.debug('delete', device_id=device_id, class_id=class_id,
- entity_id=entity_id)
-
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID should be an string')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError('class-id is 0..0xFFFF')
-
- if not 0 <= entity_id <= 0xFFFF:
- raise ValueError('instance-id is 0..0xFFFF')
-
- try:
- instpath = self._get_instance_path(device_id, class_id, entity_id)
- del self._kv_store[instpath]
-
- self._modified = datetime.utcnow()
- return True
-
- except KeyError:
- return False # Not found
-
- except Exception as e:
- self.log.exception('get-last-data-exception', device_id=device_id, e=e)
- raise
-
- def query(self, device_id, class_id=None, instance_id=None, attributes=None):
- """
- Get database information.
-
- This method can be used to request information from the database to the detailed
- level requested
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) Managed Entity class ID
- :param instance_id: (int) Managed Entity instance
- :param attributes: (list/set or str) Managed Entity instance's attributes
-
- :return: (dict) The value(s) requested. If class/inst/attribute is
- not found, an empty dictionary is returned
- :raises KeyError: If the requested device does not exist
- :raises DatabaseStateError: If the database is not enabled
- """
- self.log.debug('query', device_id=device_id, class_id=class_id,
- entity_instance_id=instance_id, attributes=attributes)
- data = dict()
-
- try:
- if class_id is None:
- # Get full device info
- self.log.warn('get-all-instances-by-device-not-supported')
- try:
- dev_data = AlarmDeviceData()
- device_path = self._get_device_path(device_id)
- dev_data.ParseFromString(self._kv_store[device_path])
- data = self._device_to_dict(dev_data)
-
- except KeyError:
- data = dict()
-
- elif instance_id is None:
- # Get all instances of the class
- self.log.warn('get-all-instances-by-class-not-supported')
- try:
- cls_data = AlarmClassData()
- class_path = self._get_class_path(device_id, class_id)
- cls_data.ParseFromString(self._kv_store[class_path])
- data = self._class_to_dict(cls_data)
-
- except KeyError:
- data = dict()
-
- else:
- # Get all attributes of a specific ME
- try:
- inst_data = AlarmInstanceData()
- inst_path = self._get_instance_path(device_id, class_id, instance_id)
- inst_data.ParseFromString(self._kv_store[inst_path])
-
- if attributes is None:
- # All Attributes
- data = self._instance_to_dict(inst_data)
- self.log.debug('query-result-all', data=data)
- else:
- # Specific attribute(s)
- if isinstance(attributes, six.string_types):
- attributes = {attributes}
-
- data = {
- attr.name: self._string_to_attribute(attr.value)
- for attr in inst_data.attributes if attr.name in attributes}
-
- except KeyError:
- self.log.debug('no-instance-data-keyError')
- data = dict()
-
- except Exception as e:
- self.log.exception('get-last-sync-exception', device_id=device_id, e=e)
- raise
-
- self.log.debug('query-result', data=data)
- return data
-
- except KeyError:
- self.log.warn('query-no-device', device_id=device_id)
- raise
-
- except Exception as e:
- self.log.exception('get-last-sync-exception', device_id=device_id, e=e)
- raise
-
- def _instance_to_dict(self, instance):
- if not isinstance(instance, AlarmInstanceData):
- raise TypeError('{} is not of type AlarmInstanceData'.format(type(instance)))
-
- data = {
- INSTANCE_ID_KEY: instance.instance_id,
- CREATED_KEY: self._string_to_time(instance.created),
- MODIFIED_KEY: self._string_to_time(instance.modified),
- ATTRIBUTES_KEY: dict()
- }
- for attribute in instance.attributes:
- data[ATTRIBUTES_KEY][attribute.name] = self._string_to_attribute(attribute.value)
- return data
-
- def _class_to_dict(self, val):
- if not isinstance(val, AlarmClassData):
- raise TypeError('{} is not of type AlarmClassData'.format(type(val)))
-
- data = {
- CLASS_ID_KEY: val.class_id,
- }
- for instance in val.instances:
- data[instance.instance_id] = self._instance_to_dict(instance)
- return data
-
- def _device_to_dict(self, val):
- if not isinstance(val, AlarmDeviceData):
- raise TypeError('{} is not of type AlarmDeviceData'.format(type(val)))
-
- data = {
- DEVICE_ID_KEY: val.device_id,
- CREATED_KEY: self._string_to_time(val.created),
- VERSION_KEY: val.version,
- ME_KEY: dict(),
- MSG_TYPE_KEY: set()
- }
- for class_data in val.classes:
- data[class_data.class_id] = self._class_to_dict(class_data)
- for managed_entity in val.managed_entities:
- data[ME_KEY][managed_entity.class_id] = managed_entity.name
-
- for msg_type in val.message_types:
- data[MSG_TYPE_KEY].add(msg_type.message_type)
-
- return data
+ self.log.debug("string-to-attribute", device_id=device_id, class_id=class_id,
+ attr_name=attr_name, value=value)
+ return value
diff --git a/pyvoltha/adapters/extensions/omci/database/mib_db_api.py b/pyvoltha/adapters/extensions/omci/database/mib_db_api.py
index 9aa414f..ad6a664 100644
--- a/pyvoltha/adapters/extensions/omci/database/mib_db_api.py
+++ b/pyvoltha/adapters/extensions/omci/database/mib_db_api.py
@@ -244,3 +244,21 @@
:param msg_types: (set) Message Type values (ints)
"""
raise NotImplementedError('Implement this in your derive class')
+
+ def load_from_template(self, device_id, template):
+ """
+ Load a device instance database from a dictionary
+
+ :param device_id: (str) ONU Device ID
+ :param template: (dict) Dictionary of the template read from storage
+ """
+ raise NotImplementedError('Implement this in your derive class')
+
+ def dump_to_json(self, device_id):
+ """
+ Return a JSON encoded string of the device instance data
+
+ :param device_id: (str) ONU Device ID
+ :return: (str) JSON encoded string representing the device instance
+ """
+ raise NotImplementedError('Implement this in your derive class')
diff --git a/pyvoltha/adapters/extensions/omci/database/mib_db_dict.py b/pyvoltha/adapters/extensions/omci/database/mib_db_dict.py
index e4066f6..080b849 100644
--- a/pyvoltha/adapters/extensions/omci/database/mib_db_dict.py
+++ b/pyvoltha/adapters/extensions/omci/database/mib_db_dict.py
@@ -553,7 +553,6 @@
def dump_to_json(self, device_id):
device_db = self._data.get(device_id, dict())
- device_db = self._fix_dev_json_attributes(device_db, device_id)
def json_converter(o):
if isinstance(o, datetime):
@@ -561,6 +560,6 @@
if isinstance(o, six.binary_type):
return o.decode('ascii')
- json_string = json.dumps(device_db, default=json_converter, indent=4)
+ json_string = json.dumps(device_db, default=json_converter, indent=2)
return json_string
diff --git a/pyvoltha/adapters/extensions/omci/database/mib_db_dict_lazy.py b/pyvoltha/adapters/extensions/omci/database/mib_db_dict_lazy.py
new file mode 100644
index 0000000..2fa3c51
--- /dev/null
+++ b/pyvoltha/adapters/extensions/omci/database/mib_db_dict_lazy.py
@@ -0,0 +1,144 @@
+#
+# Copyright 2020 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 __future__ import absolute_import
+import json
+from twisted.internet.task import LoopingCall
+from twisted.internet.defer import inlineCallbacks, returnValue
+from .mib_db_api import *
+from .mib_db_dict import MibDbVolatileDict
+from pyvoltha.adapters.common.kvstore.twisted_etcd_store import TwistedEtcdStore
+from pyvoltha.common.utils.registry import registry
+
+LAST_LAZY_WRITE_KEY = 'last_lazy_write'
+DIRTY_DB_KEY = 'dirty_db'
+LAZY_DEFERRED_KEY = 'lazy_deferred'
+PENDING_DELETE = 'pending_delete'
+CHECK_INTERVAL = 60
+
+
+class MibDbLazyWriteDict(MibDbVolatileDict):
+
+ # Paths from kv store
+ MIB_PATH = 'service/voltha/omci_mibs'
+ DEVICE_PATH = '{}' # .format(device_id)
+
+ def __init__(self, omci_agent):
+ super(MibDbLazyWriteDict, self).__init__(omci_agent)
+
+ self.args = registry('main').get_args()
+ host, port = self.args.etcd.split(':', 1)
+ self._kv_store = TwistedEtcdStore(host, port, MibDbLazyWriteDict.MIB_PATH)
+ self._lazymetadata = dict()
+
+ @inlineCallbacks
+ def add(self, device_id, overwrite=False):
+
+ existing_db = yield self._load_device_data(device_id)
+ if existing_db:
+ # populate device database if exists in etcd
+ self._data[device_id] = existing_db
+ now = datetime.utcnow()
+ self._lazymetadata[device_id] = {
+ LAST_LAZY_WRITE_KEY: now,
+ DIRTY_DB_KEY: False,
+ PENDING_DELETE: False
+ }
+ self.log.debug('recovered-device-from-storage', device_id=device_id, metadata=self._lazymetadata[device_id])
+ else:
+ # new device with no persistent storage.
+ self._lazymetadata[device_id] = {
+ LAST_LAZY_WRITE_KEY: None,
+ DIRTY_DB_KEY: True,
+ PENDING_DELETE: False
+ }
+ self.log.debug('add-device-for-lazy-sync', device_id=device_id, metadata=self._lazymetadata[device_id])
+
+ self._lazymetadata[device_id][LAZY_DEFERRED_KEY] = LoopingCall(self._check_dirty, device_id)
+ self._lazymetadata[device_id][LAZY_DEFERRED_KEY].start(CHECK_INTERVAL, now=False)
+
+ super(MibDbLazyWriteDict, self).add(device_id, overwrite)
+
+ def remove(self, device_id):
+ super(MibDbLazyWriteDict, self).remove(device_id)
+ self._lazymetadata[device_id][DIRTY_DB_KEY] = True
+ self._lazymetadata[device_id][PENDING_DELETE] = True
+ self.log.debug('setting-sync-remove-flag', device_id=device_id, metadata=self._lazymetadata[device_id])
+
+ def on_mib_reset(self, device_id):
+ super(MibDbLazyWriteDict, self).on_mib_reset(device_id)
+ self._lazymetadata[device_id][DIRTY_DB_KEY] = True
+
+ def save_mib_data_sync(self, device_id, value):
+ results = super(MibDbLazyWriteDict, self).save_mib_data_sync(device_id, value)
+ self._lazymetadata[device_id][DIRTY_DB_KEY] = True
+ return results
+
+ def _check_dirty(self, device_id):
+ if self._lazymetadata[device_id][DIRTY_DB_KEY] is True:
+ self.log.debug('dirty-cache-writing-data', device_id=device_id, metadata=self._lazymetadata[device_id])
+ self._sync(device_id)
+ else:
+ self.log.debug('clean-cache-checking-later', device_id=device_id, metadata=self._lazymetadata[device_id])
+
+ @inlineCallbacks
+ def _sync(self, device_id):
+ json = super(MibDbLazyWriteDict, self).dump_to_json(device_id)
+ now = datetime.utcnow()
+ device_path = self._get_device_path(device_id)
+
+ if self._lazymetadata[device_id][PENDING_DELETE] is True:
+ yield self._kv_store.delete(device_path)
+ self.log.debug('removed-synced-data', device_id=device_id, metadata=self._lazymetadata[device_id])
+ d = self._lazymetadata[device_id][LAZY_DEFERRED_KEY]
+ del self._lazymetadata[device_id]
+ d.stop()
+ else:
+ yield self._kv_store.set(device_path, json)
+ self._lazymetadata[device_id][LAST_LAZY_WRITE_KEY] = now
+ self._lazymetadata[device_id][DIRTY_DB_KEY] = False
+ self.log.debug('synced-data', device_id=device_id, metadata=self._lazymetadata[device_id])
+
+ @inlineCallbacks
+ def _load_device_data(self, device_id):
+ device_path = self._get_device_path(device_id)
+ json = yield self._kv_store.get(device_path)
+ if json:
+ lookupdb = self._load_from_json(json)
+ self.log.debug('looked-up-device', device_path=device_path)
+ returnValue(lookupdb)
+ else:
+ returnValue(None)
+
+ def _load_from_json(self, jsondata):
+
+ def json_obj_parser(x):
+ if isinstance(x, dict):
+ results = dict()
+ for (key, value) in x.items():
+ try:
+ key = int(key)
+ except (ValueError, TypeError):
+ pass
+
+ results.update({key: value})
+ return results
+ return x
+
+ device_data = json.loads(jsondata, object_hook=json_obj_parser)
+ return device_data
+
+ def _get_device_path(self, device_id):
+ return MibDbLazyWriteDict.DEVICE_PATH.format(device_id)
diff --git a/pyvoltha/adapters/extensions/omci/database/mib_db_ext.py b/pyvoltha/adapters/extensions/omci/database/mib_db_ext.py
index 375894b..bf7f2ed 100644
--- a/pyvoltha/adapters/extensions/omci/database/mib_db_ext.py
+++ b/pyvoltha/adapters/extensions/omci/database/mib_db_ext.py
@@ -19,13 +19,13 @@
MibDeviceData, MibAttributeData, MessageType, ManagedEntity
from pyvoltha.adapters.extensions.omci.omci_entities import *
from pyvoltha.adapters.extensions.omci.omci_fields import *
-from pyvoltha.common.config.config_backend import EtcdStore
-from scapy.fields import StrField, FieldListField, PacketField
+from pyvoltha.adapters.common.kvstore.etcd_store import EtcdStore
+from scapy.fields import StrField, FieldListField
from pyvoltha.common.utils.registry import registry
import six
-import codecs
from six.moves import range
+
class MibDbStatistic(object):
"""
For debug/tuning purposes.
@@ -49,10 +49,11 @@
o Gets: Avg: 3.3 mS, Min: 0 mS, Max: 8 mS ( 9 samples)
o Deletes: No samples
"""
+
def __init__(self, name):
self._name = name
self._count = 0
- self._total_time = 0 # Total milliseconds
+ self._total_time = 0 # Total milliseconds
self._min_time = 99999999
self._max_time = 0
@@ -68,13 +69,13 @@
def clear_statistics(self):
self._count = 0
- self._total_time = 0 # Total milliseconds
+ self._total_time = 0 # Total milliseconds
self._min_time = 99999999
self._max_time = 0
def increment(self, time):
self._count += 1
- self._total_time += time # Total milliseconds
+ self._total_time += time # Total milliseconds
if self._min_time > time:
self._min_time = time
if self._max_time < time:
@@ -85,23 +86,14 @@
"""
A persistent external OpenOMCI MIB Database
"""
- CURRENT_VERSION = 1 # VOLTHA v1.3.0 release
+ CURRENT_VERSION = 1
_TIME_FORMAT = '%Y%m%d-%H%M%S.%f'
- # Paths from root proxy
- MIB_PATH = '/omci_mibs'
- DEVICE_PATH = '{}' # .format(device_id)
-
- # Classes, Instances, and Attributes as lists from root proxy
- CLASS_PATH = DEVICE_PATH + '/classes/{}' # .format(device_id)
- INSTANCE_PATH = DEVICE_PATH + '/classes/{}/instances/{}' # .format(device_id, class_id)
- ATTRIBUTE_PATH = DEVICE_PATH + '/classes/{}/instances/{}/attributes/{}' # .format(device_id, class_id, instance_id)
-
- # Single Class, Instance, and Attribute as objects from device proxy
- #CLASS_PATH = '/classes/{}' # .format(class_id)
- #INSTANCE_PATH = '/classes/{}/instances/{}' # .format(class_id, instance_id)
- #ATTRIBUTE_PATH = '/classes/{}/instances/{}/attributes/{}' # .format(class_id, instance_id
+ # Paths from kv store
+ MIB_PATH = 'service/voltha/omci_mibs'
+ DEVICE_PATH = '{}' # .format(device_id)
+ CLASS_PATH = DEVICE_PATH + '/classes/{}' # .format(device_id, class_id)
def __init__(self, omci_agent):
"""
@@ -146,13 +138,713 @@
super(MibDbExternal, self).stop()
# TODO: Delete this method if 6nothing else is done except calling the base class
+ def add(self, device_id, overwrite=False):
+ """
+ Add a new ONU to database
+
+ :param device_id: (str) Device ID of ONU to add
+ :param overwrite: (bool) Overwrite existing entry if found.
+
+ :raises KeyError: If device already exists and 'overwrite' is False
+ """
+ self.log.debug('add-device', device_id=device_id, overwrite=overwrite)
+
+ now = datetime.utcnow()
+
+ path = self._get_device_path(device_id)
+ new_device_data = self._create_new_device(device_id)
+ self.log.debug('new_device', new_device_data=new_device_data, device_id=device_id, path=path)
+
+ try:
+ search_device = self._kv_store.get(path)
+ if search_device is None:
+ # device not found, add new
+ self._kv_store.set(path, new_device_data.SerializeToString())
+ self._created = now
+ self._modified = now
+ else:
+ # device is found
+ if not overwrite:
+ # Device already exists
+ raise KeyError('Device with ID {} already exists in MIB database'.
+ format(device_id))
+
+ self._kv_store.set(path, new_device_data.SerializeToString())
+ self._modified = now
+
+ except Exception as e:
+ self.log.exception('add-exception', device_id=device_id, e=e)
+ raise
+
+ def remove(self, device_id):
+ """
+ Remove an ONU from the database
+
+ :param device_id: (str) Device ID of ONU to remove from database
+ """
+ self.log.debug('remove-device', device_id=device_id)
+
+ if not self._started:
+ raise DatabaseStateError('The Database is not currently active')
+
+ if not isinstance(device_id, six.string_types):
+ raise TypeError('Device ID should be an string')
+
+ try:
+ path = self._get_device_path(device_id)
+ self._kv_store.delete(path)
+ self._modified = datetime.utcnow()
+
+ except Exception as e:
+ self.log.exception('remove-exception', device_id=device_id, e=e)
+ raise
+
+ def set(self, device_id, class_id, entity_id, attributes):
+ """
+ Set a database value. This should only be called by the MIB synchronizer
+ and its related tasks
+
+ :param device_id: (str) ONU Device ID
+ :param class_id: (int) ME Class ID
+ :param entity_id: (int) ME Entity ID
+ :param attributes: (dict) Attribute dictionary
+
+ :returns: (bool) True if the value was saved to the database. False if the
+ value was identical to the current instance
+
+ :raises KeyError: If device does not exist
+ :raises DatabaseStateError: If the database is not enabled
+ """
+ self.log.debug('set', device_id=device_id, class_id=class_id,
+ entity_id=entity_id, attributes=attributes)
+
+ operation = 'set'
+ start_time = datetime.utcnow()
+ try:
+ if not isinstance(device_id, six.string_types):
+ raise TypeError('Device ID should be a string')
+
+ if not 0 <= class_id <= 0xFFFF:
+ raise ValueError("Invalid Class ID: {}, should be 0..65535".format(class_id))
+
+ if not 0 <= entity_id <= 0xFFFF:
+ raise ValueError("Invalid Instance ID: {}, should be 0..65535".format(entity_id))
+
+ if not isinstance(attributes, dict):
+ raise TypeError("Attributes should be a dictionary")
+
+ if not self._started:
+ raise DatabaseStateError('The Database is not currently active')
+
+ class_path = self._get_class_path(device_id, class_id)
+ class_data = MibClassData()
+ query_data = self._kv_store.get(class_path)
+ if query_data is None:
+ # Here if the class-id does not yet exist in the database
+
+ # This is needed to create a "slimmed down" reference to the class in the device object.
+ # This would be used later if querying the entire device and needed to pull all the classes and instances
+ new_class_data_ptr = self._create_new_class(device_id, class_id)
+ dev_data = MibDeviceData()
+ device_path = self._get_device_path(device_id)
+
+ start_time = datetime.utcnow()
+ query_data = self._kv_store.get(device_path)
+ dev_data.ParseFromString(query_data)
+ dev_data.classes.extend([new_class_data_ptr])
+
+ self._kv_store.set(device_path, dev_data.SerializeToString())
+
+ # Create fully populated class/entity instance data in its own place in the KV store
+ new_class_data = self._create_new_class(device_id, class_id, entity_id,
+ attributes)
+
+ self._kv_store.set(class_path, new_class_data.SerializeToString())
+
+ return True
+ else:
+ # Here if the class-id exists in the database and we are updating instances or attributes
+ class_data.ParseFromString(query_data)
+
+ inst_data = next((inst for inst in class_data.instances
+ if inst.instance_id == entity_id), None)
+
+ modified = False
+ new_data = None
+ if inst_data is None:
+ # Creating a new instance
+ operation = 'create'
+ new_data = self._create_new_instance(device_id, class_id, entity_id, attributes)
+ modified = True
+ else:
+ # Possibly adding to or updating an existing instance
+ new_data = self._update_existing_instance(device_id, class_id, entity_id, attributes, inst_data)
+ if new_data is not None:
+ modified = True
+
+ if modified:
+ inst_index = next((index for index in range(len(class_data.instances)) if
+ class_data.instances[index].instance_id == entity_id), None)
+ # Delete the old instance
+ if inst_index is not None:
+ del class_data.instances[inst_index]
+
+ # Add the new/updated instance
+ class_data.instances.extend([new_data])
+ self._kv_store.set(class_path, class_data.SerializeToString())
+
+ return modified
+
+ except Exception as e:
+ self.log.exception('set-exception', device_id=device_id, class_id=class_id,
+ entity_id=entity_id, attributes=attributes, e=e)
+ raise
+
+ finally:
+ if start_time is not None:
+ diff = datetime.utcnow() - start_time
+ # NOTE: Change to 'debug' when checked in, manually change to 'info'
+ # for development testing.
+ self.log.debug('db-{}-time'.format(operation), milliseconds=diff.microseconds / 1000)
+ self._statistics[operation].increment(diff.microseconds / 1000)
+
+ def delete(self, device_id, class_id, entity_id):
+ """
+ Delete an entity from the database if it exists. If all instances
+ of a class are deleted, the class is deleted as well.
+
+ :param device_id: (str) ONU Device ID
+ :param class_id: (int) ME Class ID
+ :param entity_id: (int) ME Entity ID
+
+ :returns: (bool) True if the instance was found and deleted. False
+ if it did not exist.
+
+ :raises KeyError: If device does not exist
+ :raises DatabaseStateError: If the database is not enabled
+ """
+ self.log.debug('delete', device_id=device_id, class_id=class_id,
+ entity_id=entity_id)
+
+ if not self._started:
+ raise DatabaseStateError('The Database is not currently active')
+
+ if not isinstance(device_id, six.string_types):
+ raise TypeError('Device ID should be an string')
+
+ if not 0 <= class_id <= 0xFFFF:
+ raise ValueError('class-id is 0..0xFFFF')
+
+ if not 0 <= entity_id <= 0xFFFF:
+ raise ValueError('instance-id is 0..0xFFFF')
+
+ start_time = datetime.utcnow()
+ try:
+ now = datetime.utcnow()
+ class_path = self._get_class_path(device_id, class_id)
+ class_data = MibClassData()
+ query_data = self._kv_store.get(class_path)
+ if query_data is not None:
+ class_data.ParseFromString(query_data)
+
+ inst_index = next((index for index in range(len(class_data.instances)) if
+ class_data.instances[index].instance_id == entity_id), None)
+
+ # Remove instance
+ if inst_index is not None:
+ del class_data.instances[inst_index]
+ self._kv_store.set(class_path, class_data.SerializeToString())
+
+ # If resulting class has no instance, remove it as well
+ if len(class_data.instances) == 0:
+ self._kv_store.delete(class_path)
+
+ # Clean up Device class pointer
+ dev_data = MibDeviceData()
+ device_path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(device_path)
+ dev_data.ParseFromString(query_data)
+
+ class_index = next((index for index in range(len(dev_data.classes)) if
+ dev_data.classes[index].class_id == class_id), None)
+
+ if class_index is not None:
+ del dev_data.classes[class_index]
+ self._kv_store.set(device_path, dev_data.SerializeToString())
+
+ self._modified = now
+ return True
+ else:
+ self.log.warn('delete-key-not-found', device_id=device_id, class_id=class_id, entity_id=entity_id)
+ return False # Not found
+
+ except Exception as e:
+ self.log.exception('delete-exception', device_id=device_id, class_id=class_id, entity_id=entity_id, e=e)
+ raise
+
+ finally:
+ diff = datetime.utcnow() - start_time
+ # NOTE: Change to 'debug' when checked in, manually change to 'info'
+ # for development testing.
+ self.log.debug('db-delete-time', milliseconds=diff.microseconds / 1000)
+ self._statistics['delete'].increment(diff.microseconds / 1000)
+
+ def query(self, device_id, class_id=None, instance_id=None, attributes=None):
+ """
+ Get database information.
+
+ This method can be used to request information from the database to the detailed
+ level requested
+
+ :param device_id: (str) ONU Device ID
+ :param class_id: (int) Managed Entity class ID
+ :param instance_id: (int) Managed Entity instance
+ :param attributes: (list/set or str) Managed Entity instance's attributes
+
+ :return: (dict) The value(s) requested. If class/inst/attribute is
+ not found, an empty dictionary is returned
+ :raises KeyError: If the requested device does not exist
+ :raises DatabaseStateError: If the database is not enabled
+ """
+ self.log.debug('query', device_id=device_id, class_id=class_id,
+ entity_id=instance_id, attributes=attributes)
+
+ start_time = datetime.utcnow()
+ end_time = None
+ try:
+ if class_id is None:
+ # Get full device info. This is painful given the recursive lookups involved!
+ dev_data = MibDeviceData()
+ device_path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(device_path)
+ if query_data is not None:
+ dev_data.ParseFromString(query_data)
+
+ class_ids = [c.class_id for c in dev_data.classes]
+
+ class_data_dict = dict()
+ if len(class_ids):
+ for class_id in class_ids:
+ # Recursively call query with the class_id passed, so below can do what it already does
+ class_data_dict[class_id] = self.query(device_id, class_id)
+
+ end_time = datetime.utcnow()
+ data = self._device_to_dict(dev_data, class_data_dict)
+ else:
+ self.log.debug('query-no-device', device_id=device_id)
+ data = dict()
+
+ elif instance_id is None:
+ # Get all instances of the class
+ class_data = MibClassData()
+ class_path = self._get_class_path(device_id, class_id)
+ query_data = self._kv_store.get(class_path)
+ if query_data is not None:
+ class_data.ParseFromString(query_data)
+ end_time = datetime.utcnow()
+ data = self._class_to_dict(device_id, class_data)
+ else:
+ self.log.debug('query-no-class', device_id=device_id, class_id=class_id)
+ data = dict()
+ else:
+ # Get all attributes of a specific ME
+ class_data = MibClassData()
+ instance_data = None
+ class_path = self._get_class_path(device_id, class_id)
+ query_data = self._kv_store.get(class_path)
+ if query_data is not None:
+ class_data.ParseFromString(query_data)
+ end_time = datetime.utcnow()
+
+ for inst in class_data.instances:
+ if inst.instance_id == instance_id:
+ instance_data = inst
+
+ if instance_data is not None:
+ if attributes is None:
+ # All Attributes
+ data = self._instance_to_dict(device_id, class_id, instance_data)
+
+ else:
+ # Specific attribute(s)
+ if isinstance(attributes, six.string_types):
+ attributes = {attributes}
+
+ data = {
+ attr.name: self._string_to_attribute(device_id,
+ class_id,
+ attr.name,
+ attr.value)
+ for attr in instance_data.attributes if attr.name in attributes}
+ else:
+ self.log.debug('query-no-instance', device_id=device_id, class_id=class_id, entity_id=instance_id)
+ data = dict()
+
+ else:
+ self.log.debug('query-no-class', device_id=device_id, class_id=class_id)
+ data = dict()
+
+ return data
+
+ except Exception as e:
+ self.log.exception('query-exception', device_id=device_id, e=e)
+ raise
+
+ finally:
+ if end_time is not None:
+ diff = end_time.utcnow() - start_time
+ # NOTE: Change to 'debug' when checked in, manually change to 'info'
+ # for development testing.
+ self.log.debug('db-get-time', milliseconds=diff.microseconds / 1000, class_id=class_id,
+ entity_id=instance_id)
+ self._statistics['get'].increment(diff.microseconds / 1000)
+
+ def on_mib_reset(self, device_id):
+ """
+ Reset/clear the database for a specific Device
+
+ :param device_id: (str) ONU Device ID
+ :raises DatabaseStateError: If the database is not enabled
+ :raises KeyError: If the device does not exist in the database
+ """
+ self.log.debug('on-mib-reset', device_id=device_id)
+
+ data = MibDeviceData()
+
+ try:
+ path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(path)
+ if query_data is not None:
+ data.ParseFromString(query_data)
+
+ # data = MibDeviceData(Wipe out any existing class IDs
+ class_ids = [c.class_id for c in data.classes]
+
+ if len(class_ids):
+ for class_id in class_ids:
+ class_path = self._get_class_path(device_id, class_id)
+ # Delete detailed classes and instances
+ self._kv_store.delete(class_path)
+
+ # Reset MIB Data Sync to zero
+ now = datetime.utcnow()
+ new_data = MibDeviceData(device_id=device_id,
+ created=data.created,
+ last_sync_time=data.last_sync_time,
+ mib_data_sync=0,
+ version=MibDbExternal.CURRENT_VERSION)
+
+ # Update with blanked out device object
+ self._kv_store.set(path, new_data.SerializeToString())
+ self._modified = now
+ self.log.debug('mib-reset-complete', device_id=device_id)
+ else:
+ self.log.warn("mib-reset-no-data-to-reset", device_id=device_id)
+
+ except Exception as e:
+ self.log.exception('mib-reset-exception', device_id=device_id, e=e)
+ raise
+
+ def save_mib_data_sync(self, device_id, value):
+ """
+ Save the MIB Data Sync to the database in an easy location to access
+
+ :param device_id: (str) ONU Device ID
+ :param value: (int) Value to save
+ """
+ self.log.debug('save-mds', device_id=device_id, value=value)
+
+ try:
+ if not isinstance(value, int):
+ raise TypeError('MIB Data Sync is an integer')
+
+ if not 0 <= value <= 255:
+ raise ValueError('Invalid MIB-data-sync value {}. Must be 0..255'.
+ format(value))
+ data = MibDeviceData()
+ path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(path)
+ data.ParseFromString(query_data)
+
+ now = datetime.utcnow()
+ data.mib_data_sync = value
+
+ # Update
+ self._kv_store.set(path,data.SerializeToString())
+ self._modified = now
+ self.log.debug('save-mds-complete', device_id=device_id)
+
+ except Exception as e:
+ self.log.exception('save-mds-exception', device_id=device_id, e=e)
+ raise
+
+ def get_mib_data_sync(self, device_id):
+ """
+ Get the MIB Data Sync value last saved to the database for a device
+
+ :param device_id: (str) ONU Device ID
+ :return: (int) The Value or None if not found
+ """
+ self.log.debug('get-mds', device_id=device_id)
+
+ try:
+ data = MibDeviceData()
+ path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(path)
+ if query_data is not None:
+ data.ParseFromString(query_data)
+ return int(data.mib_data_sync)
+ else:
+ self.log.warn("mib-mds-no-data", device_id=device_id)
+ return None # OMCI MIB_DB entry has not yet been created
+
+ except Exception as e:
+ self.log.exception('get-mds-exception', device_id=device_id, e=e)
+ raise
+
+ def save_last_sync(self, device_id, value):
+ """
+ Save the Last Sync time to the database in an easy location to access
+
+ :param device_id: (str) ONU Device ID
+ :param value: (DateTime) Value to save
+ """
+ self.log.debug('save-last-sync', device_id=device_id, time=str(value))
+
+ try:
+ if not isinstance(value, datetime):
+ raise TypeError('Expected a datetime object, got {}'.
+ format(type(datetime)))
+ data = MibDeviceData()
+ path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(path)
+ data.ParseFromString(query_data)
+
+ now = datetime.utcnow()
+ data.last_sync_time = self._time_to_string(value)
+
+ # Update
+ self._kv_store.set(path, data.SerializeToString())
+ self._modified = now
+ self.log.debug('save-mds-complete', device_id=device_id)
+
+ except Exception as e:
+ self.log.exception('save-last-sync-exception', device_id=device_id, e=e)
+ raise
+
+ def get_last_sync(self, device_id):
+ """
+ Get the Last Sync Time saved to the database for a device
+
+ :param device_id: (str) ONU Device ID
+ :return: (int) The Value or None if not found
+ """
+ self.log.debug('get-last-sync', device_id=device_id)
+
+ try:
+ data = MibDeviceData()
+ path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(path)
+ if query_data is not None:
+ data.ParseFromString(query_data)
+ return self._string_to_time(data.last_sync_time)
+ else:
+ self.log.warn("mib-last-sync-no-data", device_id=device_id)
+ return None # OMCI MIB_DB entry has not yet been created
+
+ except Exception as e:
+ self.log.exception('get-last-sync-exception', e=e)
+ raise
+
+ def update_supported_managed_entities(self, device_id, managed_entities):
+ """
+ Update the supported OMCI Managed Entities for this device
+ :param device_id: (str) ONU Device ID
+ :param managed_entities: (set) Managed Entity class IDs
+ """
+ try:
+ me_list = [ManagedEntity(class_id=class_id,
+ name=self._managed_entity_to_name(device_id,
+ class_id))
+ for class_id in managed_entities]
+ data = MibDeviceData()
+ device_path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(device_path)
+ data.ParseFromString(query_data)
+
+ now = datetime.utcnow()
+ data.managed_entities.extend(me_list)
+
+ # Update
+ self._kv_store.set(device_path, data.SerializeToString())
+ self._modified = now
+ self.log.debug('save-me-list-complete', device_id=device_id)
+
+ except Exception as e:
+ self.log.exception('add-me-failure', e=e, me_list=managed_entities)
+ raise
+
+ def update_supported_message_types(self, device_id, msg_types):
+ """
+ Update the supported OMCI Managed Entities for this device
+ :param device_id: (str) ONU Device ID
+ :param msg_types: (set) Message Type values (ints)
+ """
+ try:
+ now = datetime.utcnow()
+ msg_type_list = [MessageType(message_type=msg_type.value)
+ for msg_type in msg_types]
+ data = MibDeviceData()
+ device_path = self._get_device_path(device_id)
+ query_data = self._kv_store.get(device_path)
+ data.ParseFromString(query_data)
+ data.message_types.extend(msg_type_list)
+
+ # Update
+ self._kv_store.set(device_path, data.SerializeToString())
+ self._modified = now
+ self.log.debug('save-msg-types-complete', device_id=device_id)
+
+ except Exception as e:
+ self.log.exception('add-msg-types-failure', e=e, msg_types=msg_types)
+ raise
+
+ # Private Helper Functions
+
+ def _get_device_path(self, device_id):
+ return MibDbExternal.DEVICE_PATH.format(device_id)
+
+ def _get_class_path(self, device_id, class_id):
+ if not self._started:
+ raise DatabaseStateError('The Database is not currently active')
+
+ if not 0 <= class_id <= 0xFFFF:
+ raise ValueError('class-id is 0..0xFFFF')
+
+ fmt = MibDbExternal.CLASS_PATH
+ return fmt.format(device_id, class_id)
+
+ def _create_new_device(self, device_id):
+ """
+ Create an entry for new device object returning device proto object
+
+ :param device_id: (str) ONU Device ID
+
+ :returns: (MibDeviceData) The new populated device object
+ """
+ now = self._time_to_string(datetime.utcnow())
+ device_data = MibDeviceData(device_id=device_id,
+ created=now,
+ last_sync_time='',
+ mib_data_sync=0,
+ version=MibDbExternal.CURRENT_VERSION)
+
+ return device_data
+
+ def _create_new_class(self, device_id, class_id, entity_id=None, attributes=None):
+ """
+ Create an entry for a new class optionally with its first instance returning class proto object
+
+ :param device_id: (str) ONU Device ID
+ :param class_id: (int) ME Class ID
+ :param entity_id: (int) ME Entity ID
+ :param attributes: (dict) Attribute dictionary
+
+ :returns: (MibClassData) The new populated class data object
+ """
+ class_data = None
+ if entity_id is not None:
+ instance_data = self._create_new_instance(device_id, class_id, entity_id, attributes)
+ class_data = MibClassData(class_id=class_id,
+ instances=[instance_data])
+ else:
+ class_data = MibClassData(class_id=class_id)
+
+ return class_data
+
+ def _create_new_instance(self, device_id, class_id, entity_id, attributes):
+ """
+ Create an entry for a instance of a class and returning instance proto object
+
+ :param device_id: (str) ONU Device ID
+ :param class_id: (int) ME Class ID
+ :param entity_id: (int) ME Entity ID
+ :param attributes: (dict) Attribute dictionary
+
+ :returns: (MibInstanceData) The new populated instance object
+ """
+ attrs = []
+ for k, v in attributes.items():
+ str_value = self._attribute_to_string(device_id, class_id, k, v)
+ attrs.append(MibAttributeData(name=k, value=str_value))
+
+ now = self._time_to_string(datetime.utcnow())
+ instance_data = MibInstanceData(instance_id=entity_id,
+ created=now,
+ modified=now,
+ attributes=attrs)
+
+ return instance_data
+
+ def _update_existing_instance(self, device_id, class_id, entity_id, attributes, existing_instance):
+ """
+ Update the attributes of an existing instance of a class and returning the modified instance proto object
+
+ :param device_id: (str) ONU Device ID
+ :param class_id: (int) ME Class ID
+ :param entity_id: (int) ME Entity ID
+ :param existing_instance: (MibInstanceData) current instance object
+ :param attributes: (dict) Attribute dictionary
+
+ :returns: (MibInstanceData) The updated instance object or None if nothing changed
+ """
+ new_attributes = []
+ exist_attr_indexes = dict()
+ attr_len = len(existing_instance.attributes)
+
+ modified = False
+ for index in range(0, attr_len):
+ name = existing_instance.attributes[index].name
+ value = existing_instance.attributes[index].value
+ exist_attr_indexes[name] = index
+ new_attributes.append(MibAttributeData(name=name, value=value))
+
+ for k, v in attributes.items():
+ try:
+ old_value = None if k not in exist_attr_indexes \
+ else new_attributes[exist_attr_indexes[k]].value
+
+ str_value = self._attribute_to_string(device_id, class_id, k, v, old_value)
+
+ if k not in exist_attr_indexes:
+ new_attributes.append(MibAttributeData(name=k, value=str_value))
+ modified = True
+
+ elif new_attributes[exist_attr_indexes[k]].value != str_value:
+ new_attributes[exist_attr_indexes[k]].value = str_value
+ modified = True
+
+ except Exception as e:
+ self.log.exception('save-error', e=e, class_id=class_id,
+ attr=k, value_type=type(v))
+
+ if modified:
+ now = self._time_to_string(datetime.utcnow())
+ new_instance_data = MibInstanceData(instance_id=entity_id,
+ created=existing_instance.created,
+ modified=now,
+ attributes=new_attributes)
+ return new_instance_data
+ else:
+ return None
+
def _time_to_string(self, time):
return time.strftime(MibDbExternal._TIME_FORMAT) if time is not None else ''
def _string_to_time(self, time):
return datetime.strptime(time, MibDbExternal._TIME_FORMAT) if len(time) else None
- def _attribute_to_string(self, device_id, class_id, attr_name, value, old_value = None):
+ def _attribute_to_string(self, device_id, class_id, attr_name, value, old_value=None):
"""
Convert an ME's attribute value to string representation
@@ -184,12 +876,16 @@
str_value = value.to_json()
elif isinstance(field.default, Packet_metaclass) \
and hasattr(field.default, 'json_from_value'):
- #and not isinstance(value, basestring):
# Value/hex of Packet Class to string
str_value = field.default.json_from_value(value)
else:
str_value = str(value)
+ elif isinstance(field, OmciSerialNumberField):
+ # For some reason some ONU encode quotes in the serial number...
+ other_value = value.replace("'", "")
+ str_value = str(other_value)
+
elif isinstance(field, (StrField, MACField, IPField)):
# For StrField, value is an str already
# For MACField, value is a string in ':' delimited form
@@ -258,6 +954,9 @@
else:
value = str_value
+ elif isinstance(field, OmciSerialNumberField):
+ value = str_value
+
elif isinstance(field, MACField):
value = str_value
@@ -291,688 +990,6 @@
value=str_value, e=e)
raise
- def add(self, device_id, overwrite=False):
- """
- Add a new ONU to database
-
- :param device_id: (str) Device ID of ONU to add
- :param overwrite: (bool) Overwrite existing entry if found.
-
- :raises KeyError: If device already exists and 'overwrite' is False
- """
- self.log.debug('add-device', device_id=device_id, overwrite=overwrite)
-
- now = datetime.utcnow()
- found = False
-
- data = MibDeviceData(device_id=device_id,
- created=self._time_to_string(now),
- last_sync_time='',
- mib_data_sync=0,
- version=MibDbExternal.CURRENT_VERSION)
- path = self._get_device_path(device_id)
- self.log.debug("add-device-path", device_id=device_id, path=path)
- try:
- #raises KeyError if not found
- device = self._kv_store[path]
-
- #device is found at this point
- found = True
- if not overwrite:
- # Device already exists
- raise KeyError('Device with ID {} already exists in MIB database'.
- format(device_id))
-
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
-
- except KeyError:
- #KeyError after finding the device should be raised
- if found:
- raise
- # Did not exist, add it now
- self._kv_store[path] = data.SerializeToString()
- self._created = now
- self._modified = now
-
- def remove(self, device_id):
- """
- Remove an ONU from the database
-
- :param device_id: (str) Device ID of ONU to remove from database
- """
- self.log.debug('remove-device', device_id=device_id)
-
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID should be an string')
-
- try:
- path = self._get_device_path(device_id)
- del self._kv_store[path]
- self._modified = datetime.utcnow()
-
- except KeyError:
- # Did not exists, which is not a failure
- pass
-
- except Exception as e:
- self.log.exception('remove-exception', device_id=device_id, e=e)
- raise
-
- def _get_device_path(self, device_id):
- return MibDbExternal.DEVICE_PATH.format(device_id)
-
- def _get_class_path(self, device_id, class_id):
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError('class-id is 0..0xFFFF')
-
- fmt = MibDbExternal.CLASS_PATH
- return fmt.format(device_id, class_id)
-
-
- def _get_class(self, device_id, class_id, create=False):
- """
- Get a config proxy to a specific managed entity class
- :param device_id: (str) ONU Device ID
- :param class_id: (int) Class ID
- :param create: (bool) If true, create default instance (and class)
- :return: (ConfigProxy) Class configuration proxy
-
- :raises DatabaseStateError: If database is not started
- :raises KeyError: If Instance does not exist and 'create' is False
- """
-
- path = self._get_class_path(device_id, class_id)
-
- try:
- return self._kv_store[path]
-
- except KeyError:
- if not create:
- # This can occur right after a MIB Reset if the ONU publishes AVCs right away
- # and during the MIB audit resync for ONU created MEs in response to an OLT
- # created ME. Fail since for these test cases they occur during a verification
- # 'query' and not the ME creation during resync. Calling code should handle
- # they exception if it is expected to occur on occasion.
- self.log.debug('class-proxy-does-not-exist', device_id=device_id,
- class_id=class_id)
- raise
-
- # Create class
- data = MibClassData(class_id=class_id)
- root_path = MibDbExternal.CLASS_PATH.format(device_id, class_id)
- self._kv_store[root_path] = data
-
- return data
-
- def _get_instance_path(self, device_id, class_id, instance_id):
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID is a string')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError('class-id is 0..0xFFFF')
-
- if not 0 <= instance_id <= 0xFFFF:
- raise ValueError('instance-id is 0..0xFFFF')
-
- fmt = MibDbExternal.INSTANCE_PATH
- return fmt.format(device_id, class_id, instance_id)
-
- def _get_instance(self, device_id, class_id, instance_id, create=False):
- """
- Get a config proxy to a specific managed entity instance
- :param device_id: (str) ONU Device ID
- :param class_id: (int) Class ID
- :param instance_id: (int) Instance ID
- :param create: (bool) If true, create default instance (and class)
- :return: (ConfigProxy) Instance configuration proxy
-
- :raises DatabaseStateError: If database is not started
- :raises KeyError: If Instance does not exist and 'create' is False
- """
-
- path = self._get_instance_path(device_id, class_id, instance_id)
-
- try:
- return self._kv_store[path]
-
- except KeyError:
- if not create:
- # This can occur right after a MIB Reset if the ONU publishes AVCs right away
- # and during the MIB audit resync for ONU created MEs in response to an OLT
- # created ME. Fail since for these test cases they occur during a verification
- # 'query' and not the ME creation during resync. Calling code should handle
- # they exception if it is expected to occur on occasion.
- self.log.info('instance-proxy-does-not-exist', device_id=device_id,
- class_id=class_id, instance_id=instance_id)
- raise
-
- # Create instance, first make sure class exists
- self._get_class(device_id, class_id, create=True)
-
- now = self._time_to_string(datetime.utcnow())
- data = MibInstanceData(instance_id=instance_id, created=now, modified=now)
- root_path = MibDbExternal.INSTANCE_PATH.format(device_id, class_id)
- self._kv_store[root_path] = data
-
- return data
-
- def on_mib_reset(self, device_id):
- """
- Reset/clear the database for a specific Device
-
- :param device_id: (str) ONU Device ID
- :raises DatabaseStateError: If the database is not enabled
- :raises KeyError: If the device does not exist in the database
- """
- self.log.debug('on-mib-reset', device_id=device_id)
-
- data = MibDeviceData()
-
- try:
- path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[path])
-
- # data = MibDeviceData(Wipe out any existing class IDs
- class_ids = [c.class_id for c in data.classes]
-
- if len(class_ids):
- for class_id in class_ids:
- classpath = MibDbExternal.CLASS_PATH.format(device_id, class_id)
- del self._kv_store[classpath]
-
- # Reset MIB Data Sync to zero
- now = datetime.utcnow()
- data = MibDeviceData(device_id=device_id,
- created=data.created,
- last_sync_time=data.last_sync_time,
- mib_data_sync=0,
- version=MibDbExternal.CURRENT_VERSION)
- # Update
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
- self.log.debug('mib-reset-complete', device_id=device_id)
-
- except Exception as e:
- self.log.exception('mib-reset-exception', device_id=device_id, e=e)
- raise
- except KeyError as e:
- self.log.debug("mib-reset-no-data-to-reset", device_id=device_id)
-
- def save_mib_data_sync(self, device_id, value):
- """
- Save the MIB Data Sync to the database in an easy location to access
-
- :param device_id: (str) ONU Device ID
- :param value: (int) Value to save
- """
- self.log.debug('save-mds', device_id=device_id, value=value)
-
- try:
- if not isinstance(value, int):
- raise TypeError('MIB Data Sync is an integer')
-
- if not 0 <= value <= 255:
- raise ValueError('Invalid MIB-data-sync value {}. Must be 0..255'.
- format(value))
- data = MibDeviceData()
- path = self._get_device_path(device_id)
-
- data.ParseFromString(self._kv_store[path])
-
- now = datetime.utcnow()
- data.mib_data_sync = value
-
- # Update
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
- self.log.debug('save-mds-complete', device_id=device_id)
-
- except Exception as e:
- self.log.exception('save-mds-exception', device_id=device_id, e=e)
- raise
-
- def get_mib_data_sync(self, device_id):
- """
- Get the MIB Data Sync value last saved to the database for a device
-
- :param device_id: (str) ONU Device ID
- :return: (int) The Value or None if not found
- """
- self.log.debug('get-mds', device_id=device_id)
-
- try:
- data = MibDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[path])
- return int(data.mib_data_sync)
-
- except KeyError:
- return None # OMCI MIB_DB entry has not yet been created
-
- except Exception as e:
- self.log.exception('get-mds-exception', device_id=device_id, e=e)
- raise
-
- def save_last_sync(self, device_id, value):
- """
- Save the Last Sync time to the database in an easy location to access
-
- :param device_id: (str) ONU Device ID
- :param value: (DateTime) Value to save
- """
- self.log.debug('save-last-sync', device_id=device_id, time=str(value))
-
- try:
- if not isinstance(value, datetime):
- raise TypeError('Expected a datetime object, got {}'.
- format(type(datetime)))
- data = MibDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[path])
-
- now = datetime.utcnow()
- data.last_sync_time = self._time_to_string(value)
-
- # Update
- self._kv_store[path] = data.SerializeToString()
- self._modified = now
- self.log.debug('save-mds-complete', device_id=device_id)
-
- except Exception as e:
- self.log.exception('save-last-sync-exception', device_id=device_id, e=e)
- raise
-
- def get_last_sync(self, device_id):
- """
- Get the Last Sync Time saved to the database for a device
-
- :param device_id: (str) ONU Device ID
- :return: (int) The Value or None if not found
- """
- self.log.debug('get-last-sync', device_id=device_id)
-
- try:
- data = MibDeviceData()
- path = self._get_device_path(device_id)
- data.ParseFromString(self.kv_store[path])
- return self._string_to_time(data.last_sync_time)
-
- except KeyError:
- return None # OMCI MIB_DB entry has not yet been created
-
- except Exception as e:
- self.log.exception('get-last-sync-exception', e=e)
- raise
-
- def _add_new_class(self, device_id, class_id, instance_id, attributes):
- """
- Create an entry for a new class in the external database
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param instance_id: (int) ME Entity ID
- :param attributes: (dict) Attribute dictionary
-
- :returns: (bool) True if the value was saved to the database. False if the
- value was identical to the current instance
- """
- self.log.debug('add-new-class', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
-
- now = self._time_to_string(datetime.utcnow())
- attrs = []
- for k, v in attributes.items():
- if k == 'serial_number':
- vendor_id = str(v[0:4])
- vendor_specific_ext = v[4:]
- vendor_specific = codecs.encode(vendor_specific_ext, 'hex')
- str_value = vendor_id + vendor_specific
- attrs.append(MibAttributeData(name=k, value=str_value))
- else:
- str_value = self._attribute_to_string(device_id, class_id, k, v)
- attrs.append(MibAttributeData(name=k, value=str_value))
-
- class_data = MibClassData(class_id=class_id,
- instances=[MibInstanceData(instance_id=instance_id,
- created=now,
- modified=now,
- attributes=attrs)])
-
- classpath = MibDbExternal.CLASS_PATH.format(device_id, class_id)
- self._kv_store[classpath] = class_data.SerializeToString()
- self.log.debug('set-complete', device_id=device_id, class_id=class_id,
- entity_id=instance_id, attributes=attributes)
- return True
-
- def _create_new_instance(self, device_id, class_id, instance_id, attributes):
- """
- Create an entry for a instance of an existing class in the external database
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param instance_id: (int) ME Entity ID
- :param attributes: (dict) Attribute dictionary
-
- :returns: (MibInstanceDate) The new instance object
- """
- self.log.debug('create-new-instance', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
-
- now = self._time_to_string(datetime.utcnow())
- attrs = []
- for k, v in attributes.items():
- if k == 'serial_number':
- vendor_id = str(v[0:4])
- vendor_specific_ext = v[4:]
- vendor_specific = codecs.encode(vendor_specific_ext, 'hex')
- str_value = vendor_id+vendor_specific
- attrs.append(MibAttributeData(name=k, value=str_value))
- else:
- str_value = self._attribute_to_string(device_id, class_id, k, v)
- attrs.append(MibAttributeData(name=k, value=str_value))
-
- instance_data = MibInstanceData(instance_id=instance_id,
- created=now,
- modified=now,
- attributes=attrs)
-
- #update class with instance
- #class_path = self._get_class_path(device_id, class_id)
- #class_data = MibClassData()
- #class_data.ParseFromString(self._kv_store[class_path])
-
- #class_data.instances.extend([instance_data])
-
- #self._kv_store[class_path] = class_data.SerializeToString()
-
- self.log.debug('create-new-instance-complete', device_id=device_id, class_id=class_id,
- entity_id=instance_id, attributes=attributes)
- return instance_data
-
- def set(self, device_id, class_id, instance_id, attributes):
- """
- Set a database value. This should only be called by the MIB synchronizer
- and its related tasks
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param instance_id: (int) ME Entity ID
- :param attributes: (dict) Attribute dictionary
-
- :returns: (bool) True if the value was saved to the database. False if the
- value was identical to the current instance
-
- :raises KeyError: If device does not exist
- :raises DatabaseStateError: If the database is not enabled
- """
- self.log.debug('set', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
- try:
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID should be a string')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError("Invalid Class ID: {}, should be 0..65535".format(class_id))
-
- if not 0 <= instance_id <= 0xFFFF:
- raise ValueError("Invalid Instance ID: {}, should be 0..65535".format(instance_id))
-
- if not isinstance(attributes, dict):
- raise TypeError("Attributes should be a dictionary")
-
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- operation = 'set'
- start_time = None
- try:
- class_data = MibClassData()
- class_path = MibDbExternal.CLASS_PATH.format(device_id, class_id)
- class_data.ParseFromString(self._kv_store[class_path])
-
- inst_data = next((inst for inst in class_data.instances
- if inst.instance_id == instance_id), None)
- modified = False
- new_data = None
- if inst_data is None:
- operation = 'create'
- start_time = datetime.utcnow()
- new_data = self._create_new_instance(device_id, class_id, instance_id, attributes)
- modified = True
- else:
-
- # Possibly adding to or updating an existing instance
- # Get instance proxy, creating it if needed
- new_attributes = []
- exist_attr_indexes = dict()
- attr_len = len(inst_data.attributes)
-
- for index in range(0, attr_len):
- name = inst_data.attributes[index].name
- value = inst_data.attributes[index].value
- exist_attr_indexes[name] = index
- new_attributes.append(MibAttributeData(name=name, value=value))
-
- for k, v in attributes.items():
- try:
- old_value = None if k not in exist_attr_indexes \
- else new_attributes[exist_attr_indexes[k]].value
-
- str_value = self._attribute_to_string(device_id, class_id, k, v, old_value)
-
- if k not in exist_attr_indexes:
- new_attributes.append(MibAttributeData(name=k, value=str_value))
- modified = True
-
- elif new_attributes[exist_attr_indexes[k]].value != str_value:
- new_attributes[exist_attr_indexes[k]].value = str_value
- modified = True
-
- except Exception as e:
- self.log.exception('save-error', e=e, class_id=class_id,
- attr=k, value_type=type(v))
- now = datetime.utcnow()
- start_time = now
- new_data = MibInstanceData(instance_id=instance_id,
- created=inst_data.created,
- modified=self._time_to_string(now),
- attributes=new_attributes)
-
- if modified:
- inst_index = next((index for index in range(len(class_data.instances)) if class_data.instances[index].instance_id == instance_id), None)
- if not inst_index is None:
- del class_data.instances[inst_index]
- class_data.instances.extend([new_data])
- self._kv_store[class_path] = class_data.SerializeToString()
- return modified
-
- except KeyError:
- # Here if the class-id does not yet exist in the database
- self.log.debug("adding-key-not-found", class_id=class_id)
- return self._add_new_class(device_id, class_id, instance_id,
- attributes)
- finally:
- if start_time is not None:
- diff = datetime.utcnow() - start_time
- # NOTE: Change to 'debug' when checked in, manually change to 'info'
- # for development testing.
- self.log.debug('db-{}-time'.format(operation), milliseconds=diff.microseconds/1000)
- self._statistics[operation].increment(diff.microseconds/1000)
-
- except Exception as e:
- self.log.exception('set-exception', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes, e=e)
- raise
-
- def delete(self, device_id, class_id, entity_id):
- """
- Delete an entity from the database if it exists. If all instances
- of a class are deleted, the class is deleted as well.
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) ME Class ID
- :param entity_id: (int) ME Entity ID
-
- :returns: (bool) True if the instance was found and deleted. False
- if it did not exist.
-
- :raises KeyError: If device does not exist
- :raises DatabaseStateError: If the database is not enabled
- """
- self.log.debug('delete', device_id=device_id, class_id=class_id,
- entity_id=entity_id)
-
- if not self._started:
- raise DatabaseStateError('The Database is not currently active')
-
- if not isinstance(device_id, six.string_types):
- raise TypeError('Device ID should be an string')
-
- if not 0 <= class_id <= 0xFFFF:
- raise ValueError('class-id is 0..0xFFFF')
-
- if not 0 <= entity_id <= 0xFFFF:
- raise ValueError('instance-id is 0..0xFFFF')
-
- start_time = datetime.utcnow()
- try:
- # Remove instance
- instpath = self._get_instance_path(device_id, class_id, entity_id)
- del self._kv_store[instpath]
- now = datetime.utcnow()
-
- # If resulting class has no instance, remove it as well
- classpath = self._get_class_path(device_id, class_id)
- class_data = MibClassData()
- class_data.ParseFromString(self._kv_store[classpath])
-
- if len(class_data.instances) == 0:
- del self._kv_store[classpath]
-
- self._modified = now
- return True
-
- except KeyError:
- return False # Not found
-
- except Exception as e:
- self.log.exception('get-last-sync-exception', device_id=device_id, e=e)
- raise
-
- finally:
- diff = datetime.utcnow() - start_time
- # NOTE: Change to 'debug' when checked in, manually change to 'info'
- # for development testing.
- self.log.debug('db-delete-time', milliseconds=diff.microseconds/1000)
- self._statistics['delete'].increment(diff.microseconds/1000)
-
- def query(self, device_id, class_id=None, instance_id=None, attributes=None):
- """
- Get database information.
-
- This method can be used to request information from the database to the detailed
- level requested
-
- :param device_id: (str) ONU Device ID
- :param class_id: (int) Managed Entity class ID
- :param instance_id: (int) Managed Entity instance
- :param attributes: (list/set or str) Managed Entity instance's attributes
-
- :return: (dict) The value(s) requested. If class/inst/attribute is
- not found, an empty dictionary is returned
- :raises KeyError: If the requested device does not exist
- :raises DatabaseStateError: If the database is not enabled
- """
- self.log.debug('query', device_id=device_id, class_id=class_id,
- instance_id=instance_id, attributes=attributes)
-
- start_time = datetime.utcnow()
- end_time = None
- try:
- if class_id is None:
- # Get full device info
- dev_data = MibDeviceData()
- device_path = self._get_device_path(device_id)
- dev_data.ParseFromString(self._kv_store[device_path])
- end_time = datetime.utcnow()
- data = self._device_to_dict(dev_data)
-
- elif instance_id is None:
- # Get all instances of the class
- try:
- class_data = MibClassData()
- class_path = self._get_class_path(device_id, class_id)
- class_data.ParseFromString(self._kv_store[class_path])
- end_time = datetime.utcnow()
- data = self._class_to_dict(device_id, class_data)
-
- except KeyError:
- data = dict()
-
- else:
- # Get all attributes of a specific ME
- try:
- class_data = MibClassData()
- instance_data = None
- class_path = self._get_class_path(device_id, class_id)
- class_data.ParseFromString(self._kv_store[class_path])
- end_time = datetime.utcnow()
-
- for inst in class_data.instances:
- if inst.instance_id == instance_id:
- instance_data = inst
-
- if instance_data == None:
- raise KeyError
-
- if attributes is None:
- # All Attributes
- data = self._instance_to_dict(device_id, class_id, instance_data)
-
- else:
- # Specific attribute(s)
- if isinstance(attributes, six.string_types):
- attributes = {attributes}
-
- data = {
- attr.name: self._string_to_attribute(device_id,
- class_id,
- attr.name,
- attr.value)
- for attr in instance_data.attributes if attr.name in attributes}
-
- except KeyError:
- data = dict()
-
- return data
-
- except KeyError:
- self.log.warn('query-no-device', device_id=device_id)
- raise
-
- except Exception as e:
- self.log.exception('get-last-sync-exception', device_id=device_id, e=e)
- raise
-
- finally:
- if end_time is not None:
- diff = end_time.utcnow() - start_time
- # NOTE: Change to 'debug' when checked in, manually change to 'info'
- # for development testing.
- self.log.debug('db-get-time', milliseconds=diff.microseconds/1000, class_id=class_id,
- instance_id=instance_id)
- self._statistics['get'].increment(diff.microseconds/1000)
-
def _instance_to_dict(self, device_id, class_id, instance):
if not isinstance(instance, MibInstanceData):
raise TypeError('{} is not of type MibInstanceData'.format(type(instance)))
@@ -995,6 +1012,7 @@
raise TypeError('{} is not of type MibClassData'.format(type(val)))
data = {
+ CLASS_ID_KEY: val.class_id
}
for instance in val.instances:
data[instance.instance_id] = self._instance_to_dict(device_id,
@@ -1002,7 +1020,7 @@
instance)
return data
- def _device_to_dict(self, val):
+ def _device_to_dict(self, val, classes_dict=None):
if not isinstance(val, MibDeviceData):
raise TypeError('{} is not of type MibDeviceData'.format(type(val)))
@@ -1015,9 +1033,13 @@
ME_KEY: dict(),
MSG_TYPE_KEY: set()
}
- for class_data in val.classes:
- data[class_data.class_id] = self._class_to_dict(val.device_id,
- class_data)
+ if classes_dict is None:
+ for class_data in val.classes:
+ data[class_data.class_id] = self._class_to_dict(val.device_id,
+ class_data)
+ else:
+ data.update(classes_dict)
+
for managed_entity in val.managed_entities:
data[ME_KEY][managed_entity.class_id] = managed_entity.name
@@ -1032,53 +1054,33 @@
return entity.__name__ if entity is not None else 'UnknownManagedEntity'
- def update_supported_managed_entities(self, device_id, managed_entities):
- """
- Update the supported OMCI Managed Entities for this device
- :param device_id: (str) ONU Device ID
- :param managed_entities: (set) Managed Entity class IDs
- """
- try:
- me_list = [ManagedEntity(class_id=class_id,
- name=self._managed_entity_to_name(device_id,
- class_id))
- for class_id in managed_entities]
- data = MibDeviceData()
- device_path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[device_path])
+ def load_from_template(self, device_id, template):
+ now = datetime.utcnow()
+ headerdata = {
+ DEVICE_ID_KEY: device_id,
+ CREATED_KEY: now,
+ LAST_SYNC_KEY: None,
+ MDS_KEY: 0,
+ VERSION_KEY: MibDbExternal.CURRENT_VERSION,
+ ME_KEY: dict(),
+ MSG_TYPE_KEY: set()
+ }
+ template.update(headerdata)
- now = datetime.utcnow()
- data.managed_entities.extend(me_list)
+ for cls_id, cls_data in template.items():
+ if isinstance(cls_id, int):
+ for inst_id, inst_data in cls_data.items():
+ if isinstance(inst_id, int):
+ self.set(device_id, cls_id, inst_id, template[cls_id][inst_id][ATTRIBUTES_KEY])
- # Update
- self._kv_store[device_path] = data.SerializeToString()
- self._modified = now
- self.log.debug('save-me-list-complete', device_id=device_id)
+ def dump_to_json(self, device_id):
+ device_db = self.query(device_id)
+ def json_converter(o):
+ if isinstance(o, datetime):
+ return o.__str__()
+ if isinstance(o, six.binary_type):
+ return o.decode('ascii')
- except Exception as e:
- self.log.exception('add-me-failure', e=e, me_list=managed_entities)
- raise
+ json_string = json.dumps(device_db, default=json_converter, indent=2)
- def update_supported_message_types(self, device_id, msg_types):
- """
- Update the supported OMCI Managed Entities for this device
- :param device_id: (str) ONU Device ID
- :param msg_types: (set) Message Type values (ints)
- """
- try:
- now = datetime.utcnow()
- msg_type_list = [MessageType(message_type=msg_type.value)
- for msg_type in msg_types]
- data = MibDeviceData()
- device_path = self._get_device_path(device_id)
- data.ParseFromString(self._kv_store[device_path])
- data.message_types.extend(msg_type_list)
-
- # Update
- self._kv_store[device_path] = data.SerializeToString()
- self._modified = now
- self.log.debug('save-msg-types-complete', device_id=device_id)
-
- except Exception as e:
- self.log.exception('add-msg-types-failure', e=e, msg_types=msg_types)
- raise
+ return json_string
diff --git a/pyvoltha/adapters/extensions/omci/database/mib_template_db.py b/pyvoltha/adapters/extensions/omci/database/mib_template_db.py
index a452c4f..0028839 100644
--- a/pyvoltha/adapters/extensions/omci/database/mib_template_db.py
+++ b/pyvoltha/adapters/extensions/omci/database/mib_template_db.py
@@ -14,12 +14,13 @@
# limitations under the License.
#
from __future__ import absolute_import
+from twisted.internet.defer import inlineCallbacks, returnValue
from .mib_db_api import CREATED_KEY, MODIFIED_KEY
import json
from datetime import datetime
import structlog
from pyvoltha.common.utils.registry import registry
-from pyvoltha.common.config.config_backend import EtcdStore
+from pyvoltha.adapters.common.kvstore.twisted_etcd_store import TwistedEtcdStore
import six
@@ -43,11 +44,13 @@
self.args = registry('main').get_args()
host, port = self.args.etcd.split(':', 1)
- self._kv_store = EtcdStore(host, port, MibTemplateDb.BASE_PATH)
+ self._kv_store = TwistedEtcdStore(host, port, MibTemplateDb.BASE_PATH)
self.loaded = False
- self._load_template()
def get_template_instance(self):
+ if not self._jsonstring:
+ return None
+
# swap out tokens with specific data
fixup = self._jsonstring.decode('ascii')
fixup = fixup.replace('%SERIAL_NUMBER%', self._serial_number)
@@ -67,15 +70,20 @@
return newdb
- def _load_template(self):
+ @inlineCallbacks
+ def load_template(self):
path = self._get_template_path()
- try:
- self._jsonstring = self._kv_store[path]
+ results = yield self._kv_store.get(path)
+ if results:
+ self._jsonstring = results
self.log.debug('found-template-data', path=path)
self.loaded = True
- except KeyError:
+ returnValue(True)
+ else:
self.log.warn('no-template-found', path=path)
+ returnValue(False)
+
def _get_template_path(self):
if not isinstance(self._vendor_id, six.string_types):
raise TypeError('Vendor ID is a string')
diff --git a/pyvoltha/adapters/extensions/omci/state_machines/mib_sync.py b/pyvoltha/adapters/extensions/omci/state_machines/mib_sync.py
index 640da14..27cad96 100644
--- a/pyvoltha/adapters/extensions/omci/state_machines/mib_sync.py
+++ b/pyvoltha/adapters/extensions/omci/state_machines/mib_sync.py
@@ -18,6 +18,7 @@
from datetime import datetime, timedelta
from transitions import Machine
from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks, returnValue
from pyvoltha.adapters.extensions.omci.omci_frame import OmciFrame
from pyvoltha.adapters.extensions.omci.database.mib_db_api import MDS_KEY
from pyvoltha.adapters.extensions.omci.omci_defs import EntityOperations, ReasonCodes, \
@@ -305,12 +306,13 @@
# TODO: Stop and remove any currently running or scheduled tasks
# TODO: Anything else?
+ @inlineCallbacks
def _seed_database(self):
if not self._device_in_db:
try:
try:
self._database.start()
- self._database.add(self._device_id)
+ yield self._database.add(self._device_id)
self.log.debug('seed-db-does-not-exist', device_id=self._device_id)
except KeyError:
@@ -324,6 +326,7 @@
except Exception as e:
self.log.exception('seed-database-failure', e=e)
+ @inlineCallbacks
def on_enter_starting(self):
"""
Determine ONU status and start/re-start MIB Synchronization tasks
@@ -332,7 +335,7 @@
self.advertise(OpenOmciEventType.state_change, self.state)
# Make sure root of external MIB Database exists
- self._seed_database()
+ yield self._seed_database()
# Set up Response and Autonomous notification subscriptions
try:
diff --git a/pyvoltha/adapters/extensions/omci/tasks/mib_template_task.py b/pyvoltha/adapters/extensions/omci/tasks/mib_template_task.py
index 505a444..48e528c 100644
--- a/pyvoltha/adapters/extensions/omci/tasks/mib_template_task.py
+++ b/pyvoltha/adapters/extensions/omci/tasks/mib_template_task.py
@@ -127,15 +127,17 @@
# Lookup template base on unique onu type info
template = None
+ found = False
if vendor_id and equipment_id and software_version:
self.log.debug('looking-up-template', vendor_id=vendor_id, equipment_id=equipment_id,
software_version=software_version)
template = MibTemplateDb(vendor_id, equipment_id, software_version, serial_number, mac_address)
+ found = yield template.load_template()
else:
self.log.info('no-usable-template-lookup-data', vendor_id=vendor_id, equipment_id=equipment_id,
software_version=software_version)
- if template and template.loaded:
+ if template and found:
# generate db instance
loaded_template_instance = template.get_template_instance()
self.deferred.callback(loaded_template_instance)