Commit 2944c65f authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device component - IETF Slice: 

- Added support for TFS-API-based topology discovery
- Corrected driver selection rules
- Renamed driver files to right names
- Code formatting and polishing
- Minor bug fixes
- Improved Credential check
- Improved log reporting
parent 81aa9b8e

src/device/service/drivers/__init__.py

+2 −2
Original line number Diff line number Diff line
@@ -100,11 +100,11 @@ DRIVERS.append( 
        }

    ]))



from .ietf_slice.driver import IetfSliceDriver # pylint: disable=wrong-import-position

from .ietf_slice.IetfSliceDriver import IetfSliceDriver # pylint: disable=wrong-import-position

DRIVERS.append(

    (IetfSliceDriver, [

        {

            FilterFieldEnum.DEVICE_TYPE: DeviceTypeEnum.IETF_SLICE,

            FilterFieldEnum.DEVICE_TYPE: DeviceTypeEnum.TERAFLOWSDN_CONTROLLER,

            FilterFieldEnum.DRIVER: DeviceDriverEnum.DEVICEDRIVER_IETF_SLICE,

        }

    ]))

src/device/service/drivers/ietf_slice/driver.py→src/device/service/drivers/ietf_slice/IetfSliceDriver.py

+90 −128
Original line number Diff line number Diff line
@@ -12,38 +12,19 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



import json

import logging

import re

import threading

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



import anytree

import requests

from requests.auth import HTTPBasicAuth



import anytree, json, logging, re, threading

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

from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method

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

from device.service.driver_api._Driver import (

    RESOURCE_ENDPOINTS,

    RESOURCE_SERVICES,

    _Driver,

)

from device.service.driver_api.AnyTreeTools import (

    TreeNode,

    dump_subtree,

    get_subnode,

    set_subnode_value,

)

from device.service.driver_api.ImportTopologyEnum import (

    ImportTopologyEnum,

    get_import_topology,

)



from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS, RESOURCE_SERVICES

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

from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum, get_import_topology

from .Constants import SPECIAL_RESOURCE_MAPPINGS

from .tfs_slice_nbi_client import TfsApiClient

from .TfsApiClient import TfsApiClient

from .Tools import compose_resource_endpoint





LOGGER = logging.getLogger(__name__)
@@ -52,11 +33,12 @@ ALL_RESOURCE_KEYS = [ 
    RESOURCE_SERVICES,

]



RE_IETF_SLICE_DATA = re.compile(r"^\/service\[[^\]]+\]\/IETFSlice$")

RE_IETF_SLICE_OPERATION = re.compile(r"^\/service\[[^\]]+\]\/IETFSlice\/operation$")



DRIVER_NAME = "ietf_slice"

METRICS_POOL = MetricsPool("Device", "Driver", labels={"driver": DRIVER_NAME})

RE_IETF_SLICE_DATA = re.compile(r'^\/service\[[^\]]+\]\/IETFSlice$')

RE_IETF_SLICE_OPERATION = re.compile(r'^\/service\[[^\]]+\]\/IETFSlice\/operation$')



DRIVER_NAME = 'ietf_slice'

METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})





class IetfSliceDriver(_Driver):
@@ -65,30 +47,24 @@ class IetfSliceDriver(_Driver): 
        self.__lock = threading.Lock()

        self.__started = threading.Event()

        self.__terminate = threading.Event()

        self.__running = TreeNode(".")

        scheme = self.settings.get("scheme", "http")

        username = self.settings.get("username")

        password = self.settings.get("password")

        self.__running = TreeNode('.')

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

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

        scheme   = self.settings.get('scheme', 'http')

        timeout  = int(self.settings.get('timeout', 60))

        self.tac = TfsApiClient(

            self.address,

            self.port,

            scheme=scheme,

            username=username,

            password=password,

        )

        self.__auth = None

        # (

        #     HTTPBasicAuth(username, password)

        #     if username is not None and password is not None

        #     else None

        # )

        self.__tfs_nbi_root = "{:s}://{:s}:{:d}".format(

            scheme, self.address, int(self.port)

        )

        self.__timeout = int(self.settings.get("timeout", 120))

        self.__import_topology = get_import_topology(

            self.settings, default=ImportTopologyEnum.DEVICES

            self.address, self.port, scheme=scheme, username=username,

            password=password, timeout=timeout

        )



        # Options are:

        #    disabled --> just import endpoints as usual

        #    devices  --> imports sub-devices but not links connecting them.

        #                 (a remotely-controlled transport domain might exist between them)

        #    topology --> imports sub-devices and links connecting them.

        #                 (not supported by XR driver)

        self.__import_topology = get_import_topology(self.settings, default=ImportTopologyEnum.DEVICES)



        endpoints = self.settings.get("endpoints", [])

        endpoint_resources = []

        for endpoint in endpoints:
@@ -115,7 +91,7 @@ class IetfSliceDriver(_Driver): 
                    resource_key, resource_value = resource

                    chk_string(str_resource_name, resource_key, allow_empty=False)

                    resource_path = resource_key.split("/")

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

                except Exception as e:

                    LOGGER.exception(

                        "Exception validating {:s}: {:s}".format(

                            str_resource_name, str(resource_key)
@@ -137,22 +113,11 @@ class IetfSliceDriver(_Driver): 
        return results



    def Connect(self) -> bool:

        url = self.__tfs_nbi_root + "/restconf/data/ietf-network-slice-service:ietf-nss"

        with self.__lock:

            if self.__started.is_set():

                return True

            try:

                # requests.get(url, timeout=self.__timeout)

                ...

            except requests.exceptions.Timeout:

                LOGGER.exception("Timeout connecting {:s}".format(url))

                return False

            except Exception:  # pylint: disable=broad-except

                LOGGER.exception("Exception connecting {:s}".format(url))

                return False

            else:

                self.__started.set()

                return True

            if self.__started.is_set(): return True

            checked = self.tac.check_credentials(raise_if_fail=False)

            if checked: self.__started.set()

            return checked



    def Disconnect(self) -> bool:

        with self.__lock:
@@ -168,37 +133,35 @@ class IetfSliceDriver(_Driver): 
    def GetConfig(

        self, resource_keys : List[str] = []

    ) -> List[Tuple[str, Union[Any, None, Exception]]]:

        chk_type("resources", resource_keys, list)

        chk_type('resources', resource_keys, list)

        results = []

        with self.__lock:

            self.tac.check_credentials()

            if len(resource_keys) == 0:

                return dump_subtree(self.__running)

            results = []

            resolver = anytree.Resolver(pathattr="name")

            resolver = anytree.Resolver(pathattr='name')

            for i, resource_key in enumerate(resource_keys):

                str_resource_name = "resource_key[#{:d}]".format(i)

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

                try:

                    chk_string(str_resource_name, resource_key, allow_empty=False)

                    if resource_key == RESOURCE_ENDPOINTS:

                        # return endpoints through TFS NBI API and list-devices method

                        results.extend(self.tac.get_devices_endpoints(self.__import_topology))

                    else:

                        resource_key = SPECIAL_RESOURCE_MAPPINGS.get(

                            resource_key, resource_key

                        )

                    resource_path = resource_key.split("/")

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

                    LOGGER.exception(

                        "Exception validating {:s}: {:s}".format(

                            str_resource_name, str(resource_key)

                        )

                    )

                    results.append(

                        (resource_key, e)

                    )  # if validation fails, store the exception

                    continue

                        resource_path = resource_key.split('/')

                        resource_node = get_subnode(

                            resolver, self.__running, resource_path, default=None

                        )

                        # if not found, resource_node is None

                if resource_node is None:

                    continue

                        if resource_node is None: continue

                        results.extend(dump_subtree(resource_node))

                except Exception as e:

                    MSG = 'Unhandled error processing {:s}: resource_key({:s})'

                    LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))

                    results.append((resource_key, e))

        return results



    @metered_subclass_method(METRICS_POOL)
@@ -206,58 +169,52 @@ class IetfSliceDriver(_Driver): 
        self, resources : List[Tuple[str, Any]]

    ) -> List[Union[bool, Exception]]:

        results = []



        if len(resources) == 0:

            return results



        if len(resources) == 0: return results

        with self.__lock:

            for resource in resources:

                resource_key, resource_value = resource

                if RE_IETF_SLICE_OPERATION.match(resource_key):

                    operation_type = json.loads(resource_value)["type"]

                    operation_type = json.loads(resource_value)['type']

                    results.append((resource_key, True))

                    break

            else:

                raise Exception("operation type not found in resources")

            for resource in resources:

                LOGGER.info("resource = {:s}".format(str(resource)))

                raise Exception('operation type not found in resources')



            for i, resource in enumerate(resources):

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

                LOGGER.info('resource = {:s}'.format(str(resource)))

                resource_key, resource_value = resource

                if not RE_IETF_SLICE_DATA.match(resource_key):

                    continue

                try:

                    resource_value = json.loads(resource_value)



                    slice_name = resource_value["network-slice-services"][

                        "slice-service"

                    ][0]["id"]

                    slice_data = resource_value['network-slice-services'][

                        'slice-service'

                    ][0]

                    slice_name = slice_data['id']



                    if operation_type == "create":

                    if operation_type == 'create':

                        self.tac.create_slice(resource_value)



                    elif operation_type == "update":

                        connection_groups = resource_value["network-slice-services"][

                            "slice-service"

                        ][0]["connection-groups"]["connection-group"]



                    elif operation_type == 'update':

                        connection_groups = slice_data['connection-groups']['connection-group']

                        if len(connection_groups) != 1:

                            raise Exception("only one connection group is supported")



                            MSG = 'Exactly one ConnectionGroup({:s}) is supported'

                            raise Exception(MSG.format(str(connection_groups)))

                        connection_group = connection_groups[0]



                        self.tac.update_slice(

                            slice_name, connection_group["id"], connection_group

                            slice_name, connection_group['id'], connection_group

                        )



                    elif operation_type == "delete":

                    elif operation_type == 'delete':

                        self.tac.delete_slice(slice_name)

                    else:

                        MSG = 'OperationType({:s}) not supported'

                        raise Exception(MSG.format(str(operation_type)))



                    results.append((resource_key, True))

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

                    LOGGER.exception(

                        "Unhandled error processing resource_key({:s})".format(

                            str(resource_key)

                        )

                    )

                except Exception as e:

                    MSG = 'Unhandled error processing {:s}: resource_key({:s})'

                    LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))

                    results.append((resource_key, e))

        return results
@@ -266,22 +223,27 @@ class IetfSliceDriver(_Driver): 
        self, resources : List[Tuple[str, Any]]

    ) -> List[Union[bool, Exception]]:

        results = []



        if len(resources) == 0:

            return results



        with self.__lock:

            for resource in resources:

                LOGGER.info("resource = {:s}".format(str(resource)))

            for i, resource in enumerate(resources):

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

                LOGGER.info('resource = {:s}'.format(str(resource)))

                resource_key, resource_value = resource



                if not RE_IETF_SLICE_DATA.match(resource_key):

                    continue



                try:

                    resource_value = json.loads(resource_value)

                    slice_name = resource_value['network-slice-services'][

                        'slice-service'

                    ][0]['id']

                    self.tac.delete_slice(slice_name)

                    results.append((resource_key, True))

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

                    LOGGER.exception(

                        "Unhandled error processing resource_key({:s})".format(

                            str(resource_key)

                        )

                    )

                except Exception as e:

                    MSG = 'Unhandled error processing {:s}: resource_key({:s})'

                    LOGGER.exception(MSG.format(str_resource_name, str(resource_key)))

                    results.append((resource_key, e))

        return results
@@ -289,18 +251,18 @@ class IetfSliceDriver(_Driver): 
    def SubscribeState(

        self, subscriptions : List[Tuple[str, float, float]]

    ) -> List[Union[bool, Exception]]:

        # TODO: IETF Slice does not support monitoring by now

        # TODO: does not support monitoring by now

        return [False for _ in subscriptions]



    @metered_subclass_method(METRICS_POOL)

    def UnsubscribeState(

        self, subscriptions : List[Tuple[str, float, float]]

    ) -> List[Union[bool, Exception]]:

        # TODO: IETF Slice does not support monitoring by now

        # TODO: does not support monitoring by now

        return [False for _ in subscriptions]



    def GetState(

        self, blocking=False, terminate : Optional[threading.Event] = None

    ) -> Iterator[Tuple[float, str, Any]]:

        # TODO: IETF Slice does not support monitoring by now

        # TODO: does not support monitoring by now

        return []

src/device/service/drivers/ietf_slice/TfsApiClient.py

0 → 100644
+209 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (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, requests

from typing import Dict, List, Optional

from common.tools.client.RestApiClient import RestApiClient

from device.service.driver_api.ImportTopologyEnum import ImportTopologyEnum





GET_CONTEXT_IDS_URL = '/tfs-api/context_ids'

GET_DEVICES_URL     = '/tfs-api/devices'

GET_LINKS_URL       = '/tfs-api/links'





IETF_SLICE_ALL_URL  = '/restconf/data/ietf-network-slice-service:network-slice-services'

IETF_SLICE_ONE_URL  = IETF_SLICE_ALL_URL + '/slice-service={:s}'

IETF_SLICE_CG_URL   = IETF_SLICE_ONE_URL + '/connection-groups/connection-group={:s}'





MAPPING_STATUS = {

    'DEVICEOPERATIONALSTATUS_UNDEFINED': 0,

    'DEVICEOPERATIONALSTATUS_DISABLED' : 1,

    'DEVICEOPERATIONALSTATUS_ENABLED'  : 2,

}



MAPPING_DRIVER = {

    'DEVICEDRIVER_UNDEFINED'            : 0,

    'DEVICEDRIVER_OPENCONFIG'           : 1,

    'DEVICEDRIVER_TRANSPORT_API'        : 2,

    'DEVICEDRIVER_P4'                   : 3,

    'DEVICEDRIVER_IETF_NETWORK_TOPOLOGY': 4,

    'DEVICEDRIVER_ONF_TR_532'           : 5,

    'DEVICEDRIVER_XR'                   : 6,

    'DEVICEDRIVER_IETF_L2VPN'           : 7,

    'DEVICEDRIVER_GNMI_OPENCONFIG'      : 8,

    'DEVICEDRIVER_OPTICAL_TFS'          : 9,

    'DEVICEDRIVER_IETF_ACTN'            : 10,

    'DEVICEDRIVER_OC'                   : 11,

    'DEVICEDRIVER_QKD'                  : 12,

    'DEVICEDRIVER_IETF_L3VPN'           : 13,

    'DEVICEDRIVER_IETF_SLICE'           : 14,

    'DEVICEDRIVER_NCE'                  : 15,

    'DEVICEDRIVER_SMARTNIC'             : 16,

    'DEVICEDRIVER_MORPHEUS'             : 17,

    'DEVICEDRIVER_RYU'                  : 18,

}





LOGGER = logging.getLogger(__name__)





class TfsApiClient(RestApiClient):

    def __init__(

        self, address : str, port : int, scheme : str = 'http',

        username : Optional[str] = None, password : Optional[str] = None,

        timeout : Optional[int] = 30

    ) -> None:

        super().__init__(

            address, port, scheme=scheme, username=username, password=password,

            timeout=timeout, verify_certs=False, allow_redirects=True, logger=LOGGER

        )





    def check_credentials(self, raise_if_fail : bool = True) -> None:

        try:

            LOGGER.info('Checking credentials...')

            self.get(GET_CONTEXT_IDS_URL, expected_status_codes={requests.codes['OK']})

            LOGGER.info('Credentials checked')

            return True

        except requests.exceptions.Timeout as e:

            MSG = 'Timeout connecting {:s}'

            msg = MSG.format(GET_CONTEXT_IDS_URL)

            LOGGER.exception(msg)

            if raise_if_fail: raise Exception(msg) from e

            return False

        except Exception as e:

            MSG = 'Exception connecting credentials: {:s}'

            msg = MSG.format(GET_CONTEXT_IDS_URL)

            LOGGER.exception(msg)

            if raise_if_fail: raise Exception(msg) from e

            return False





    def get_devices_endpoints(

        self, import_topology : ImportTopologyEnum = ImportTopologyEnum.DEVICES

    ) -> List[Dict]:

        LOGGER.debug('[get_devices_endpoints] begin')

        MSG = '[get_devices_endpoints] import_topology={:s}'

        LOGGER.debug(MSG.format(str(import_topology)))



        if import_topology == ImportTopologyEnum.DISABLED:

            MSG = 'Unsupported import_topology mode: {:s}'

            raise Exception(MSG.format(str(import_topology)))



        devices = self.get(GET_DEVICES_URL, expected_status_codes={requests.codes['OK']})



        result = list()

        for json_device in devices['devices']:

            device_uuid : str = json_device['device_id']['device_uuid']['uuid']

            device_type : str = json_device['device_type']

            #if not device_type.startswith('emu-'): device_type = 'emu-' + device_type

            device_status = json_device['device_operational_status']



            ctrl_id : Dict[str, Dict] = json_device.get('controller_id', dict())

            ctrl_uuid : Optional[str] = ctrl_id.get('device_uuid', dict()).get('uuid')



            device_url = '/devices/device[{:s}]'.format(device_uuid)

            device_data = {

                'uuid': json_device['device_id']['device_uuid']['uuid'],

                'name': json_device['name'],

                'type': device_type,

                'status': MAPPING_STATUS[device_status],

                'drivers': [

                    MAPPING_DRIVER[driver]

                    for driver in json_device['device_drivers']

                ],

            }

            if ctrl_uuid is not None and len(ctrl_uuid) > 0:

                device_data['ctrl_uuid'] = ctrl_uuid

            result.append((device_url, device_data))



            for json_endpoint in json_device['device_endpoints']:

                endpoint_uuid = json_endpoint['endpoint_id']['endpoint_uuid']['uuid']

                endpoint_url = '/endpoints/endpoint[{:s}]'.format(endpoint_uuid)

                endpoint_data = {

                    'device_uuid': device_uuid,

                    'uuid': endpoint_uuid,

                    'name': json_endpoint['name'],

                    'type': json_endpoint['endpoint_type'],

                }

                result.append((endpoint_url, endpoint_data))



        if import_topology == ImportTopologyEnum.DEVICES:

            LOGGER.debug('[get_devices_endpoints] devices only; returning')

            return result



        links = self.get(GET_LINKS_URL, expected_status_codes={requests.codes['OK']})



        for json_link in links['links']:

            link_uuid : str = json_link['link_id']['link_uuid']['uuid']

            link_url = '/links/link[{:s}]'.format(link_uuid)

            link_endpoint_ids = [

                (

                    json_endpoint_id['device_id']['device_uuid']['uuid'],

                    json_endpoint_id['endpoint_uuid']['uuid'],

                )

                for json_endpoint_id in json_link['link_endpoint_ids']

            ]

            link_data = {

                'uuid': json_link['link_id']['link_uuid']['uuid'],

                'name': json_link['name'],

                'endpoints': link_endpoint_ids,

            }

            result.append((link_url, link_data))



        LOGGER.debug('[get_devices_endpoints] topology; returning')

        return result





    def create_slice(self, data : Dict) -> None:

        MSG = '[create_slice] data={:s}'

        LOGGER.debug(MSG.format(str(data)))

        try:

            MSG = '[create_slice] POST {:s}: {:s}'

            LOGGER.info(MSG.format(str(IETF_SLICE_ALL_URL), str(data)))

            self.post(IETF_SLICE_ALL_URL, body=data)

        except requests.exceptions.ConnectionError as e:

            MSG = 'Failed to send POST request to TFS IETF Slice NBI'

            raise Exception(MSG) from e





    def update_slice(

        self, slice_name : str, connection_group_id : str,

        updated_connection_group_data : Dict

    ) -> None:

        MSG = '[update_slice] slice_name={:s} connection_group_id={:s} updated_connection_group_data={:s}'

        LOGGER.debug(MSG.format(str(slice_name), str(connection_group_id), str(updated_connection_group_data)))

        url = IETF_SLICE_CG_URL.format(slice_name, connection_group_id)

        try:

            MSG = '[update_slice] PUT {:s}: {:s}'

            LOGGER.info(MSG.format(str(url), str(updated_connection_group_data)))

            self.put(url, body=updated_connection_group_data)

        except requests.exceptions.ConnectionError as e:

            MSG = 'Failed to send PUT request to TFS IETF Slice NBI'

            raise Exception(MSG) from e





    def delete_slice(self, slice_name : str) -> None:

        MSG = '[delete_slice] slice_name={:s}'

        LOGGER.debug(MSG.format(str(slice_name)))

        url = IETF_SLICE_ONE_URL.format(slice_name)

        try:

            MSG = '[delete_slice] DELETE {:s}'

            LOGGER.info(MSG.format(str(url)))

            self.delete(url)

        except requests.exceptions.ConnectionError as e:

            MSG = 'Failed to send DELETE request to TFS IETF Slice NBI'

            raise Exception(MSG) from e

src/device/service/drivers/ietf_slice/tfs_slice_nbi_client.py

deleted100644 → 0
+0 −76

File deleted.

Preview size limit exceeded, changes collapsed.