Device component - gNMI OpenConfig Driver:

- Added Delta Sample calculation
- Improved timestamp reporting

src/device/service/drivers/gnmi_openconfig/DeltaSampleCache.py

+# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import copy
+from typing import Any, Dict, Tuple, Union
+
+class DeltaSampleCache:
+    def __init__(self) -> None:
+        self._previous_samples : Dict[str, Tuple[float, Union[int, float]]] = dict()
+
+    def get_delta(self, path : str, current_timestamp : float, current_value : Any) -> None:
+        previous_sample = copy.deepcopy(self._previous_samples.get(path))
+        self._previous_samples[path] = current_timestamp, current_value
+
+        if not isinstance(current_value, (int, float)): return None
+        if previous_sample is None: return current_timestamp, 0
+        previous_timestamp, previous_value = previous_sample
+        if not isinstance(previous_value, (int, float)): return None
+
+        delta_value = max(0, current_value - previous_value)
+        delay = current_timestamp - previous_timestamp
+        delta_sample = current_timestamp, delta_value / delay
+
+        return delta_sample

src/device/service/drivers/gnmi_openconfig/MonitoringThread.py

@@ -16,7 +16,7 @@
 # Ref: https://github.com/openconfig/gnmi/blob/master/proto/gnmi/gnmi.proto
 
 from __future__ import annotations
-import grpc, logging, queue, threading
+import grpc, logging, queue, re, threading
 from collections.abc import Iterator
 from datetime import datetime
 from typing import Dict
@@ -26,6 +26,7 @@ from .gnmi.gnmi_pb2 import ( # pylint: disable=no-name-in-module
 )
 from .gnmi.gnmi_pb2_grpc import gNMIStub
 from .tools.Path import path_from_string, path_to_string
+from .DeltaSampleCache import DeltaSampleCache
 
 
 LOGGER = logging.getLogger(__name__)
@@ -82,6 +83,7 @@ class MonitoringThread(threading.Thread):
         self._in_subscriptions = in_subscriptions
         self._out_samples = out_samples
         self._response_iterator = None
+        self._delta_sample_cache = DeltaSampleCache()
 
     def stop(self) -> None:
         self._terminate.set()
@@ -131,16 +133,35 @@ class MonitoringThread(threading.Thread):
             timeout = None # GNMI_SUBSCRIPTION_TIMEOUT = int(sampling_duration)
             self._response_iterator = self._stub.Subscribe(request_iterator, metadata=metadata, timeout=timeout)
             for subscribe_response in self._response_iterator:
-                timestamp = datetime.timestamp(datetime.utcnow())
                 str_subscribe_response = grpc_message_to_json_string(subscribe_response)
                 self._logger.warning('[run] subscribe_response={:s}'.format(str_subscribe_response))
-                for update in subscribe_response.update.update:
-                    str_path = path_to_string(update.path)
-                    if str_path != '/system/name/host-name': continue
-                    #counter_name = update.path[-1].name
-                    value_type = update.val.WhichOneof('value')
-                    value = getattr(update.val, value_type)
-                    sample = (timestamp, str_path, value)
+                update = subscribe_response.update
+                timestamp_device = float(update.timestamp) / 1.e9
+                timestamp_local = datetime.timestamp(datetime.utcnow())
+                # if difference between timestamp from device and local is lower than 1 second
+                if abs(timestamp_device - timestamp_local) <= 1:
+                    # assume clocks are synchronized, use timestamp from device
+                    timestamp = timestamp_device
+                else:
+                    # might be clocks are not synchronized, use local timestamp
+                    timestamp = timestamp_local
+                for update_entry in update.update:
+                    str_path = path_to_string(update_entry.path)
+                    #if str_path != '/system/name/host-name': continue
+                    #counter_name = update_entry.path[-1].name
+                    value_type = update_entry.val.WhichOneof('value')
+                    value = getattr(update_entry.val, value_type)
+                    if re.match(r'^[0-9]+$', value) is not None:
+                        value = int(value)
+                    elif re.match(r'^[0-9]*\.[0-9]*$', value) is not None:
+                        value = float(value)
+                    else:
+                        value = str(value)
+                    delta_sample = self._delta_sample_cache.get_delta(str_path, timestamp, value)
+                    if delta_sample is None:
+                        sample = (timestamp, str_path, value)
+                    else:
+                        sample = (delta_sample[0], str_path, delta_sample[1])
                     self._logger.warning('[run] sample={:s}'.format(str(sample)))
                     self._out_samples.put_nowait(sample)
         except grpc.RpcError as e:

src/device/service/drivers/gnmi_openconfig/SamplesCache.py

-# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
-#
-# 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.
-
-# Collection of samples through NetConf is very slow and each request collects all the data.
-# Populate a cache periodically (when first interface is interrogated).
-# Evict data after some seconds, when data is considered as outdated
-
-import copy, queue, logging, re, threading
-from datetime import datetime
-from typing import Dict, Tuple
-from .templates_old import get_filter, parse
-from .GnmiSessionHandler import GnmiSessionHandler
-
-RE_GET_ENDPOINT_FROM_INTERFACE_KEY = re.compile(r'.*interface\[([^\]]+)\].*')
-RE_GET_ENDPOINT_FROM_INTERFACE_XPATH = re.compile(r".*interface\[oci\:name\='([^\]]+)'\].*")
-
-SAMPLE_EVICTION_SECONDS = 30.0 # seconds
-SAMPLE_RESOURCE_KEY = 'interfaces/interface/state/counters'
-
-def compute_delta_sample(previous_sample, previous_timestamp, current_sample, current_timestamp):
-    if previous_sample is None: return None
-    if previous_timestamp is None: return None
-    if current_sample is None: return None
-    if current_timestamp is None: return None
-    delay = current_timestamp - previous_timestamp
-    field_keys = set(previous_sample.keys()).union(current_sample.keys())
-    field_keys.discard('name')
-    delta_sample = {'name': previous_sample['name']}
-    for field_key in field_keys:
-        previous_sample_value = previous_sample[field_key]
-        if not isinstance(previous_sample_value, (int, float)): continue
-        current_sample_value = current_sample[field_key]
-        if not isinstance(current_sample_value, (int, float)): continue
-        delta_value = current_sample_value - previous_sample_value
-        if delta_value < 0: continue
-        delta_sample[field_key] = delta_value / delay
-    return delta_sample
-
-class SamplesCache:
-    def __init__(self, handler : GnmiSessionHandler, logger : logging.Logger) -> None:
-        self.__handler = handler
-        self.__logger = logger
-        self.__lock = threading.Lock()
-        self.__timestamp = None
-        self.__absolute_samples = {}
-        self.__delta_samples = {}
-
-    def _refresh_samples(self) -> None:
-        with self.__lock:
-            try:
-                now = datetime.timestamp(datetime.utcnow())
-                if self.__timestamp is not None and (now - self.__timestamp) < SAMPLE_EVICTION_SECONDS: return
-                str_filter = get_filter(SAMPLE_RESOURCE_KEY)
-                xml_data = self.__handler.get(filter=str_filter).data_ele
-                interface_samples = parse(SAMPLE_RESOURCE_KEY, xml_data)
-                for interface,samples in interface_samples:
-                    match = RE_GET_ENDPOINT_FROM_INTERFACE_KEY.match(interface)
-                    if match is None: continue
-                    interface = match.group(1)
-                    delta_sample = compute_delta_sample(
-                        self.__absolute_samples.get(interface), self.__timestamp, samples, now)
-                    if delta_sample is not None: self.__delta_samples[interface] = delta_sample
-                    self.__absolute_samples[interface] = samples
-                self.__timestamp = now
-            except: # pylint: disable=bare-except
-                self.__logger.exception('Error collecting samples')
-
-    def get(self, resource_key : str) -> Tuple[float, Dict]:
-        self._refresh_samples()
-        match = RE_GET_ENDPOINT_FROM_INTERFACE_XPATH.match(resource_key)
-        with self.__lock:
-            if match is None: return self.__timestamp, {}
-            interface = match.group(1)
-            return self.__timestamp, copy.deepcopy(self.__delta_samples.get(interface, {}))
-
-def do_sampling(
-    samples_cache : SamplesCache, logger : logging.Logger, resource_key : str, out_samples : queue.Queue
-) -> None:
-    try:
-        timestamp, samples = samples_cache.get(resource_key)
-        counter_name = resource_key.split('/')[-1].split(':')[-1]
-        value = samples.get(counter_name)
-        if value is None:
-            logger.warning('[do_sampling] value not found for {:s}'.format(resource_key))
-            return
-        # resource_key template: //oci:interfaces/oci:interface[oci:name='{:s}']/state/counters/{:s}
-        sample = (timestamp, resource_key, value)
-        out_samples.put_nowait(sample)
-    except: # pylint: disable=bare-except
-        logger.exception('Error retrieving samples')