# 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')