IetfActnDriver.py

# 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 json, logging, requests, 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_string, chk_type
from device.service.driver_api._Driver import _Driver, RESOURCE_ENDPOINTS, RESOURCE_SERVICES
from .handlers.EthtServiceHandler import EthtServiceHandler
from .handlers.OsuTunnelHandler import OsuTunnelHandler
from .handlers.RestApiClient import RestApiClient
from .Tools import get_etht_services, get_osu_tunnels, parse_resource_key

LOGGER = logging.getLogger(__name__)

ALL_RESOURCE_KEYS = [
    RESOURCE_ENDPOINTS,
    RESOURCE_SERVICES,
]

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

class IetfActnDriver(_Driver):
    def __init__(self, address: str, port: int, **settings) -> None:
        super().__init__(DRIVER_NAME, address, port, **settings)
        self.__lock = threading.Lock()
        self.__started = threading.Event()
        self.__terminate = threading.Event()
        self._rest_api_client = RestApiClient(address, port, settings=settings)
        self._handler_osu_tunnel = OsuTunnelHandler(self._rest_api_client)
        self._handler_etht_service = EthtServiceHandler(self._rest_api_client)

    def Connect(self) -> bool:
        with self.__lock:
            if self.__started.is_set(): return True
            try:
                self._rest_api_client.get('Check Credentials', '')
            except requests.exceptions.Timeout:
                LOGGER.exception('Timeout exception checking connectivity')
                return False
            except Exception:  # pylint: disable=broad-except
                LOGGER.exception('Unhandled exception checking connectivity')
                return False
            else:
                self.__started.set()
                return True

    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] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
        chk_type('resources', resource_keys, list)
        results = []
        with self.__lock:
            if len(resource_keys) == 0: resource_keys = ALL_RESOURCE_KEYS
            for i, resource_key in enumerate(resource_keys):
                chk_string('resource_key[#{:d}]'.format(i), resource_key, allow_empty=False)

                try:
                    _results = list()

                    if resource_key == RESOURCE_ENDPOINTS:
                        # Add mgmt endpoint by default
                        resource_key = '/endpoints/endpoint[mgmt]'
                        resource_value = {'uuid': 'mgmt', 'name': 'mgmt', 'type': 'mgmt'}
                        results.append((resource_key, resource_value))
                    elif resource_key == RESOURCE_SERVICES:
                        get_osu_tunnels(self._handler_osu_tunnel, _results)
                        get_etht_services(self._handler_etht_service, _results)
                    else:
                        # check if resource key is for a specific OSU tunnel or ETHT service, and get them accordingly
                        osu_tunnel_name, etht_service_name = parse_resource_key(resource_key)
                        if osu_tunnel_name is not None:
                            get_osu_tunnels(self._handler_osu_tunnel, _results, osu_tunnel_name=osu_tunnel_name)
                        if etht_service_name is not None:
                            get_etht_services(self._handler_etht_service, _results, etht_service_name=etht_service_name)

                    results.extend(_results)
                except Exception as e:
                    results.append((resource_key, e))

        return results

    @metered_subclass_method(METRICS_POOL)
    def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
        results = []
        if len(resources) == 0: return results
        with self.__lock:
            for resource_key, resource_value in resources:
                LOGGER.info('resource: key({:s}) => value({:s})'.format(str(resource_key), str(resource_value)))
                try:
                    _results = list()

                    if isinstance(resource_value, str): resource_value = json.loads(resource_value)
                    osu_tunnel_name, etht_service_name = parse_resource_key(resource_key)

                    if osu_tunnel_name is not None:
                        succeeded = self._handler_osu_tunnel.update(resource_value)
                        _results.append(succeeded)

                    if etht_service_name is not None:
                        succeeded = self._handler_etht_service.update(resource_value)
                        _results.append(succeeded)

                    results.extend(_results)
                except Exception as e:
                    results.append(e)

        return results

    @metered_subclass_method(METRICS_POOL)
    def DeleteConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
        results = []
        if len(resources) == 0: return results
        with self.__lock:
            for resource_key, resource_value in resources:
                LOGGER.info('resource: key({:s}) => value({:s})'.format(str(resource_key), str(resource_value)))
                try:
                    _results = list()

                    if isinstance(resource_value, str): resource_value = json.loads(resource_value)
                    osu_tunnel_name, etht_service_name = parse_resource_key(resource_key)

                    if osu_tunnel_name is not None:
                        succeeded = self._handler_osu_tunnel.delete(osu_tunnel_name)
                        _results.append(succeeded)

                    if etht_service_name is not None:
                        succeeded = self._handler_etht_service.delete(etht_service_name)
                        _results.append(succeeded)

                    results.extend(_results)
                except Exception as e:
                    results.append(e)

        return results

    @metered_subclass_method(METRICS_POOL)
    def SubscribeState(self, subscriptions : List[Tuple[str, float, float]]) -> List[Union[bool, Exception]]:
        # TODO: IETF ACTN 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 ACTN 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 ACTN does not support monitoring by now
        return []