OCDriver.py

# Copyright 2022-2024 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 time
import json
import logging, pytz, re, threading
from typing import Any, List, Tuple, Union
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.schedulers.background import BackgroundScheduler
from ncclient.manager import Manager, connect_ssh
from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
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 device.service.driver_api._Driver import _Driver
from .templates import compose_config, cli_compose_config, ufi_interface, cisco_interface
from .templates.VPN.common import seperate_port_config
from .templates.VPN.roadms import (
    create_optical_band, disable_media_channel, delete_optical_band, create_media_channel_v2
)
from .templates.VPN.transponder import edit_optical_channel, change_optical_channel_status
from .RetryDecorator import retry
from context.client.ContextClient import ContextClient
from common.proto.context_pb2 import OpticalConfig
from .templates.discovery_tool.transponders import  transponder_values_extractor
from .templates.discovery_tool.roadms import roadm_values_extractor
from .templates.discovery_tool.open_roadm import openroadm_values_extractor
from .templates.VPN.openroadm import network_media_channel_handler

DEBUG_MODE = False
logging.getLogger('ncclient.manager').setLevel(logging.DEBUG if DEBUG_MODE else logging.WARNING)
logging.getLogger('ncclient.transport.ssh').setLevel(logging.DEBUG if DEBUG_MODE else logging.WARNING)
logging.getLogger('apscheduler.executors.default').setLevel(logging.INFO if DEBUG_MODE else logging.ERROR)
logging.getLogger('apscheduler.scheduler').setLevel(logging.INFO if DEBUG_MODE else logging.ERROR)
logging.getLogger('monitoring-client').setLevel(logging.INFO if DEBUG_MODE else logging.ERROR)

RE_GET_ENDPOINT_FROM_INTERFACE_KEY = re.compile(r'.*interface\[([^\]]+)\].*')
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'
transponder_filter_fields = ["frequency", "target-output-power", "operational-mode", "line-port", "admin-state"]
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')
context_client= ContextClient()
port_xml_filter=f"/components/component[state[type='oc-platform-types:PORT']]/*"
transceiver_xml_filter="/components/component[state[type='oc-platform-types:TRANSCEIVER']]/*"

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.__version          = settings.get('version', "1")
        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_rule  = settings.get('commit_per_rule', False)
        self.__device_params    = settings.get('device_params', {})
        self.__manager_params   = settings.get('manager_params', {})
        self.__nc_params        = settings.get('nc_params', {})
        self.__message_renderer = settings.get('message_renderer','jinja')
      
        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_rule 

    @property
    def vendor(self): return self.__vendor

    @property
    def version(self): return self.__version

    @property
    def message_renderer(self): return self.__message_renderer

    @RETRY_DECORATOR
    def get(self, filter=None, with_defaults=None): # pylint: disable=redefined-builtin
        with self.__lock:
            config=self.__manager.get(filter=filter, with_defaults=with_defaults)
            return config

    @RETRY_DECORATOR
    def edit_config(
        self, config, target='running', default_operation=None, test_option=None,
        error_option=None, format='xml'                                             # pylint: disable=redefined-builtin
    ):
        response = None
        with self.__lock:
            response= self.__manager.edit_config(
                config, target=target, default_operation=default_operation, test_option=test_option,
                error_option=error_option, format=format)

        str_respones = str(response)  
        if re.search(r'<ok/>', str_respones):
           return True
        return False

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

DRIVER_NAME = 'oc'
METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
def edit_config(                                                                                                            # edit the configuration of openconfig devices
    netconf_handler : NetconfSessionHandler, logger : logging.Logger, resources : List[Tuple[str, Any]]
    ,conditions, delete=False,
    commit_per_rule=False, target='running', default_operation='merge', test_option=None, error_option=None,
    format='xml' 
):
    str_method = 'DeleteConfig' if delete else 'SetConfig'
    results = []

    logging.info(f"commmit per rule {commit_per_rule}")
    str_config_messages=[]
    if str_method == 'SetConfig':
        if (conditions['edit_type']=='optical-channel'):
            #transponder
            str_config_messages =  edit_optical_channel(resources)
        elif (conditions['edit_type']=='optical-band'):
            #roadm optical-band
            str_config_messages = create_optical_band(resources)
        elif (conditions['edit_type']=='network-media-channel'):
            commit_per_rule=True
            #openroadm network media channel
            str_config_messages = network_media_channel_handler(resources)       
        else :
            #roadm media-channel
            str_config_messages=create_media_channel_v2(resources)
    #Disabling of the Configuration         
    else:
        # Device type is Transponder
        if (conditions['edit_type'] == "optical-channel"):
            _,ports,_=seperate_port_config(resources)
            str_config_messages=change_optical_channel_status(state="DISABLED",ports=ports)
            
        #  Device type is Roadm     
        elif (conditions['edit_type']=='optical-band'):    
            str_config_messages=delete_optical_band(resources)
        else :
            str_config_messages=disable_media_channel(resources)

    for str_config_message in str_config_messages:  
        # configuration of the received templates 
        if str_config_message is None: raise UnsupportedResourceKeyException("CONFIG")
        result= netconf_handler.edit_config(                                                                               # configure the device
            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()                                                                               # configuration commit

        #results[i] = True
            results.append(result)
    
    if netconf_handler.vendor == "CISCO":
        if "L2VSI" in resources[0][1]:
            #Configure by CLI
            logger.warning("CLI Configuration")
            cli_compose_config(resources, delete=delete, host= netconf_handler._NetconfSessionHandler__address, user=netconf_handler._NetconfSessionHandler__username, passw=netconf_handler._NetconfSessionHandler__password)        
            for i,resource in enumerate(resources):
                results.append(True)
        else: 
            logger.warning("CLI Configuration CISCO INTERFACE")
            cisco_interface(resources, delete=delete, host= netconf_handler._NetconfSessionHandler__address, user=netconf_handler._NetconfSessionHandler__username, passw=netconf_handler._NetconfSessionHandler__password)        
            for i,resource in enumerate(resources):
                results.append(True)
    elif netconf_handler.vendor == "UFISPACE": 
        #Configure by CLI
        logger.warning("CLI Configuration: {:s}".format(resources))
        ufi_interface(resources, delete=delete, host= netconf_handler._NetconfSessionHandler__address, user=netconf_handler._NetconfSessionHandler__username, passw=netconf_handler._NetconfSessionHandler__password)        
        for i,resource in enumerate(resources):
            results.append(True)
    else:
        for i,resource in enumerate(resources):
            str_resource_name = 'resources[#{:d}]'.format(i)
            try:
                logger.debug('[{: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_messages = compose_config(                                                                          # get template for configuration
                    resource_key, resource_value, delete=delete, vendor=netconf_handler.vendor, message_renderer=netconf_handler.message_renderer)
                for str_config_message in str_config_messages:                                                                 # configuration of the received templates 
                    if str_config_message is None: raise UnsupportedResourceKeyException(resource_key)
                    logger.debug('[{:s}] str_config_message[{:d}] = {:s}'.format(
                    str_method, len(str_config_message), str(str_config_message)))
                    netconf_handler.edit_config(                                                                               # configure the device
                        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()                                                                               # configuration commit
                    if 'table_connections' in resource_key:
                        time.sleep(5) # CPU usage might exceed critical level after route redistribution, BGP daemon needs time to reload
                
                #results[i] = True
                results.append(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
                results.append(e)

        if not commit_per_rule:
            try:
                netconf_handler.commit()
            except Exception as e: # pylint: disable=broad-except
                msg = '[{:s}] Exception committing: {:s}'
                str_operation = 'preparing' if target == 'candidate' else ('deleting' if delete else 'setting')
                logger.exception(msg.format(str_method, str_operation, str(resources)))
                results = [e for _ in resources] # if commit fails, set exception in each resource

    return results         

class OCDriver(_Driver):
    def __init__(self, address : str, port : int,device_uuid=None, **settings) -> None:
        super().__init__(DRIVER_NAME, address, port, **settings)
        self.__logger = logging.getLogger('{:s}:[{:s}:{:s}]'.format(str(__name__), str(self.address), str(self.port)))
        self.__lock = threading.Lock()

        self.__started = threading.Event()
        self.__terminate = threading.Event()
        self.__scheduler = BackgroundScheduler(daemon=True)
        self.__scheduler.configure(
            jobstores = {'default': MemoryJobStore()},
            executors = {'default': ThreadPoolExecutor(max_workers=1)}, 
            job_defaults = {'coalesce': False, 'max_instances': 3},
            timezone=pytz.utc)
        self._temp_address=f"{address}{port}"
       
        self.__netconf_handler = NetconfSessionHandler(self.address, self.port, **(self.settings))
        self.__type = self.settings.get("type","optical-transponder")
        self.__device_uuid = device_uuid
        self.Connect()
        
    def Connect(self) -> bool:
        with self.__lock:
            if self.__started.is_set(): return True
            self.__netconf_handler.connect()
            self.__scheduler.start()
            self.__started.set()
            return True

    def Disconnect(self) -> bool:
        with self.__lock:
            self.__terminate.set()
            if not self.__started.is_set(): return True
            self.__scheduler.shutdown()
            self.__netconf_handler.disconnect()
            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 = []
        opticalConfig= OpticalConfig()

        with self.__lock:
            config = {}
            transceivers = {}
            oc_values = {}
            ports_result = []
            oc_values["type"] = self.__type
            try:
                xml_data = self.__netconf_handler.get().data_xml
                if self.__type == "optical-transponder":
                    extracted_values = transponder_values_extractor(
                        data_xml=xml_data, resource_keys=transponder_filter_fields, dic=config
                    )
                    transceivers, optical_channels_params, channel_namespace, endpoints, ports_result = extracted_values
                    oc_values["channels"         ] = optical_channels_params
                    oc_values["transceivers"     ] = transceivers
                    oc_values["channel_namespace"] = channel_namespace
                    oc_values["endpoints"        ] = endpoints
                    oc_values["ports"            ] = ports_result
                elif (self.__type =='openroadm'):
                   extracted_values=openroadm_values_extractor(data_xml=xml_data, resource_keys=[], dic=oc_values)
                   ports_result = extracted_values[1]
                   oc_values['interfaces'] = extracted_values[0]['interfaces']
                else:
                    extracted_values = roadm_values_extractor(data_xml=xml_data, resource_keys=[], dic=config)
                    ports_result = extracted_values[0]
                    oc_values['optical_bands' ] = extracted_values[1]
                    oc_values['media_channels'] = extracted_values[2]

                #results.append((resource_key, e)) # if validation fails, store the exception    

            #///////////////////////// store optical configurtaion  ////////////////////////////////////////////////////////

                opticalConfig.config=json.dumps(oc_values)
                if self.__device_uuid is not None:
                    opticalConfig.device_id.device_uuid.uuid=self.__device_uuid
                results.append((f"/opticalconfigs/opticalconfig/{self.__device_uuid}",{"opticalconfig":opticalConfig}))
                # context_client.connect()    
                # config_id=context_client.SetOpticalConfig(opticalConfig)
                # context_client.close()
            except Exception as e: # pylint: disable=broad-except
                MSG = 'Exception retrieving {:s}'
                self.__logger.info("error from getConfig %s",e)
                self.__logger.exception(MSG.format(e))

        if len(ports_result) > 0: results.extend(ports_result)
        return results

    @metered_subclass_method(METRICS_POOL)
    def SetConfig(self, resources : List[Tuple[str, Any]],conditions:dict) -> List[Union[bool, Exception]]:
        if len(resources) == 0: return []
        results = []
        with self.__lock:
            if self.__netconf_handler.use_candidate:
                with self.__netconf_handler.locked(target='candidate'):
                    results = edit_config(
                        self.__netconf_handler, self.__logger, resources, conditions, target='candidate',
                        commit_per_rule=self.__netconf_handler.commit_per_rule
                    )
            else:
                results = edit_config(
                    self.__netconf_handler, self.__logger, resources, conditions=conditions
                )
        return results

    @metered_subclass_method(METRICS_POOL)
    def DeleteConfig(
        self, resources : List[Tuple[str, Any]], conditions : dict,
        optical_device_configuration = None
    ) -> List[Union[bool, Exception]]:
        chk_type('resources', resources, list)
        if len(resources) == 0: return []

        with self.__lock:
            if self.__netconf_handler.use_candidate:
                with self.__netconf_handler.locked(target='candidate'):
                    results = edit_config(
                        self.__netconf_handler, self.__logger, resources, target='candidate', delete=True,
                        commit_per_rule=self.__netconf_handler.commit_per_rule, conditions=conditions
                    )
            else:
                results = edit_config(
                    self.__netconf_handler, self.__logger, resources, delete=True, conditions=conditions
                )
        return results