Skip to content
GitLab
Commits (27)
Hide whitespace changes
Inline Side-by-side
src/device/client/DeviceClient.py
View file @ 2673daca
......@@ -15,12 +15,12 @@
import grpc, logging
from common.Constants import ServiceNameEnum
from common.Settings import get_service_host, get_service_port_grpc
from common.proto.context_pb2 import Device, DeviceConfig, DeviceId, Empty
from common.proto.context_pb2 import Device, DeviceConfig, DeviceId, Empty,MyConfig,MyConfigId
from common.proto.device_pb2 import MonitoringSettings
from common.proto.device_pb2_grpc import DeviceServiceStub
from common.tools.client.RetryDecorator import retry, delay_exponential
from common.tools.grpc.Tools import grpc_message_to_json_string
from common.proto.openconfig_device_pb2_grpc import OpenConfigServiceStub
LOGGER = logging.getLogger(__name__)
MAX_RETRIES = 15
DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)
......@@ -34,12 +34,14 @@ class DeviceClient:
LOGGER.debug('Creating channel to {:s}...'.format(str(self.endpoint)))
self.channel = None
self.stub = None
self.openconfig_stub=None
self.connect()
LOGGER.debug('Channel created')
def connect(self):
self.channel = grpc.insecure_channel(self.endpoint)
self.stub = DeviceServiceStub(self.channel)
self.openconfig_stub=OpenConfigServiceStub(self.channel)
def close(self):
if self.channel is not None: self.channel.close()
......@@ -80,3 +82,8 @@ class DeviceClient:
response = self.stub.MonitorDeviceKpi(request)
LOGGER.debug('MonitorDeviceKpi result: {:s}'.format(grpc_message_to_json_string(response)))
return response
def ConfigureOpticalDevice(self, request : MyConfig) -> MyConfigId:
response = self.openconfig_stub.ConfigureOpticalDevice(request)
return response
src/device/service/DeviceService.py
View file @ 2673daca
......@@ -19,6 +19,8 @@ from common.tools.service.GenericGrpcService import GenericGrpcService
from .driver_api.DriverInstanceCache import DriverInstanceCache
from .DeviceServiceServicerImpl import DeviceServiceServicerImpl
from .monitoring.MonitoringLoops import MonitoringLoops
from .OpenConfigServicer import OpenConfigServicer
from common.proto.openconfig_device_pb2_grpc import add_OpenConfigServiceServicer_to_server
# Custom gRPC settings
# Multiple clients might keep connections alive waiting for RPC methods to be executed.
......@@ -31,10 +33,13 @@ class DeviceService(GenericGrpcService):
super().__init__(port, max_workers=GRPC_MAX_WORKERS, cls_name=cls_name)
self.monitoring_loops = MonitoringLoops()
self.device_servicer = DeviceServiceServicerImpl(driver_instance_cache, self.monitoring_loops)
self.openconfig_device_servicer=OpenConfigServicer(driver_instance_cache,self.monitoring_loops)
def install_servicers(self):
self.monitoring_loops.start()
add_DeviceServiceServicer_to_server(self.device_servicer, self.server)
add_OpenConfigServiceServicer_to_server(self.openconfig_device_servicer,self.server)
def stop(self):
super().stop()
......
src/device/service/DeviceServiceServicerImpl.py
View file @ 2673daca
......@@ -20,7 +20,7 @@ from common.Settings import ENVVAR_SUFIX_SERVICE_HOST, get_env_var_name
from common.method_wrappers.Decorator import MetricTypeEnum, MetricsPool, safe_and_metered_rpc_method
from common.method_wrappers.ServiceExceptions import NotFoundException, OperationFailedException
from common.proto.context_pb2 import (
Device, DeviceConfig, DeviceDriverEnum, DeviceId, DeviceOperationalStatusEnum, Empty, Link)
Device, DeviceConfig, DeviceDriverEnum, DeviceId, DeviceOperationalStatusEnum, Empty, Link,MyConfig,MyConfigId)
from common.proto.device_pb2 import MonitoringSettings
from common.proto.device_pb2_grpc import DeviceServiceServicer
from common.tools.context_queries.Device import get_device
......@@ -29,6 +29,7 @@ from context.client.ContextClient import ContextClient
from .driver_api._Driver import _Driver
from .driver_api.DriverInstanceCache import DriverInstanceCache, get_driver
from .monitoring.MonitoringLoops import MonitoringLoops
from .drivers.oc_driver.OCDriver import OCDriver
from .ErrorMessages import ERROR_MISSING_DRIVER, ERROR_MISSING_KPI
from .Tools import (
check_connect_rules, check_no_endpoints, compute_rules_to_add_delete, configure_rules, deconfigure_rules,
......@@ -58,8 +59,9 @@ class DeviceServiceServicerImpl(DeviceServiceServicer):
device_uuid = request.device_id.device_uuid.uuid
connection_config_rules = check_connect_rules(request.device_config)
check_no_endpoints(request.device_endpoints)
if (request.device_drivers[0]!= 9) :
check_no_endpoints(request.device_endpoints)
t1 = time.time()
context_client = ContextClient()
......@@ -138,7 +140,14 @@ class DeviceServiceServicerImpl(DeviceServiceServicer):
else:
# ZTP is not deployed; assume the device is ready while onboarding and set them as enabled.
device.device_operational_status = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED
# temporary line
if (request.device_drivers[0]== 9 and len(request.device_endpoints)>0):
for endpoint in request.device_endpoints:
#endpoint.endpoint_id.device_id.CopyFrom(device.device_id)
pass
device.device_endpoints.extend(request.device_endpoints)
device_id = context_client.SetDevice(device)
t10 = time.time()
......
src/device/service/OpenConfigServicer.py 0 → 100644
View file @ 2673daca
# 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 grpc, logging, os, time
from typing import Dict
from prometheus_client import Histogram
from common.Constants import ServiceNameEnum
from common.Settings import ENVVAR_SUFIX_SERVICE_HOST, get_env_var_name
from common.method_wrappers.Decorator import MetricTypeEnum, MetricsPool, safe_and_metered_rpc_method
from common.method_wrappers.ServiceExceptions import NotFoundException, OperationFailedException
from common.proto.context_pb2 import (
Device, DeviceConfig, DeviceDriverEnum, DeviceId, DeviceOperationalStatusEnum, Empty, Link,MyConfig,MyConfigId ,MyConfig,MyConfigList)
from common.proto.device_pb2 import MonitoringSettings
from common.proto.device_pb2_grpc import DeviceServiceServicer
from common.tools.context_queries.Device import get_device
from common.tools.mutex_queues.MutexQueues import MutexQueues
from context.client.ContextClient import ContextClient
from .driver_api._Driver import _Driver
from .driver_api.DriverInstanceCache import DriverInstanceCache, get_driver
from .monitoring.MonitoringLoops import MonitoringLoops
from .drivers.oc_driver.OCDriver import OCDriver
from .ErrorMessages import ERROR_MISSING_DRIVER, ERROR_MISSING_KPI
from .Tools import extract_resources
from .Tools import (
check_connect_rules, check_no_endpoints, compute_rules_to_add_delete, configure_rules, deconfigure_rules,
get_device_controller_uuid, populate_config_rules, populate_endpoint_monitoring_resources, populate_endpoints,
populate_initial_config_rules, subscribe_kpi, unsubscribe_kpi, update_endpoints)
LOGGER = logging.getLogger(__name__)
METRICS_POOL = MetricsPool('Device', 'RPC')
METRICS_POOL_DETAILS = MetricsPool('Device', 'execution', labels={
'driver': '', 'operation': '', 'step': '',
})
class OpenConfigServicer(DeviceServiceServicer):
def __init__(self, driver_instance_cache : DriverInstanceCache, monitoring_loops : MonitoringLoops) -> None:
LOGGER.debug('Creating Servicer...')
self.driver_instance_cache = driver_instance_cache
self.monitoring_loops = monitoring_loops
self.mutex_queues = MutexQueues()
LOGGER.debug('Servicer Created')
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def AddOpenConfigDevice(self, request : MyConfig, context : grpc.ServicerContext) -> DeviceId:
device_uuid = request.device_id.device_uuid.uuid
device_type=request.device_type
ocdriver=OCDriver()
connection_config_rules = check_connect_rules(request.device_config)
check_no_endpoints(request.device_endpoints)
context_client = ContextClient()
device = get_device(context_client, device_uuid, rw_copy=True)
if device is None:
# not in context, create blank one to get UUID, and populate it below
device = Device()
device.device_id.CopyFrom(request.device_id) # pylint: disable=no-member
device.name = request.name
device.device_type = request.device_type
device.device_operational_status = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_UNDEFINED
device.device_drivers.extend(request.device_drivers) # pylint: disable=no-member
device.device_config.CopyFrom(request.device_config) # pylint: disable=no-member
device_id = context_client.SetDevice(device)
device = get_device(context_client, device_id.device_uuid.uuid, rw_copy=True)
# update device_uuid to honor UUID provided by Context
device_uuid = device.device_id.device_uuid.uuid
LOGGER.debug('device type %s',device)
t2 = time.time()
self.mutex_queues.wait_my_turn(device_uuid)
t3 = time.time()
#temp fix to solve the error
#todo check what to pass here
resources_to_get = []
try:
#driver : _Driver = get_driver(self.driver_instance_cache, device)
results_getconfig=ocdriver.GetConfig(resource_keys=resources_to_get,device_uuid=device_uuid)
#results_getconfig = driver.GetConfig(resources_to_get,device_uuid)
except Exception as error :
LOGGER.debug("error %s",error)
# @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
# def ConfigureOpenConfigDevice(self, request : MyConfig, context : grpc.ServicerContext) -> DeviceId:
# device_id = request.myconfig_id
# device_uuid = device_id.myconfig_uuid
# self.mutex_queues.wait_my_turn(device_uuid)
# t1 = time.time()
# try:
# context_client = ContextClient()
# t2 = time.time()
# device = get_device(
# context_client, device_uuid, rw_copy=True, include_endpoints=False, include_components=False,
# include_config_rules=True)
# if device is None:
# raise NotFoundException('Device', device_uuid, extra_details='loading in ConfigureDevice')
# t3 = time.time()
# device_controller_uuid = get_device_controller_uuid(device)
# if device_controller_uuid is not None:
# device = get_device(
# context_client, device_controller_uuid, rw_copy=True, include_endpoints=False,
# include_components=False, include_config_rules=True)
# if device is None:
# raise NotFoundException(
# 'Device', device_controller_uuid, extra_details='loading in ConfigureDevice')
# device_uuid = device.device_id.device_uuid.uuid
# driver : _Driver = get_driver(self.driver_instance_cache, device)
# resources_to_set, resources_to_delete = compute_rules_to_add_delete(device, request)
# results_setconfig = driver.SetConfig(resources=resources_to_set)
# except Exception as e:
# LOGGER.debug("error in updating %s",e)
'''
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def ConfigureOpticalDevice (self, request : MyConfig, context:grpc.ServicerContext) -> Empty:
LOGGER.info('Updating from ConfigureOpticalDevice %s',request)
device_id = request.myconfig_id
device_uuid = device_id.myconfig_uuid
resources=[]
add_proccess=False
update_interface=False
result=None
config =eval(request.config)
filter_fields= ["frequency","target-output-power","interface","operational-mode"]
try:
context_client = ContextClient()
device = get_device(
context_client, device_uuid, rw_copy=True, include_endpoints=False, include_components=False,
include_config_rules=True)
if device is None:
LOGGER.debug("device is none")
raise NotFoundException('Device', device_uuid, extra_details='loading in ConfigureDevice')
resources,conditions=extract_resources(config=config,device=device)
driver : _Driver = get_driver(self.driver_instance_cache, device)
LOGGER.info("resource %s conditions %s",resources,conditions)
result = driver.SetConfig(resources=resources,conditions=conditions)
driver.GetConfig(resource_keys=filter_fields)
except Exception as e:
LOGGER.info("error in configuring %s",e)
LOGGER.info("result %s",result)
return Empty()
'''
#modified Andrea
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def ConfigureOpticalDevice (self, request : MyConfig, context:grpc.ServicerContext) -> Empty:
LOGGER.info('Setting from ConfigureOpticalDevice with Flows %s',request)
#device_id = request.myconfig_id
#device_uuid = device_id.myconfig_uuid
device_uuid = request.myconfig_id.myconfig_uuid
LOGGER.info("AAAAAAAAAAAAAAAAAAAAAAA device uuid {}".format(device_uuid))
resources=[]
result=None
config =eval(request.config)
filter_fields= ["frequency","target-output-power","interface","operational-mode"]
try:
context_client = ContextClient()
device = get_device(
context_client, device_uuid, rw_copy=True, include_endpoints=True, include_components=False,
include_config_rules=False)
if device is None:
LOGGER.info("AAAAAAAAAAAAAAAAAAAAAAAdevice is none")
raise NotFoundException('Device', device_uuid, extra_details='loading in ConfigureDevice')
resources,conditions=extract_resources(config=config,device=device)
driver : _Driver = get_driver(self.driver_instance_cache, device)
LOGGER.info("resource %s conditions %s",resources,conditions)
result = driver.SetConfig(resources=resources,conditions=conditions)
#todo
#add a control with the NETCONF get
#driver.GetConfig(resource_keys=filter_fields)
except Exception as e:
LOGGER.info("error in configuring %s",e)
LOGGER.info("result %s",result)
return Empty()
\ No newline at end of file
src/device/service/Tools.py
View file @ 2673daca
......@@ -11,7 +11,7 @@
# 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.
from uuid import UUID, uuid4, uuid5
import json, logging
from typing import Any, Dict, List, Optional, Tuple, Union
from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME
......@@ -27,9 +27,29 @@ from .ErrorMessages import (
ERROR_BAD_RESOURCE, ERROR_DELETE, ERROR_GET, ERROR_GET_INIT, ERROR_MISSING_KPI, ERROR_SAMPLETYPE, ERROR_SET,
ERROR_SUBSCRIBE, ERROR_UNSUBSCRIBE, ERROR_UNSUP_RESOURCE
)
from .drivers.oc_driver.OCDriver import OCDriver
from common.method_wrappers.ServiceExceptions import NotFoundException
from common.type_checkers.Checkers import chk_length, chk_type
from common.proto.context_pb2 import EndPoint
LOGGER = logging.getLogger(__name__)
def get_endpoint_matching(device : Device, endpoint_uuid_or_name : str) -> EndPoint:
for endpoint in device.device_endpoints:
choices = {endpoint.endpoint_id.endpoint_uuid.uuid, endpoint.name}
if endpoint_uuid_or_name in choices: return endpoint
device_uuid = device.device_id.device_uuid.uuid
extra_details = 'Device({:s})'.format(str(device_uuid))
raise NotFoundException('Endpoint', endpoint_uuid_or_name, extra_details=extra_details)
def get_device_endpoint_uuids(endpoint : Tuple[str, str, Optional[str]]) -> Tuple[str, str]:
chk_type('endpoint', endpoint, (tuple, list))
chk_length('endpoint', endpoint, min_length=2, max_length=3)
device_uuid, endpoint_uuid = endpoint[0:2] # ignore topology_uuid by now
return device_uuid, endpoint_uuid
def check_connect_rules(device_config : DeviceConfig) -> Dict[str, Any]:
connection_config_rules = dict()
unexpected_config_rules = list()
......@@ -434,3 +454,77 @@ def update_endpoints(src_device : Device, dst_device : Device) -> None:
dst_topology_id = dst_endpoint_id.topology_id
if len(src_topology_uuid) > 0: dst_topology_id.topology_uuid.uuid = src_topology_uuid
if len(src_context_uuid) > 0: dst_topology_id.context_id.context_uuid.uuid = src_context_uuid
def oc_default_endpoints(
device : Device, driver : _Driver, monitoring_loops : MonitoringLoops,
new_sub_devices : Dict[str, Device], new_sub_links : Dict[str, Link]
) -> List[str]:
pass
#def get_enpoint_name (device:Device,endpoint_id:str):
# str(UUID(str_uuid_or_name))
def get_edit_target (device:Device,is_opticalband:bool)-> str:
if (is_opticalband): return "optical-band"
else :
if device.device_type =='optical-roadm': return 'media-channel'
else : return 'optical-channel'
def extract_resources (config:dict,device:Device)-> list :
conditions={}
resources=[]
resources.append({"resource_key":"channel_namespace","value":config["channel_namespace"] if "channel_namespace" in config else None})
resources.append({"resource_key":'add_transceiver',"value":config['add_transceiver'] if 'add_transceiver' in config else None})
resources.append({"resource_key":"interface","value":config["update_interface"] if 'update_interface' in config else None})
is_opticalband=config['is_opticalband'] if 'is_opticalband' in config else False
conditions["is_opticalband"]=is_opticalband
conditions["edit_type"]=get_edit_target(device=device,is_opticalband=is_opticalband)
if ('flow' in config):
#for tuple_value in config['flow'][device.name]:
source_vals = []
dest_vals = []
for tuple_value in config['flow']:
source_port=None
destination_port=None
#resources.append({"resource_key":"source_port","value":source_port})
#resources.append({"resource_key":"destination_port","value":destination_port})
source_port_uuid,destination_port_uuid=tuple_value
if (source_port_uuid !='0'):
src_endpoint_obj = get_endpoint_matching(device, source_port_uuid)
source_port = src_endpoint_obj.name
source_vals.append(source_port)
if (destination_port_uuid !='0'):
dst_endpoint_obj = get_endpoint_matching(device, destination_port_uuid)
destination_port = dst_endpoint_obj.name
dest_vals.append(destination_port)
resources.append({"resource_key":"source_port","value":source_vals})
resources.append({"resource_key":"destination_port","value":dest_vals})
if ('new_config' in config):
lower_frequency=None
upper_frequency=None
resources.append({"resource_key":"target-output-power","value":config["new_config"]["target-output-power"] if "target-output-power" in config["new_config"] else None })
#resources.append({"resource_key":"frequency","value":config["new_config"]["frequency"] if "frequency" in config else config["new_config"]["freqency"]})
resources.append({"resource_key":"frequency","value":config["new_config"]["frequency"] if "frequency" in config["new_config"] else None})
resources.append({"resource_key":"operational-mode","value":config["new_config"]["operational-mode"] if "operational-mode" in config["new_config"] else None})
resources.append({"resource_key":"line-port","value":config["new_config"]["line-port"] if "line-port" in config["new_config"] else None})
resources.append({"resource_key":"name","value":config['new_config']['band_type'] if 'band_type' in config['new_config'] else None})
resources.append({"resource_key":"optical-band-parent","value":config["new_config"]["ob_id"] if "ob_id" in config["new_config"] else None })
resources.append({"resource_key":"channel_name","value":config["new_config"]["name"] if "name" in config["new_config"] else None})
if not is_opticalband :
if 'frequency' in config['new_config'] and 'band' in config['new_config'] and conditions["edit_type"] == 'media-channel':
lower_frequency= int(int(config['new_config']['frequency']) - (int(config['new_config']['band'])/2))
upper_frequency= int(int(config['new_config']['frequency']) + (int(config['new_config']['band'])/2))
#lower_frequency= (config['new_config']['frequency']- config['new_config']['band'])/2
#upper_frequency=(config['new_config']['frequency']+ config['new_config']['band'])/2
resources.append({"resource_key":"index","value":config["new_config"]["flow_id"] if "flow_id" in config["new_config"] else None})
else :
lower_frequency=config['new_config']['low-freq'] if "low-freq" in config['new_config'] else None
upper_frequency=config['new_config']['up-freq'] if 'up-freq' in config['new_config'] else None
resources.append({"resource_key":"index","value":config["new_config"]["ob_id"] if "ob_id" in config["new_config"] else None})
resources.append({"resource_key":"lower-frequency","value":lower_frequency})
resources.append({"resource_key":"upper-frequency","value":upper_frequency})
return [resources,conditions]
src/device/service/driver_api/DriverInstanceCache.py
View file @ 2673daca
......@@ -52,7 +52,7 @@ class DriverInstanceCache:
driver_class = self._driver_factory.get_driver_class(**filter_fields)
MSG = 'Driver({:s}) selected for device({:s}) with filter_fields({:s})...'
LOGGER.info(MSG.format(str(driver_class.__name__), str(device_uuid), str(filter_fields)))
driver_instance : _Driver = driver_class(address, port, **settings)
driver_instance : _Driver =driver_class(address, port,device_uuid=device_uuid, **settings) if (driver_class.__name__ == "OCDriver") else driver_class(address, port, **settings)
self._device_uuid__to__driver_instance[device_uuid] = driver_instance
return driver_instance
......
src/device/service/drivers/__init__.py
View file @ 2673daca
......@@ -167,3 +167,19 @@ if LOAD_ALL_DEVICE_DRIVERS:
FilterFieldEnum.DRIVER: DeviceDriverEnum.DEVICEDRIVER_FLEXSCALE,
}
]))
if LOAD_ALL_DEVICE_DRIVERS:
from .oc_driver.OCDriver import OCDriver # pylint: disable=wrong-import-position
DRIVERS.append(
(OCDriver, [
{
# Real Packet Router, specifying OpenConfig Driver => use OpenConfigDriver
FilterFieldEnum.DEVICE_TYPE:[
DeviceTypeEnum.NETCONFIG_AGENT,
DeviceTypeEnum.OPTICAL_ROADM,
DeviceTypeEnum.OPTICAL_TRANSPONDER
],
FilterFieldEnum.DRIVER : DeviceDriverEnum.DEVICEDRIVER_OC,
}
]))
src/device/service/drivers/oc_driver/OCDriver.py 0 → 100644
View file @ 2673daca
# 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 json
import anytree, copy, logging, pytz, queue, re, threading
#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
import xml.etree.ElementTree as ET
from apscheduler.job import Job
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, 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, get_subnode, set_subnode_value #dump_subtree
#from .Tools import xml_pretty_print, xml_to_dict, xml_to_file
from .templates.Interfaces.interfaces import interface_template
from .templates.VPN.physical import create_optical_channel,add_transceiver,create_media_channel,create_optical_band
from .RetryDecorator import retry
from context.client.ContextClient import ContextClient
from common.proto.context_pb2 import (
MyConfig,
ConfigActionEnum, Device, DeviceDriverEnum, DeviceId, DeviceList, DeviceOperationalStatusEnum, Empty
,MyConfigId,Uuid)
from .templates.Tools import extractor
from .Tools import generate_uuid_from_numbers
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\='([^\]]+)'\].*")
# Collection of samples through NetConf is very slow and each request collects all the data.
# Populate a cache periodically (when first interface is interrogated).
# Evict data after some seconds, when data is considered as outdated
SAMPLE_EVICTION_SECONDS = 30.0 # seconds
SAMPLE_RESOURCE_KEY = 'interfaces/interface/state/counters'
filter_fields= ["frequency","target-output-power","interface","operational-mode","line-port"]
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
):
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)
logging.info("response message %s",response)
@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 = []
str_config_messages=[]
#try:
# if add_proccess:
# str_config_messages = add_transceiver(resources[0]['value'])
# elif update_interface:
# str_config_messages = interface_template(resources[0]["value"])
# else:
if (conditions['edit_type']=='optical-channel'):
#transponder
str_config_messages = create_optical_channel(resources)
elif (conditions['edit_type']=='optical-band'):
#roadm optical-band
str_config_messages = create_optical_band(resources)
else :
#roadm media-channel
str_config_messages=create_media_channel(resources)
logging.info("config message %s",str_config_messages)
# if (add_proccess or update_interface):
# netconf_handler.edit_config( # configure the device
# config=str_config_messages, target=target, default_operation=default_operation,
# test_option=test_option, error_option=error_option, format=format)
# if commit_per_rule:
# netconf_handler.commit()
for str_config_message in str_config_messages:
# configuration of the received templates
if str_config_message is None: raise UnsupportedResourceKeyException("CONFIG")
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(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.info("error %s",e)
# logger.exception(msg.format(e))
# #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.__initial = TreeNode('.')
#self.__running = TreeNode('.')
self.__subscriptions = TreeNode('.')
self.__started = threading.Event()
self.__terminate = threading.Event()
self.__scheduler = BackgroundScheduler(daemon=True) # scheduler used to emulate sampling events
self.__scheduler.configure(
jobstores = {'default': MemoryJobStore()},
executors = {'default': ThreadPoolExecutor(max_workers=1)}, # important! 1 = avoid concurrent requests
job_defaults = {'coalesce': False, 'max_instances': 3},
timezone=pytz.utc)
self._temp_address=f"{address}{port}"
self.__out_samples = queue.Queue()
self.__netconf_handler = NetconfSessionHandler(self.address, self.port, **(self.settings))
self.__logger.info("settings are %s",settings)
self.__device_uuid=device_uuid
self.Connect()
#self.GetConfig()
#self.__samples_cache = SamplesCache(self.__netconf_handler, self.__logger)
def Connect(self) -> bool:
with self.__lock:
if self.__started.is_set(): return True
self.__netconf_handler.connect()
# Connect triggers activation of sampling events that will be scheduled based on subscriptions
self.__scheduler.start()
self.__started.set()
return True
def Disconnect(self) -> bool:
with self.__lock:
# Trigger termination of loops and processes
self.__terminate.set()
# If not started, assume it is already disconnected
if not self.__started.is_set(): return True
# Disconnect triggers deactivation of sampling events
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]]]:
self.__logger.info("device_uuid %s",self.__device_uuid)
chk_type('resources', resource_keys, list)
results = []
myConfig= MyConfig()
j=0
with self.__lock:
self.__logger.info(" resources_key %s",resource_keys)
context_client.connect()
config={}
channels_lst=[]
transceivers={}
try:
xml_data = self.__netconf_handler.get().data_xml
transceivers,interfaces,channels_lst,channel_namespace,endpoints=extractor(data_xml=xml_data,resource_keys=filter_fields,dic=config)
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))
#results.append((resource_key, e)) # if validation fails, store the exception
#///////////////////////// divider ////////////////////////////////////////////////////////
value_dic={}
value_dic["channels"]=channels_lst
value_dic["transceivers"]=transceivers
value_dic["interfaces"]=interfaces
value_dic["channel_namespace"]=channel_namespace
value_dic["endpoints"]=endpoints
self.__logger.info("config from get config %s",str(value_dic))
myConfig.config=str(value_dic)
myconfig_id=MyConfigId()
myConfig.myconfig_id.myconfig_uuid=self.__device_uuid if self.__device_uuid is not None else ""
config_id=context_client.SetMyConfig(myConfig)
context_client.close()
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]]) -> 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)
else:
results = edit_config(self.__netconf_handler, self.__logger, resources, delete=True)
return results
src/device/service/drivers/oc_driver/RetryDecorator.py 0 → 100644
View file @ 2673daca
# 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 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/oc_driver/Tools.py 0 → 100644
View file @ 2673daca
# 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 hashlib
import uuid
import xml.dom.minidom, xmltodict
def xml_pretty_print(data : str):
return xml.dom.minidom.parseString(data).toprettyxml()
def xml_to_file(data : str, file_path : str) -> None:
with open(file_path, mode='w', encoding='UTF-8') as f:
f.write(xml_pretty_print(data))
def xml_to_dict(data : str):
return xmltodict.parse(data)
def generate_uuid_from_numbers(code:str) ->str:
# Concatenate the numbers into a single string
# Generate a hash value using MD5 algorithm
hash_value = hashlib.md5(code.encode()).hexdigest()
# Convert the hash value into a UUID
generated_uuid = uuid.UUID(hash_value)
return str(generated_uuid)
src/device/service/drivers/oc_driver/__init__.py 0 → 100644
View file @ 2673daca
# 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
src/device/service/drivers/oc_driver/templates/Interfaces/__init__.py 0 → 100644
View file @ 2673daca
# 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.
src/device/service/drivers/oc_driver/templates/Interfaces/interfaces.py 0 → 100644
View file @ 2673daca
# 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.
from yattag import Doc, indent
import logging
def interface_template (interface_data:dict) :
data={"name":"eth0","ip":"192.168.1.1","prefix-length":'24'}
doc, tag, text = Doc().tagtext()
with tag('config',xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"):
with tag('interfaces', xmlns="http://openconfig.net/yang/interfaces"):
with tag('interface'):
with tag('name'):text(interface_data['name'])
with tag('config'):
with tag('name'):text(interface_data['name'])
with tag("enabled"):text(interface_data["enabled"])
with tag('ipv4',xmlns="http://openconfig.net/yang/interfaces/ip") :
with tag("addresses"):
with tag("address"):
with tag('ip'):text(interface_data["ip"])
with tag('config'):
with tag('ip'):text(interface_data["ip"])
with tag('prefix-length'):text(interface_data["prefix-length"])
result = indent(
doc.getvalue(),
indentation = ' '*2,
newline = '\r\n'
)
logging.info("interfaces %s",result)
return result
\ No newline at end of file
src/device/service/drivers/oc_driver/templates/Tools.py 0 → 100644
View file @ 2673daca
# 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 re,logging
import json
import lxml.etree as ET
from typing import Collection, Dict, Any
from yattag import Doc, indent
from .VPN.physical import create_optical_channel
def add_value_from_tag(target : Dict, field_name: str, field_value : ET.Element, cast=None) -> None:
if isinstance(field_value,str) or field_value is None or field_value.text is None: return
field_value = field_value.text
if cast is not None: field_value = cast(field_value)
target[field_name] = field_value
def add_value_from_collection(target : Dict, field_name: str, field_value : Collection) -> None:
if field_value is None or len(field_value) == 0: return
target[field_name] = field_value
"""
# Method Name: generate_templates
# Parameters:
- resource_key: [str] Variable to identify the rule to be executed.
- resource_value: [str] Variable with the configuration parameters of the rule to be executed.
- delete: [bool] Variable to identify whether to create or delete the rule.
- vendor: [str] Variable to identify the vendor of the equipment to be configured.
# Functionality:
This method generates the template to configure the equipment using pyangbind.
To generate the template the following steps are performed:
1) Get the first parameter of the variable "resource_key" to identify the main path of the rule.
2) Search for the specific configuration path
3) Call the method with the configuration parameters (resource_data variable).
# Return:
[dict] Set of templates generated according to the configuration rule
"""
def generate_templates(resource_key: str, resource_value: str, channel:str) -> str: # template management to be configured
result_templates = []
data={}
data['name']=channel
data['resource_key']=resource_key
data['value']=resource_value
result_templates.append(create_physical_config(data))
return result_templates
def extract_channel_xmlns (data_xml:str,is_opticalband:bool):
xml_bytes = data_xml.encode("utf-8")
root = ET.fromstring(xml_bytes)
namespace=None
channels=None
if (not is_opticalband) :
optical_channel_namespaces = {
'ns': 'urn:ietf:params:xml:ns:netconf:base:1.0',
'oc': 'http://openconfig.net/yang/platform',
}
channels= root.find('.//{*}optical-channel',optical_channel_namespaces)
if channels is not None :
optical_channel_namespace = channels.tag.replace("optical-channel", "")
namespace=optical_channel_namespace.replace("{", "").replace("}", "")
else :
optical_band_namespaces= {
'oc':'http://openconfig.net/yang/wavelength-router'
}
channels= root.find('.//{*}optical-bands',optical_band_namespaces)
if channels is not None:
optical_channel_namespace = channels.tag.replace("optical-bands", "")
namespace=optical_channel_namespace.replace("{", "").replace("}", "")
return namespace
def extract_channels_based_on_channelnamespace (xml_data:str,channel_namespace:str,is_opticalband:bool):
xml_bytes = xml_data.encode("utf-8")
root = ET.fromstring(xml_bytes)
channels=[]
logging.info("channel namespace %s opticalband %s",channel_namespace ,is_opticalband)
# Find the component names whose children include the "optical-channel" element
if (not is_opticalband):
namespace = {'namespace': 'http://openconfig.net/yang/platform','cn':channel_namespace}
component_names = root.findall('.//namespace:component[cn:optical-channel]',namespace)
# Extract and print the component names
for component in component_names:
component_name = component.find('namespace:name', namespace).text
channels.append({"index":component_name})
else :
namespaces = {
'wr': 'http://openconfig.net/yang/wavelength-router',
'fs': channel_namespace
}
wl = root.findall('.//fs:optical-band',namespaces=namespaces)
for component in wl :
index=component.find('.//fs:index',namespaces).text
dest_port_name = component.find('.//fs:dest/fs:config/fs:port-name', namespaces).text
# Retrieve port-name for source (assuming it exists in the XML structure)
source_port_name = component.find('.//fs:source/fs:config/fs:port-name', namespaces).text
channels.append({"index":index,"endpoints":(source_port_name,dest_port_name)})
# Retrieve port-name for dest
logging.info("extract channels %s",channels)
return channels
def extract_channels_based_on_type (xml_data:str):
xml_bytes = xml_data.encode("utf-8")
root = ET.fromstring(xml_bytes)
namespace = {'oc': 'http://openconfig.net/yang/platform', 'typex': 'http://openconfig.net/yang/platform-types'}
channel_names = []
components = root.findall('.//oc:component', namespace)
for component in components:
type_element = component.find('.//oc:state/oc:type[.="oc-opt-types:OPTICAL_CHANNEL"]',namespaces=namespace)
if type_element is not None and type_element.text == 'oc-opt-types:OPTICAL_CHANNEL':
name_element = component.find('oc:name', namespace)
if name_element is not None:
channel_names.append(name_element.text)
return channel_names
def extract_value(resource_key:str,xml_data:str,dic:dict,channel_name:str,channel_namespace:str):
xml_bytes = xml_data.encode("utf-8")
root = ET.fromstring(xml_bytes)
namespace = {'oc': 'http://openconfig.net/yang/platform',
'td': channel_namespace}
element = root.find(f'.//oc:component[oc:name="{channel_name}"]', namespace)
if element is not None:
parameter= element.find(f'.//td:{resource_key}',namespace)
if (parameter is not None):
value = parameter.text
dic[resource_key]=value
else:
print(" element not found.")
return dic
def extract_port_value (xml_string:list,port_name:str):
xml_bytes = xml_string.encode("utf-8")
root = ET.fromstring(xml_bytes)
namespace = {"oc": "http://openconfig.net/yang/platform"}
component=root.find(f".//oc:component[oc:name='{port_name}']", namespace)
onos_index = component.find(
f".//oc:property//oc:state/oc:name[.='onos-index']/../oc:value", namespace
).text
return (port_name,onos_index)
def extract_tranceiver (data_xml:str,dic:dict):
xml_bytes = data_xml.encode("utf-8")
root = ET.fromstring(xml_bytes)
namespaces = {
'ns': 'urn:ietf:params:xml:ns:netconf:base:1.0',
'oc': 'http://openconfig.net/yang/platform',
'oc-terminal': 'http://openconfig.net/yang/terminal-device',
'oc-platform-types': 'http://openconfig.net/yang/platform-types'
}
transceiver_components = root.findall('.//oc:component/oc:state/[oc:type="oc-platform-types:TRANSCEIVER"]../oc:state/oc:name', namespaces)
component_names = [component.text for component in transceiver_components]
dic['transceiver']=component_names
return dic
def extract_interface (xml_data:str,dic:dict):
xml_bytes = xml_data.encode("utf-8")
root = ET.fromstring(xml_bytes)
namespaces = {
'ns': 'urn:ietf:params:xml:ns:netconf:base:1.0',
'oc': 'http://openconfig.net/yang/interfaces',
}
ip_namespaces = {
'oc': 'http://openconfig.net/yang/interfaces',
'ip': 'http://openconfig.net/yang/interfaces/ip',
}
interfaces = root.findall('.//oc:interfaces/oc:interface', namespaces)
interface_names = [interface.find('oc:name', namespaces).text for interface in interfaces]
interface_enabled=[interface.find('oc:config/oc:enabled', namespaces).text for interface in interfaces]
ip_address_element = root.find('.//ip:ip', ip_namespaces)
interface_prefix_length=root.find('.//ip:prefix-length',ip_namespaces)
if (len(interface_names) > 0):
dic['interface']={"name":interface_names[0],'ip':ip_address_element.text,'enabled':interface_enabled[0],"prefix-length":interface_prefix_length.text}
else :
dic['interface']={}
return dic
def has_opticalbands(xml_data:str):
xml_bytes = xml_data.encode("utf-8")
root = ET.fromstring(xml_bytes)
has_opticalbands=False
elements= root.find('.//{*}optical-bands')
if (elements is not None and len(elements) >0):
has_opticalbands=True
else :
has_opticalbands=False
return has_opticalbands
def extractor(data_xml:str,resource_keys:list,dic:dict):
logging.info('data xml %s',data_xml)
endpoints=[]
is_opticalband=has_opticalbands(xml_data=data_xml)
logging.info("from extractor opticalband %s",is_opticalband)
channel_namespace=extract_channel_xmlns(data_xml=data_xml,is_opticalband=is_opticalband)
# channel_names=extract_channels_based_on_type(xml_data=data_xml)
# if len(channel_names)==0 :
channel_names= extract_channels_based_on_channelnamespace(xml_data=data_xml,channel_namespace=channel_namespace,is_opticalband=is_opticalband)
lst_dic=[]
if (is_opticalband):
endpoints=channel_names
else:
for channel_name in channel_names:
dic={}
for resource_key in resource_keys :
if (resource_key != 'interface'):
dic=extract_value(dic=dic,resource_key=resource_key,xml_data=data_xml,channel_name=channel_name,channel_namespace=channel_namespace)
dic["name"]=channel_name
endpoints.append({"endpoint_uuid":{"uuid":channel_name}})
lst_dic.append(dic)
transceivers_dic=extract_tranceiver(data_xml=data_xml,dic={})
interfaces_dic=extract_interface(xml_data=data_xml,dic={})
return [transceivers_dic,interfaces_dic,lst_dic,channel_namespace,endpoints]
\ No newline at end of file
src/device/service/drivers/oc_driver/templates/VPN/__init__.py 0 → 100644
View file @ 2673daca
# 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.
src/device/service/drivers/oc_driver/templates/VPN/physical.py 0 → 100644
View file @ 2673daca
# 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.
from yattag import Doc, indent
import logging
def seperate_port_config(resources:list,unwanted_keys:list[str])->list[list,dict,str]:
config=[]
ports={}
index=None
for item in resources :
logging.info("Andrea223344 item={}".format(item['resource_key']))
if (item['value'] is not None and (item['resource_key'] not in unwanted_keys)):
config.append({'resource_key':item['resource_key'], 'value':item['value']} )
#if (item['resource_key'] == 'destination_port' or item['resource_key'] == 'source_port') and item['value'] is not None:
# ports[item['resource_key']]=item['value']
if (item['resource_key'] == 'destination_port' or item['resource_key'] == 'source_port'):
ports[item['resource_key']]=item['value']
if (item['resource_key']=='index' and item['value'] is not None) :
index=item['value']
logging.info("from create_templates config %s ports %s index %s",config,ports,index)
return [config,ports,index]
def create_optical_channel(resources):
logging.debug("TRANSPONDERconfiguration, resources={}".format(resources))
unwanted_keys=['destination_port','source_port','channel_namespace','optical-band-parent','index', 'name']
results =[]
data={"name":i["value"] for i in resources if i["resource_key"]=="channel_name"}
data["channel_namespace"]=next((i["value"] for i in resources if i["resource_key"] == "channel_namespace"), None)
config,ports,index=seperate_port_config(resources,unwanted_keys=unwanted_keys)
logging.info("from physical %s",resources)
port_val = ""
if 'destination_port' in ports and ports['destination_port'][0] is not None:
port_val = ports['destination_port'][0]
else:
port_val = ports['source_port'][0]
logging.info("transponder port={}".format(port_val))
doc, tag, text = Doc().tagtext()
#with tag('config'):
with tag('config',xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"):
with tag('components', xmlns="http://openconfig.net/yang/platform"):
with tag('component'):
with tag('name'):text("channel-{}".format(port_val))
with tag('config'):
with tag('name'):text("channel-{}".format(port_val))
with tag('optical-channel',xmlns=data["channel_namespace"]):
with tag('config'):
for resource in config:
with tag(resource['resource_key']):text(resource['value'])
result = indent(
doc.getvalue(),
indentation = ' '*2,
newline = ''
)
results.append(result)
logging.info("xml %s",results)
return results
def add_transceiver (transceiver_name:str):
doc, tag, text = Doc().tagtext()
with tag('config',xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"):
with tag('components', xmlns="http://openconfig.net/yang/platform"):
with tag('component'):
with tag('name'):text(transceiver_name)
with tag("config"):
with tag('name'):text(transceiver_name)
with tag("state"):
with tag('name'):text(transceiver_name)
with tag("type",('xmlns:oc-platform-types',"http://openconfig.net/yang/platform-types")):text("oc-platform-types:TRANSCEIVER")
with tag("transceiver",xmlns="http://openconfig.net/yang/platform/transceiver"):
with tag("config"):
with tag("enabled"):text("true")
with tag("form-factor-preconf",("xmlns:oc-opt-types","http://openconfig.net/yang/transport-types")):text("oc-opt-types:QSFP56_DD_TYPE1")
with tag("ethernet-pmd-preconf",("xmlns:oc-opt-types","http://openconfig.net/yang/transport-types")):text("oc-opt-types:ETH_400GBASE_ZR")
with tag("fec-mode",("xmlns:oc-platform-types","http://openconfig.net/yang/platform-types")):text("oc-platform-types:FEC_AUTO")
with tag("module-functional-type",("xmlns:oc-opt-types","http://openconfig.net/yang/transport-types")):text("oc-opt-types:TYPE_DIGITAL_COHERENT_OPTIC")
with tag("state"):
with tag("enabled"):text("true")
with tag("form-factor-preconf",("xmlns:oc-opt-types","http://openconfig.net/yang/transport-types")):text("oc-opt-types:QSFP56_DD_TYPE1")
with tag("ethernet-pmd-preconf",("xmlns:oc-opt-types","http://openconfig.net/yang/transport-types")):text("oc-opt-types:ETH_400GBASE_ZR")
with tag("fec-mode",("xmlns:oc-platform-types","http://openconfig.net/yang/platform-types")):text("oc-platform-types:FEC_AUTO")
with tag("module-functional-type",("xmlns:oc-opt-types","http://openconfig.net/yang/transport-types")):text("oc-opt-types:TYPE_DIGITAL_COHERENT_OPTIC")
with tag("vendor"):text("Cisco")
with tag("vendor-part"):text("400zr-QSFP-DD")
with tag("vendor-rev"):text("01")
with tag("serial-no"):text("1567321")
with tag("physical-channels"):
with tag("channel"):
with tag("index"):text("1")
with tag("config"):
with tag("index"):text("1")
with tag("associated-optical-channel"):text("channel-4")
result = indent(
doc.getvalue(),
indentation = ' '*2,
newline = ''
)
return result
def create_optical_band (resources) :
results =[]
unwanted_keys=['destination_port','source_port','channel_namespace','frequency','optical-band-parent']
config,ports,index= seperate_port_config(resources,unwanted_keys=unwanted_keys)
doc, tag, text = Doc().tagtext()
#with tag('config'):
with tag('config',xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"):
with tag('wavelength-router', xmlns="http://openconfig.net/yang/wavelength-router"):
with tag('optical-bands',xmlns="http://flex-scale-project.eu/yang/flex-scale-mg-on"):
n = 0
if 'destination_port' in ports:
n = len(ports['destination_port'])
else:
n = len(ports['source_port'])
for i in range(0, n):
#with tag('optical-band', operation="create"):
with tag('optical-band'):
if index is not None:
with tag('index'):text(str(int(index)+i))
with tag('config'):
#if index is not None:
# with tag('index'):text(str(int(index)+i))
for resource in config:
if resource['resource_key'] == "index":
with tag('index'):text(str(int(index)+i))
else:
with tag(resource['resource_key']):text(resource['value'])
with tag('admin-status'):text('ENABLED')
#with tag('fiber-parent'):text(ports['destination_port'] if 'destination_port' in ports else ports['source_port'])
if ('destination_port' in ports) and (ports['destination_port'][i] is not None):
with tag('dest'):
with tag('config'):
with tag('port-name'):text(ports['destination_port'][i])
if ('source_port' in ports) and (ports['source_port'][i] is not None):
with tag('source'):
with tag('config'):
with tag('port-name'):text(ports['source_port'][i])
result = indent(
doc.getvalue(),
indentation = ' '*2,
newline = ''
)
results.append(result)
return results
def create_media_channel (resources):
results=[]
unwanted_keys=['destination_port','source_port','channel_namespace','frequency','operational-mode', 'optical-band-parent']
config,ports,index= seperate_port_config(resources,unwanted_keys=unwanted_keys)
doc, tag, text = Doc().tagtext()
#with tag('config'):
with tag('config',xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"):
with tag('wavelength-router', xmlns="http://openconfig.net/yang/wavelength-router"):
with tag('media-channels'):
n = 0
if 'destination_port' in ports:
n = len(ports['destination_port'])
else:
n = len(ports['source_port'])
for i in range(0, n):
#with tag('channel', operation="create"):
with tag('channel'):
with tag('index'):text(str(int(index)+i))
with tag('config'):
#with tag('index'):text(index)
for resource in config:
logging.info("Andrea223344 resources_key= {}".format(resource['resource_key']))
if resource['resource_key'] == "index":
with tag('index'):text(str(int(index)+i))
else:
with tag(resource['resource_key']):text(resource['value'])
if ('destination_port' in ports) and (ports['destination_port'][i] is not None):
with tag('dest'):
with tag('config'):
with tag('port-name'):text(ports['destination_port'][i])
if ('source_port' in ports) and (ports['source_port'][i] is not None):
with tag('source'):
with tag('config'):
with tag('port-name'):text(ports['source_port'][i])
result = indent(
doc.getvalue(),
indentation = ' '*2,
newline = ''
)
results.append(result)
return results
\ No newline at end of file
src/device/service/drivers/oc_driver/templates/__init__.py 0 → 100644
View file @ 2673daca
# 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.
src/device/service/drivers/openconfig/templates/Interfaces.py
View file @ 2673daca
......@@ -31,10 +31,6 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
interface = {}
interface_name = xml_interface.find('oci:name', namespaces=NAMESPACES)
if interface_name is None or interface_name.text is None: continue
add_value_from_tag(interface, 'name', interface_name)
#interface_type = xml_interface.find('oci:config/oci:type', namespaces=NAMESPACES)
#add_value_from_tag(interface, 'type', interface_type)
......@@ -42,8 +38,11 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
interface_type = xml_interface.find('oci:config/oci:type', namespaces=NAMESPACES)
elif xml_interface.find('oci:state/oci:type', namespaces=NAMESPACES) is not None:
interface_type = xml_interface.find('oci:state/oci:type', namespaces=NAMESPACES)
else:
interface_type = ''
else: continue
interface_name = xml_interface.find('oci:name', namespaces=NAMESPACES)
if interface_name is None or interface_name.text is None: continue
add_value_from_tag(interface, 'name', interface_name)
# Get the type of interface according to the vendor's type
if 'ianaift:' in interface_type.text:
......
src/device/service/drivers/openconfig/templates/Inventory.py
View file @ 2673daca
......@@ -54,7 +54,6 @@ XPATH_PORTS = "//ocp:components/ocp:component"
def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
response = []
LOGGER.debug("InventoryPrueba")
parent_types = {}
for xml_component in xml_data.xpath(XPATH_PORTS, namespaces=NAMESPACES):
LOGGER.info('xml_component inventario = {:s}'.format(str(ET.tostring(xml_component))))
......@@ -78,9 +77,9 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
add_value_from_tag(inventory['attributes'], 'location', component_location)
component_type = xml_component.find('ocp:state/ocp:type', namespaces=NAMESPACES)
component_type.text = component_type.text.replace('oc-platform-types:','')
if component_type is None: continue
add_value_from_tag(inventory, 'class', component_type)
if component_type is not None:
component_type.text = component_type.text.replace('oc-platform-types:','')
add_value_from_tag(inventory, 'class', component_type)
if inventory['class'] == 'CPU' or inventory['class'] == 'STORAGE': continue
......
src/device/service/drivers/openconfig/templates/RoutingPolicy.py
View file @ 2673daca
......@@ -35,7 +35,7 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
#LOGGER.info('xml_policy_definition = {:s}'.format(str(ET.tostring(xml_policy_definition))))
policy_definition = {}
statement_name = ''
policy_name = xml_policy_definition.find('ocrp:name', namespaces=NAMESPACES)
if policy_name is None or policy_name.text is None: continue
add_value_from_tag(policy_definition, 'policy_name', policy_name)
......