Device component - OpenConfig Driver:

# 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.

# 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 logging, threading

from typing import Any, List, Tuple

from ncclient.manager import Manager, connect_ssh

from common.tools.client.RetryDecorator import delay_exponential

from common.type_checkers.Checkers import chk_length, chk_string, chk_type

from device.service.driver_api.Exceptions import UnsupportedResourceKeyException

from .templates import EMPTY_CONFIG, compose_config

from .RetryDecorator import retry

MAX_RETRIES = 15

DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)

RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')

LOGGER = logging.getLogger(__name__)

class NetconfSessionHandler:

    def __init__(self, address : str, port : int, **settings) -> None:

        self.__lock = threading.RLock()

        self.__connected = threading.Event()

        self.__address = address

        self.__port = int(port)

        self.__username       = settings.get('username')

        self.__password       = settings.get('password')

        self.__vendor         = settings.get('vendor')

        self.__key_filename   = settings.get('key_filename')

        self.__hostkey_verify = settings.get('hostkey_verify', True)

        self.__look_for_keys  = settings.get('look_for_keys', True)

        self.__allow_agent    = settings.get('allow_agent', True)

        self.__force_running  = settings.get('force_running', False)

        self.__commit_per_delete  = settings.get('delete_rule', False)

        self.__device_params  = settings.get('device_params', {})

        self.__manager_params = settings.get('manager_params', {})

        self.__nc_params      = settings.get('nc_params', {})

        self.__manager : Manager   = None

        self.__candidate_supported = False

    def connect(self):

        with self.__lock:

            self.__manager = connect_ssh(

                host=self.__address, port=self.__port, username=self.__username, password=self.__password,

                device_params=self.__device_params, manager_params=self.__manager_params, nc_params=self.__nc_params,

                key_filename=self.__key_filename, hostkey_verify=self.__hostkey_verify, allow_agent=self.__allow_agent,

                look_for_keys=self.__look_for_keys)

            self.__candidate_supported = ':candidate' in self.__manager.server_capabilities

            self.__connected.set()

    def disconnect(self):

        if not self.__connected.is_set(): return

        with self.__lock:

            self.__manager.close_session()

    @property

    def use_candidate(self): return self.__candidate_supported and not self.__force_running

    @property

    def commit_per_rule(self): return self.__commit_per_delete 

    @property

    def vendor(self): return self.__vendor

    @RETRY_DECORATOR

    def get(self, filter=None, with_defaults=None): # pylint: disable=redefined-builtin

        with self.__lock:

            return self.__manager.get(filter=filter, with_defaults=with_defaults)

    @RETRY_DECORATOR

    def edit_config(

        self, config, target='running', default_operation=None, test_option=None,

        error_option=None, format='xml'                                             # pylint: disable=redefined-builtin

    ):

        if config == EMPTY_CONFIG: return

        with self.__lock:

            self.__manager.edit_config(

                config, target=target, default_operation=default_operation, test_option=test_option,

                error_option=error_option, format=format)

    def locked(self, target):

        return self.__manager.locked(target=target)

    def commit(self, confirmed=False, timeout=None, persist=None, persist_id=None):

        return self.__manager.commit(confirmed=confirmed, timeout=timeout, persist=persist, persist_id=persist_id)

def edit_config(

    netconf_handler : NetconfSessionHandler, resources : List[Tuple[str, Any]], delete=False, commit_per_rule= False,

    target='running', default_operation='merge', test_option=None, error_option=None,

    format='xml' # pylint: disable=redefined-builtin

):

    str_method = 'DeleteConfig' if delete else 'SetConfig'

    LOGGER.info('[{:s}] resources = {:s}'.format(str_method, str(resources)))

    results = [None for _ in resources]

    for i,resource in enumerate(resources):

        str_resource_name = 'resources[#{:d}]'.format(i)

        try:

            LOGGER.info('[{:s}] resource = {:s}'.format(str_method, str(resource)))

            chk_type(str_resource_name, resource, (list, tuple))

            chk_length(str_resource_name, resource, min_length=2, max_length=2)

            resource_key,resource_value = resource

            chk_string(str_resource_name + '.key', resource_key, allow_empty=False)

            str_config_message = compose_config(

                resource_key, resource_value, delete=delete, vendor=netconf_handler.vendor)

            if str_config_message is None: raise UnsupportedResourceKeyException(resource_key)

            LOGGER.info('[{:s}] str_config_message[{:d}] = {:s}'.format(

                str_method, len(str_config_message), str(str_config_message)))

            netconf_handler.edit_config(

                config=str_config_message, target=target, default_operation=default_operation,

                test_option=test_option, error_option=error_option, format=format)

            if commit_per_rule:

                netconf_handler.commit()

            results[i] = True

        except Exception as e: # pylint: disable=broad-except

            str_operation = 'preparing' if target == 'candidate' else ('deleting' if delete else 'setting')

            msg = '[{:s}] Exception {:s} {:s}: {:s}'

            LOGGER.exception(msg.format(str_method, str_operation, str_resource_name, str(resource)))

            results[i] = e # if validation fails, store the exception

    return results

# See the License for the specific language governing permissions and

# limitations under the License.

import anytree, copy, logging, pytz, queue, re, threading, json, os, sys

import anytree, copy, logging, pytz, queue, re, threading

#import lxml.etree as ET

from datetime import datetime, timedelta

from typing import Any, Dict, Iterator, List, Optional, Tuple, Union

from .templates import ALL_RESOURCE_KEYS, EMPTY_CONFIG, compose_config, get_filter, parse

from .RetryDecorator import retry

import grpc

from google.protobuf.json_format import MessageToJson

gnmi_path__ = os.path.dirname(os.path.abspath(__file__))

sys.path.append(gnmi_path__)

import gnmi_pb2_grpc

import gnmi_pb2

DEBUG_MODE = False

logging.getLogger('ncclient.manager').setLevel(logging.DEBUG if DEBUG_MODE else logging.WARNING)

    def commit(self, confirmed=False, timeout=None, persist=None, persist_id=None):

        return self.__manager.commit(confirmed=confirmed, timeout=timeout, persist=persist, persist_id=persist_id)

class gNMISessionHandler:

    def __init__(self, address : str, **settings) -> None:

        self.__lock = threading.RLock()

        self.__connected = threading.Event()

        self.__address = address

        self.__port           = settings.get('gnmi_port')

        self.__username       = settings.get('username')

        self.__password       = settings.get('password')

        self.__vendor         = settings.get('vendor')

        self.__key_filename   = settings.get('key_filename')

        self.__hostkey_verify = settings.get('hostkey_verify', True)

        self.__look_for_keys  = settings.get('look_for_keys', True)

        self.__allow_agent    = settings.get('allow_agent', True)

        self.__force_running  = settings.get('force_running', False)

        self.__commit_per_delete  = settings.get('delete_rule', False)

        self.__device_params  = settings.get('device_params', {})

        self.__manager_params = settings.get('manager_params', {})

        self.__nc_params      = settings.get('nc_params', {})

        self.__stub = None

        self.__candidate_supported = False

        self.__channel = None

        self.__supportedEncodings = None

        self.__options = Options()

    def connect(self):

        with self.__lock:

            self.__channel = grpc.insecure_channel(str(self.__address)+':'+self.__port)

            self.__stub = gnmi_pb2_grpc.gNMIStub(self.__channel)

            metadata = [('username',self.__username ), ('password', self.__password)]

            req = gnmi_pb2.CapabilityRequest()

            response = self.__stub.Capabilities(req, metadata=metadata)

            data = json.loads(MessageToJson(response))

            self.__supportedEncodings = data['supportedEncodings']

            # TODO: self.__candidate_supported = 

            self.__connected.set()

    def disconnect(self):

        if not self.__connected.is_set(): return

        with self.__lock:

            self.__channel.close()

    def subscribeStreaming(self, subscription : Tuple[str, float, float], out_samples : queue.Queue) -> None:

        resource_key, sampling_duration, sampling_interval = subscription

        options = copy.deepcopy(self.__options)

        options.xpaths = [parse_xpath(resource_key)]

        options.timeout = int(sampling_duration)

        options.interval = int(sampling_interval)

        req_iterator = gen_request(options)

        metadata = [('username',self.__username), ('password', self.__password)]

        responses = self.__stub.Subscribe(req_iterator, self.__options.timeout, metadata=metadata)

        previous_sample = None

        delta = 0.0

        previous_timestamp = datetime.timestamp(datetime.utcnow())

        for response in responses:

            data = json.loads(MessageToJson(response))

            if data.get("update") is not None and data.get("update").get("update") != None:

                now = datetime.timestamp(datetime.utcnow())

                for element in data['update']['update']:

                    counter_name = split_resource_key(dict_to_xpath(element['path']))

                    if counter_name == split_resource_key(resource_key):

                        value = int(element['val']['uintVal'])

                        delay = now - previous_timestamp

                        if previous_sample is not None: delta = (value - previous_sample)/delay

                        previous_sample = int(value)

                        previous_timestamp = now

                        sample = (now, resource_key, delta)

                        out_samples.put_nowait(sample)

    @property

    def use_candidate(self): return self.__candidate_supported and not self.__force_running

    @property

    def commit_per_rule(self): return self.__commit_per_delete 

    @property

    def vendor(self): return self.__vendor

    @RETRY_DECORATOR

    def get(self): # pylint: disable=redefined-builtin

        return False

    @RETRY_DECORATOR

    def edit_config(

        self, config, target='running', default_operation=None, test_option=None,

        error_option=None, format='xml'                                             # pylint: disable=redefined-builtin

    ):

        if config == EMPTY_CONFIG: return

        with self.__lock:

            self.__manager.edit_config(

                config, target=target, default_operation=default_operation, test_option=test_option,

                error_option=error_option, format=format)

    @RETRY_DECORATOR

    def locked(self, target):

        return self.__manager.locked(target=target)

    @RETRY_DECORATOR

    def commit(self, confirmed=False, timeout=None, persist=None, persist_id=None):

        return self.__manager.commit(confirmed=confirmed, timeout=timeout, persist=persist, persist_id=persist_id)

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

    return delta_sample

class SamplesCache:

    def __init__(self, netconf_handler : NetconfSessionHandler, gNMI_handler : gNMISessionHandler, logger : logging.Logger) -> None:

    def __init__(self, netconf_handler : NetconfSessionHandler, logger : logging.Logger) -> None:

        self.__netconf_handler = netconf_handler

        self.__logger = logger

        self.__lock = threading.Lock()

        self.__timestamp = None

        self.__absolute_samples = {}

        self.__delta_samples = {}

        self.__gNMI_handler = gNMI_handler

    def _refresh_samples(self) -> None:

        with self.__lock:

    except: # pylint: disable=bare-except

        logger.exception('Error retrieving samples')

class Options:

    def __init__(self, xpaths=None, prefix=None, mode=0, submode=0, suppress=False, interval=0,

                 encoding='JSON', heartbeat=0, qos=None, aggregate=False, server=None, username='admin', password='admin', timeout=None):

        self.xpaths = xpaths

        self.prefix = prefix

        self.mode = mode

        self.submode = submode

        self.suppress = suppress

        self.interval = interval

        self.encoding = encoding

        self.heartbeat = heartbeat

        self.qos = qos

        self.aggregate = aggregate

        self.server = server

        self.username = username

        self.password = password

        self.timeout = timeout

def edit_config(                                                                                                            # edit the configuration of openconfig devices

    netconf_handler : NetconfSessionHandler, logger : logging.Logger, resources : List[Tuple[str, Any]], delete=False,

    commit_per_rule=False, target='running', default_operation='merge', test_option=None, error_option=None,

        self.__subscriptions = TreeNode('.')

        self.__started = threading.Event()

        self.__terminate = threading.Event()

        self.__gnmi_monitoring = settings.get('monitoring_protocol') == 'gnmi'

        self.__scheduler = BackgroundScheduler(daemon=True) # scheduler used to emulate sampling events

        self.__scheduler.configure(

            jobstores = {'default': MemoryJobStore()},

            timezone=pytz.utc)

        self.__out_samples = queue.Queue()

        self.__netconf_handler = NetconfSessionHandler(self.address, self.port, **(self.settings))

        self.__gNMI_handler : gNMISessionHandler = gNMISessionHandler(address, **settings)

        self.__samples_cache = SamplesCache(self.__netconf_handler, self.__gNMI_handler, self.__logger)

        self.__samples_cache = SamplesCache(self.__netconf_handler, self.__logger)

    def Connect(self) -> bool:

        with self.__lock:

            if self.__started.is_set(): return True

            self.__netconf_handler.connect()

            if self.__gnmi_monitoring: 

                self.__gNMI_handler.connect()

                LOGGER.debug('Using gNMI as monitoring protocol')

            # Connect triggers activation of sampling events that will be scheduled based on subscriptions

            self.__scheduler.start()

            self.__started.set()

            if not self.__started.is_set(): return True

            # Disconnect triggers deactivation of sampling events

            self.__scheduler.shutdown()

            if self.__gnmi_monitoring: self.__netconf_handler.disconnect()

            self.__netconf_handler.disconnect()

            return True

    @metered_subclass_method(METRICS_POOL)

                job_id = 'k={:s}/d={:f}/i={:f}'.format(resource_key, sampling_duration, sampling_interval)

                if self.__gnmi_monitoring:

                    LOGGER.debug('Processing gNMI subscription: '+ str(subscription))

                    job = threading.Thread(target=self.__gNMI_handler.subscribeStreaming, args=(subscription, self.__out_samples))

                    job.start()

                else:

                job = self.__scheduler.add_job(

                    do_sampling, args=(self.__samples_cache, self.__logger, resource_key, self.__out_samples),

                    kwargs={}, id=job_id, trigger='interval', seconds=sampling_interval,

# 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 import get_filter, parse

from .NetconfSessionHandler import NetconfSessionHandler

SAMPLE_EVICTION_SECONDS = 30.0 # seconds

SAMPLE_RESOURCE_KEY = 'interfaces/interface/state/counters'

RE_GET_ENDPOINT_FROM_INTERFACE_KEY = re.compile(r'.*interface\[([^\]]+)\].*')

RE_GET_ENDPOINT_FROM_INTERFACE_XPATH = re.compile(r".*interface\[oci\:name\='([^\]]+)'\].*")

LOGGER = logging.getLogger(__name__)

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, netconf_handler : NetconfSessionHandler) -> None:

        self.__netconf_handler = netconf_handler

        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.__netconf_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

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