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

Device component - IETF L3VPN Driver: 

- Implemented subscriptions
- Code cleanup
parent 6ca22402

src/device/service/drivers/ietf_l3vpn/IetfL3VpnDriver.py

+55 −10
Original line number Diff line number Diff line
@@ -14,12 +14,16 @@ 




import anytree, json, logging, re, threading

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

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

from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method

from common.tools.rest_conf.client.RestConfClient import RestConfClient

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

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 .handlers.SubscriptionHandler import (

    SubscribedNotificationsSchema, SubscriptionHandler, UnsubscribedNotificationsSchema

)

from .Constants import SPECIAL_RESOURCE_MAPPINGS

from .TfsApiClient import TfsApiClient

from .Tools import compose_resource_endpoint
@@ -37,6 +41,7 @@ ALL_RESOURCE_KEYS = [ 
RE_IETF_L3VPN_DATA = re.compile(r'^\/service\[[^\]]+\]\/IETFL3VPN$')

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





DRIVER_NAME = 'ietf_l3vpn'

METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
@@ -57,6 +62,13 @@ class IetfL3VpnDriver(_Driver): 
            password=password, timeout=timeout

        )



        restconf_settings = copy.deepcopy(settings)

        restconf_settings.pop('base_url', None)

        restconf_settings.pop('import_topology', None)

        restconf_settings['logger'] = logging.getLogger(__name__ + '.RestConfClient')

        self._rest_conf_client = RestConfClient(address, port=port, **restconf_settings)

        self._handler_subscription = SubscriptionHandler(self._rest_conf_client)



        # Options are:

        #    disabled --> just import endpoints as usual

        #    devices  --> imports sub-devices but not links connecting them.
@@ -119,16 +131,19 @@ class IetfL3VpnDriver(_Driver): 
            if checked: self.__started.set()

            return checked





    def Disconnect(self) -> bool:

        with self.__lock:

            self.__terminate.set()

            return True





    @metered_subclass_method(METRICS_POOL)

    def GetInitialConfig(self) -> List[Tuple[str, Any]]:

        with self.__lock:

            return []





    @metered_subclass_method(METRICS_POOL)

    def GetConfig(

        self, resource_keys : List[str] = []
@@ -165,6 +180,7 @@ class IetfL3VpnDriver(_Driver): 
                    results.append((resource_key, e))

        return results





    @metered_subclass_method(METRICS_POOL)

    def SetConfig(

        self, resources : List[Tuple[str, Any]]
@@ -174,9 +190,11 @@ class IetfL3VpnDriver(_Driver): 
        with self.__lock:

            for i, resource in enumerate(resources):

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

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

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

                resource_key, resource_value = resource



                if not RE_IETF_L3VPN_DATA.match(resource_key): continue



                try:

                    resource_value = json.loads(resource_value)

                    self.tac.create_connectivity_service(resource_value)
@@ -187,6 +205,7 @@ class IetfL3VpnDriver(_Driver): 
                    results.append((resource_key, e))

        return results





    @metered_subclass_method(METRICS_POOL)

    def DeleteConfig(

        self, resources : List[Tuple[str, Any]]
@@ -196,9 +215,11 @@ class IetfL3VpnDriver(_Driver): 
        with self.__lock:

            for i, resource in enumerate(resources):

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

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

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

                resource_key, resource_value = resource



                if not RE_IETF_L3VPN_DATA.match(resource_key): continue



                try:

                    resource_value = json.loads(resource_value)

                    #service_uuid = resource_value['ietf-l3vpn-svc:l3vpn-svc'][
@@ -213,22 +234,46 @@ class IetfL3VpnDriver(_Driver): 
                    results.append((resource_key, e))

        return results





    @metered_subclass_method(METRICS_POOL)

    def SubscribeState(

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

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

        # TODO: does not support monitoring by now

        return [False for _ in subscriptions]

    ) -> List[Union[bool, Dict[str, Any], Exception]]:

        if len(subscriptions) != 1:

            raise ValueError('IETF L3VPN Driver supports only one subscription at a time')

        s = subscriptions[0]

        uri = s[0]

        #sampling_duration = s[1]

        sampling_interval = s[2]

        s_data : SubscribedNotificationsSchema = {

            'ietf-subscribed-notifications:input': {

                'datastore': 'operational',

                'ietf-yang-push:datastore-xpath-filter': uri,

                'ietf-yang-push:periodic': {'ietf-yang-push:period': str(sampling_interval)},

            }

        }

        s_id = self._handler_subscription.subscribe(s_data)

        return [s_id]





    @metered_subclass_method(METRICS_POOL)

    def UnsubscribeState(

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

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

        # TODO: does not support monitoring by now

        return [False for _ in subscriptions]

    ) -> List[Union[bool, Dict[str, Any], Exception]]:

        if len(subscriptions) != 1:

            raise ValueError('IETF L3VPN Driver supports only one subscription at a time')

        s = subscriptions[0]

        identifier = s[0]

        s_data : UnsubscribedNotificationsSchema = {

            'delete-subscription': {

                'identifier': identifier,

            }

        }

        self._handler_subscription.unsubscribe(s_data)

        return [True]





    def GetState(

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

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

        # TODO: does not support monitoring by now

        return []

src/device/service/drivers/ietf_l3vpn/handlers/SubscriptionHandler.py

0 → 100644
+82 −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_extensions import TypedDict

from common.tools.rest_conf.client.RestConfClient import RestConfClient





LOGGER = logging.getLogger(__name__)





Periodic = TypedDict('Periodic', {'ietf-yang-push:period': str})



Input = TypedDict(

    'Input',

    {

        'datastore': str,

        'ietf-yang-push:datastore-xpath-filter': str,

        'ietf-yang-push:periodic': Periodic,

    },

)



SubscribedNotificationsSchema = TypedDict(

    'SubscribedNotificationsSchema', {'ietf-subscribed-notifications:input': Input}

)



SubscriptionSchema = TypedDict('SubscriptionSchema', {'identifier': str})



UnsubscribedNotificationsSchema = TypedDict(

    'UnsubscribedNotificationsSchema', {'delete-subscription': SubscriptionSchema}

)





class SubscriptionId(TypedDict):

    identifier: str

    uri: str





class SubscriptionHandler:

    def __init__(self, rest_conf_client : RestConfClient) -> None:

        self._rest_conf_client = rest_conf_client



    def subscribe(

        self, subscription_data : SubscribedNotificationsSchema

    ) -> SubscriptionId:

        MSG = '[subscribe] subscription_data={:s}'

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

        try:

            url = '/subscriptions:establish-subscription'

            LOGGER.debug('Subscribing to telemetry: {:s}'.format(str(subscription_data)))

            reply = self._rest_conf_client.rpc(url, body=subscription_data)

            LOGGER.debug('Subscription reply: {:s}'.format(str(reply)))

            return reply

        except requests.exceptions.ConnectionError as e:

            MSG = 'Failed to send RPC request'

            raise Exception(MSG) from e



    def unsubscribe(

        self, unsubscription_data : UnsubscribedNotificationsSchema

    ) -> SubscriptionId:

        MSG = '[unsubscribe] unsubscription_data={:s}'

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

        try:

            url = '/subscriptions:delete-subscription'

            LOGGER.debug('Unsubscribing from telemetry: {:s}'.format(str(unsubscription_data)))

            reply = self._rest_conf_client.rpc(url, body=unsubscription_data)

            LOGGER.debug('Unsubscription reply: {:s}'.format(str(reply)))

            return reply

        except requests.exceptions.ConnectionError as e:

            MSG = 'Failed to send RPC request'

            raise Exception(MSG) from e

src/device/service/drivers/ietf_l3vpn/handlers/__init__.py

0 → 100644
+13 −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.