Commit b5002f42 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device: 

- Added Routing Policies as special resource for interrogation during initialization
- Added flags in unitary tests to control which tests to be executed in OpenConfig.
- Added run-test-locally specific for device component.

OpenConfig Driver:
- Implemented routing policy templates, parsing and configuration.
- Implemented retry decorator for NetConf client.
parent 5d1be5cb

scripts/run_tests_locally-device.sh

0 → 100755
+28 −0
Original line number Diff line number Diff line 
#!/bin/bash

# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

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





PROJECTDIR=`pwd`



cd $PROJECTDIR/src

RCFILE=$PROJECTDIR/coverage/.coveragerc



# Run unitary tests and analyze coverage of code at same time



# Useful flags for pytest:

#-o log_cli=true -o log_file=device.log -o log_file_level=DEBUG



coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \

    device/tests/test_unitary.py

src/device/service/driver_api/_Driver.py

+1 −0
Original line number Diff line number Diff line
@@ -20,6 +20,7 @@ from typing import Any, Iterator, List, Optional, Tuple, Union 
RESOURCE_ENDPOINTS         = '__endpoints__'

RESOURCE_INTERFACES        = '__interfaces__'

RESOURCE_NETWORK_INSTANCES = '__network_instances__'

RESOURCE_ROUTING_POLICIES  = '__routing_policies__'



class _Driver:

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

src/device/service/drivers/openconfig/OpenConfigDriver.py

+69 −36
Original line number Diff line number Diff line
@@ -13,22 +13,23 @@ 
# limitations under the License.



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

import lxml.etree as ET

#import lxml.etree as ET

from datetime import datetime, timedelta

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

from apscheduler.executors.pool import ThreadPoolExecutor

from apscheduler.job import Job

from apscheduler.jobstores.memory import MemoryJobStore

from apscheduler.schedulers.background import BackgroundScheduler

from netconf_client.connect import connect_ssh

from netconf_client.connect import connect_ssh, Session

from netconf_client.ncclient import Manager

from common.tools.client.RetryDecorator import delay_exponential

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

from device.service.driver_api.Exceptions import UnsupportedResourceKeyException

from device.service.driver_api._Driver import _Driver

from device.service.driver_api.AnyTreeTools import TreeNode, dump_subtree, get_subnode, set_subnode_value

from device.service.drivers.openconfig.Tools import xml_pretty_print, xml_to_dict, xml_to_file

from device.service.drivers.openconfig.templates import (

    ALL_RESOURCE_KEYS, EMPTY_CONFIG, compose_config, get_filter, parse)

from device.service.driver_api.AnyTreeTools import TreeNode, get_subnode, set_subnode_value #dump_subtree

#from .Tools import xml_pretty_print, xml_to_dict, xml_to_file

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

from .RetryDecorator import retry



DEBUG_MODE = False

#logging.getLogger('ncclient.transport.ssh').setLevel(logging.DEBUG if DEBUG_MODE else logging.WARNING)
@@ -48,6 +49,51 @@ 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'



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



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.__timeout = int(settings.get('timeout', 120))

        self.__netconf_session : Session = None

        self.__netconf_manager : Manager = None



    def connect(self):

        with self.__lock:

            self.__netconf_session = connect_ssh(

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

            self.__netconf_manager = Manager(self.__netconf_session, timeout=self.__timeout)

            self.__netconf_manager.set_logger_level(logging.DEBUG if DEBUG_MODE else logging.WARNING)

            self.__connected.set()



    def disconnect(self):

        if not self.__connected.is_set(): return

        with self.__lock:

            self.__netconf_manager.close_session()



    @RETRY_DECORATOR

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

        with self.__lock:

            return self.__netconf_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.__netconf_manager.edit_config(

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

                error_option=error_option, format=format)



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
@@ -68,19 +114,20 @@ def compute_delta_sample(previous_sample, previous_timestamp, current_sample, cu 
    return delta_sample



class SamplesCache:

    def __init__(self) -> None:

    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, netconf_manager : Manager) -> None:

    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 = netconf_manager.get(filter=str_filter).data_ele

                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)
@@ -94,19 +141,17 @@ class SamplesCache: 
            except: # pylint: disable=bare-except

                LOGGER.exception('Error collecting samples')



    def get(self, resource_key : str, netconf_manager : Manager) -> Tuple[float, Dict]:

        self._refresh_samples(netconf_manager)

    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(

    netconf_manager : Manager, samples_cache : SamplesCache, resource_key : str, out_samples : queue.Queue

) -> None:

def do_sampling(samples_cache : SamplesCache, resource_key : str, out_samples : queue.Queue) -> None:

    try:

        timestamp, samples = samples_cache.get(resource_key, netconf_manager)

        timestamp, samples = samples_cache.get(resource_key)

        counter_name = resource_key.split('/')[-1].split(':')[-1]

        value = samples.get(counter_name)

        if value is None:
@@ -117,19 +162,14 @@ def do_sampling( 
    except: # pylint: disable=bare-except

        LOGGER.exception('Error retrieving samples')





class OpenConfigDriver(_Driver):

    def __init__(self, address : str, port : int, **settings) -> None: # pylint: disable=super-init-not-called

        self.__address = address

        self.__port = int(port)

        self.__settings = settings

        self.__lock = threading.Lock()

        #self.__initial = TreeNode('.')

        #self.__running = TreeNode('.')

        self.__subscriptions = TreeNode('.')

        self.__started = threading.Event()

        self.__terminate = threading.Event()

        self.__netconf_manager : Manager = None

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

        self.__scheduler.configure(

            jobstores = {'default': MemoryJobStore()},
@@ -137,18 +177,13 @@ class OpenConfigDriver(_Driver): 
            job_defaults = {'coalesce': False, 'max_instances': 3},

            timezone=pytz.utc)

        self.__out_samples = queue.Queue()

        self.__samples_cache = SamplesCache()

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

        self.__samples_cache = SamplesCache(self.__netconf_handler)



    def Connect(self) -> bool:

        with self.__lock:

            if self.__started.is_set(): return True

            username = self.__settings.get('username')

            password = self.__settings.get('password')

            timeout = int(self.__settings.get('timeout', 120))

            session = connect_ssh(

                host=self.__address, port=self.__port, username=username, password=password)

            self.__netconf_manager = Manager(session, timeout=timeout)

            self.__netconf_manager.set_logger_level(logging.DEBUG if DEBUG_MODE else logging.WARNING)

            self.__netconf_handler.connect()

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

            self.__scheduler.start()

            self.__started.set()
@@ -162,7 +197,7 @@ class OpenConfigDriver(_Driver): 
            if not self.__started.is_set(): return True

            # Disconnect triggers deactivation of sampling events

            self.__scheduler.shutdown()

            self.__netconf_manager.close_session()

            self.__netconf_handler.disconnect()

            return True



    def GetInitialConfig(self) -> List[Tuple[str, Any]]:
@@ -179,8 +214,9 @@ class OpenConfigDriver(_Driver): 
                try:

                    chk_string(str_resource_name, resource_key, allow_empty=False)

                    str_filter = get_filter(resource_key)

                    LOGGER.info('[GetConfig] str_filter = {:s}'.format(str(str_filter)))

                    if str_filter is None: str_filter = resource_key

                    xml_data = self.__netconf_manager.get(filter=str_filter).data_ele

                    xml_data = self.__netconf_handler.get(filter=str_filter).data_ele

                    if isinstance(xml_data, Exception): raise xml_data

                    results.extend(parse(resource_key, xml_data))

                except Exception as e: # pylint: disable=broad-except
@@ -206,8 +242,7 @@ class OpenConfigDriver(_Driver): 
                    if str_config_message is None: raise UnsupportedResourceKeyException(resource_key)

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

                        len(str_config_message), str(str_config_message)))

                    if str_config_message != EMPTY_CONFIG:

                        self.__netconf_manager.edit_config(str_config_message, target='running')

                    self.__netconf_handler.edit_config(str_config_message, target='running')

                    results.append(True)

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

                    LOGGER.exception('Exception setting {:s}: {:s}'.format(str_resource_name, str(resource)))
@@ -232,9 +267,7 @@ class OpenConfigDriver(_Driver): 
                    if str_config_message is None: raise UnsupportedResourceKeyException(resource_key)

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

                        len(str_config_message), str(str_config_message)))

                    if str_config_message != EMPTY_CONFIG:

                        self.__netconf_manager.edit_config(str_config_message, target='running')

                    self.__netconf_manager.edit_config(str_config_message, target='running')

                    self.__netconf_handler.edit_config(str_config_message, target='running')

                    results.append(True)

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

                    LOGGER.exception('Exception deleting {:s}: {:s}'.format(str_resource_name, str(resource_key)))
@@ -269,7 +302,7 @@ class OpenConfigDriver(_Driver): 


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

                job = self.__scheduler.add_job(

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

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

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

                    start_date=start_date, end_date=end_date, timezone=pytz.utc)

src/device/service/drivers/openconfig/RetryDecorator.py

0 → 100644
+46 −0
Original line number Diff line number Diff line 
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)

#

# 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, time

from common.tools.client.RetryDecorator import delay_linear



LOGGER = logging.getLogger(__name__)



def retry(max_retries=0, delay_function=delay_linear(initial=0, increment=0),

          prepare_method_name=None, prepare_method_args=[], prepare_method_kwargs={}):

    def _reconnect(func):

        def wrapper(self, *args, **kwargs):

            if prepare_method_name is not None:

                prepare_method = getattr(self, prepare_method_name, None)

                if prepare_method is None: raise Exception('Prepare Method ({}) not found'.format(prepare_method_name))

            num_try, given_up = 0, False

            while not given_up:

                try:

                    return func(self, *args, **kwargs)

                except OSError as e:

                    if str(e) != 'Socket is closed': raise



                    num_try += 1

                    given_up = num_try > max_retries

                    if given_up: raise Exception('Giving up... {:d} tries failed'.format(max_retries)) from e

                    if delay_function is not None:

                        delay = delay_function(num_try)

                        time.sleep(delay)

                        LOGGER.info('Retry {:d}/{:d} after {:f} seconds...'.format(num_try, max_retries, delay))

                    else:

                        LOGGER.info('Retry {:d}/{:d} immediate...'.format(num_try, max_retries))



                    if prepare_method_name is not None: prepare_method(*prepare_method_args, **prepare_method_kwargs)

        return wrapper

    return _reconnect

src/device/service/drivers/openconfig/templates/EndPoints.py

+1 −1
Original line number Diff line number Diff line
@@ -47,5 +47,5 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]: 
        add_value_from_collection(endpoint, 'sample_types', sample_types)



        if len(endpoint) == 0: continue

        response.append(('endpoint[{:s}]'.format(endpoint['uuid']), endpoint))

        response.append(('/endpoint[{:s}]'.format(endpoint['uuid']), endpoint))

    return response