Pre-merge code cleanup (eeb87ff2) · Commits · TFS / controller

src/service/service/ServiceServiceServicerImpl.py

+26 −25

Original line number	Diff line number	Diff line
		@@ -17,7 +17,8 @@ from typing import Optional
		from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
		from common.method_wrappers.ServiceExceptions import (
		AlreadyExistsException, InvalidArgumentException, NotFoundException, NotImplementedException,
		OperationFailedException)
		OperationFailedException
		)
		from common.proto.context_pb2 import (
		Connection, ConstraintActionEnum, Empty, Service, ServiceId, ServiceStatusEnum,
		ServiceTypeEnum, TopologyId
		@@ -42,8 +43,8 @@ from .service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory
		from .task_scheduler.TaskScheduler import TasksScheduler
		from .tools.GeodesicDistance import gps_distance
		from .tools.OpticalTools import (
		add_lightpath, delete_lightpath, adapt_reply, get_device_name_from_uuid, get_optical_band,refresh_opticalcontroller,
		DelFlexLightpath
		add_lightpath, delete_lightpath, adapt_reply, get_device_name_from_uuid,
		get_optical_band, refresh_opticalcontroller, DelFlexLightpath
		)


		@@ -383,8 +384,9 @@ class ServiceServiceServicerImpl(ServiceServiceServicer):
		bitrate = int(float(constraint.custom.constraint_value))
		break

		bitrate = int(
		float(service.service_constraints[0].custom.constraint_value))
		bitrate = int(float(
		service.service_constraints[0].custom.constraint_value
		))
		if len(service.service_config.config_rules) > 0:
		c_rules_dict = json.loads(
		service.service_config.config_rules[0].custom.resource_value)
		@@ -404,7 +406,6 @@ class ServiceServiceServicerImpl(ServiceServiceServicer):
		params['ob_id' ] = ob_id
		params['flow_id'] = flow_id


		tasks_scheduler = TasksScheduler(self.service_handler_factory)
		tasks_scheduler.compose_from_optical_service(service, params=params, is_delete=True)
		tasks_scheduler.execute_all()

src/service/service/task_scheduler/TaskExecutor.py

+44 −62

Original line number	Diff line number	Diff line
		@@ -39,7 +39,6 @@ from service.service.service_handler_api.Exceptions import (
		from service.service.service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory, get_service_handler_class
		from service.service.tools.ObjectKeys import get_connection_key, get_device_key, get_service_key, get_qkd_app_key
		from service.service.tools.object_uuid import opticalconfig_get_uuid
		from common.DeviceTypes import DeviceTypeEnum

		if TYPE_CHECKING:
		from service.service.service_handler_api._ServiceHandler import _ServiceHandler
		@@ -128,7 +127,9 @@ class TaskExecutor:
		self._store_grpc_object(CacheableObjectType.DEVICE, device_key, device)

		# New function Andrea for Optical Devices
		def configure_optical_device(self, device : Device, settings : str, flows : list, is_opticalband : bool):
		def configure_optical_device(
		self, device : Device, settings : str, flows : list, is_opticalband : bool
		):
		device_key = get_device_key(device.device_id)
		optical_config_id = OpticalConfigId()
		optical_config_id.opticalconfig_uuid = opticalconfig_get_uuid(device.device_id)
		@@ -139,7 +140,6 @@ class TaskExecutor:
		config_type = None

		try:

		result = self._context_client.SelectOpticalConfig(optical_config_id)

		new_config = json.loads(result.config)
		@@ -148,25 +148,21 @@ class TaskExecutor:
		if config_type == 'optical-transponder':
		setting['status']='ENABLED'
		if result is not None :

		new_config["new_config"] = setting

		new_config["is_opticalband"] = is_opticalband
		new_config["flow"] = flows
		result.config = json.dumps(new_config)
		optical_config.CopyFrom(result)

		self._device_client.ConfigureOpticalDevice(optical_config)
		self._store_grpc_object(CacheableObjectType.DEVICE, device_key, device)

		except Exception as e:
		LOGGER.info("error in configure_optical_device %s",e)

		# Deconfiguring Optical Devices ( CNIT )
		def deconfigure_optical_device(self, device : Device, channel_indexes :list,is_opticalband:bool,dev_flow:list):

		def deconfigure_optical_device(
		self, device : Device, channel_indexes : list, is_opticalband : bool, dev_flow : list
		):
		errors = []
		dev_flow=dev_flow
		flows = []
		indexes = {}
		new_config = {}
		@@ -179,22 +175,18 @@ class TaskExecutor:
		# if Roadm the channel index is the flow_id ,or ob_id
		else:
		for index in channel_indexes:
		if ( not is_opticalband):
		if not is_opticalband:
		indexes["flow_id"] = index
		else:
		indexes["ob_id"] = index


		try:

		result = self._context_client.SelectOpticalConfig(optical_config_id)
		# for extractor in device service to extract the index , dummy data for freq and band required
		indexes["frequency"] = None
		indexes["band"] = None
		if result is not None:
		new_config = json.loads(result.config)


		new_config["new_config"]=indexes
		new_config["flow"] = flows if len(flows)>0 else dev_flow
		new_config["is_opticalband"] = is_opticalband
		@@ -203,15 +195,14 @@ class TaskExecutor:
		# new_optical_config.opticalconfig_id.CopyFrom (optical_config_id)
		# new_optical_config.device_id.CopyFrom(device.device_id)
		self._device_client.DisableOpticalDevice(result)


		except Exception as e:
		errors.append(e)
		LOGGER.info("error in deconfigure_optical_device %s",e)
		return errors


		def delete_setting (self,service_id:ServiceId, config_key:str,config_value:str) :
		def delete_setting(
		self, service_id : ServiceId, config_key : str, config_value : str
		):
		service_configRule = ServiceConfigRule()
		service_configRule.service_id.CopyFrom( service_id)
		service_configRule.configrule_custom.resource_key = config_key
		@@ -224,9 +215,6 @@ class TaskExecutor:
		except Exception as e :
		LOGGER.info("error in delete service config rule %s",e)




		def check_service_for_media_channel(self, connections : ConnectionList, item) -> bool:
		service = item
		if (isinstance(item, ServiceId)):
		@@ -237,23 +225,18 @@ class TaskExecutor:
		connection_uuid=connection.connection_id.connection_uuid
		if class_service_handler is None:
		class_service_handler = self.get_service_handler(connection, service,**service_handler_settings)

		if class_service_handler.check_media_channel(connection_uuid):
		return True
		return False

		def check_connection_for_media_channel (self,connection:Connection,service:Service)->bool:

		def check_connection_for_media_channel(
		self, connection : Connection, service : Service
		) -> bool:
		service_handler_settings = {}
		connection_uuid = connection.connection_id.connection_uuid
		service_handler_class = self.get_service_handler(connection, service, **service_handler_settings)


		return service_handler_class.check_media_channel(connection_uuid)




		def get_device_controller(self, device : Device) -> Optional[Device]:
		#json_controller = None
		#for config_rule in device.device_config.config_rules:
		@@ -403,4 +386,3 @@ class TaskExecutor:
		except Exception as e:
		LOGGER.error(f"Failed to delete QKD app with AppId {app_id}: {str(e)}")
		raise e

src/service/service/task_scheduler/TaskScheduler.py

+114 −130

Original line number	Diff line number	Diff line
		@@ -12,11 +12,11 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import graphlib, logging, queue, time ,json
		import graphlib, logging, queue, time
		from typing import TYPE_CHECKING, Dict, Tuple
		from common.proto.context_pb2 import (Connection, ConnectionId, Service
		, ServiceId, ServiceStatusEnum,ConnectionList
		,DeviceList)
		from common.proto.context_pb2 import (
		Connection, ConnectionId, Service, ServiceId, ServiceStatusEnum, ConnectionList
		)
		from common.proto.pathcomp_pb2 import PathCompReply
		from common.tools.grpc.Tools import grpc_message_to_json_string
		from context.client.ContextClient import ContextClient
		@@ -86,12 +86,16 @@ class TasksScheduler:
		self._dag.add(service_delete_key, service_removing_key)
		return service_removing_key, service_delete_key

		def _optical_service_remove(self, service_id : ServiceId,has_media_channel:bool,has_optical_band=True) -> Tuple[str, str]:

		def _optical_service_remove(
		self, service_id : ServiceId, has_media_channel : bool, has_optical_band = True
		) -> Tuple[str, str]:
		service_removing_key = self._add_task_if_not_exists(Task_ServiceSetStatus(
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE))
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE
		))

		service_delete_key = self._add_task_if_not_exists(Task_OpticalServiceDelete(self._executor, service_id,has_media_channel,has_optical_band))
		service_delete_key = self._add_task_if_not_exists(Task_OpticalServiceDelete(
		self._executor, service_id, has_media_channel, has_optical_band
		))

		# deleting a service requires the service is in removing state
		self._dag.add(service_delete_key, service_removing_key)
		@@ -114,57 +118,57 @@ class TasksScheduler:
		return connection_configure_key

		def _connection_deconfigure(self, connection_id : ConnectionId, service_id : ServiceId) -> str:

		connection_deconfigure_key = self._add_task_if_not_exists(Task_ConnectionDeconfigure(
		self._executor, connection_id))
		self._executor, connection_id
		))

		# the connection deconfiguration depends on its connection's service being in removing state
		service_pending_removal_key = self._add_task_if_not_exists(Task_ServiceSetStatus(
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_PENDING_REMOVAL))
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_PENDING_REMOVAL
		))
		self._dag.add(connection_deconfigure_key, service_pending_removal_key)

		# the connection's service depends on the connection deconfiguration to transition to delete
		service_delete_key = self._add_task_if_not_exists(Task_ServiceDelete(
		self._executor, service_id))
		self._executor, service_id
		))
		self._dag.add(service_delete_key, connection_deconfigure_key)

		return connection_deconfigure_key



		def _optical_connection_deconfigure(self, connection_id : ConnectionId, service_id : ServiceId,has_media_channel:bool,has_optical_band=True) -> str:

		def _optical_connection_deconfigure(
		self, connection_id : ConnectionId, service_id : ServiceId,
		has_media_channel : bool, has_optical_band = True
		) -> str:
		connection_deconfigure_key = self._add_task_if_not_exists(Task_OpticalConnectionDeconfigure(
		self._executor, connection_id,has_media_channel=has_media_channel))
		self._executor, connection_id, has_media_channel=has_media_channel
		))

		# the connection deconfiguration depends on its connection's service being in removing state
		service_pending_removal_key = self._add_task_if_not_exists(Task_ServiceSetStatus(
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE))
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE
		))
		self._dag.add(connection_deconfigure_key, service_pending_removal_key)


		service_delete_key = self._add_task_if_not_exists(Task_OpticalServiceDelete(
		self._executor, service_id,has_media_channel,has_optical_band))
		self._executor, service_id, has_media_channel, has_optical_band
		))
		self._dag.add(service_delete_key, connection_deconfigure_key)

		return connection_deconfigure_key

		def _optical_service_config_remove(self, connection_id : ConnectionId
		, service_id : ServiceId
		def _optical_service_config_remove(
		self, connection_id : ConnectionId, service_id : ServiceId
		) -> str:

		service_config_key = self._add_task_if_not_exists(Task_OpticalServiceConfigDelete(
		self._executor, connection_id, service_id

		))

		service_pending_removal_key = self._add_task_if_not_exists(Task_ServiceSetStatus(
		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE))

		self._executor, service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE
		))
		self._dag.add(service_config_key, service_pending_removal_key)
		return service_config_key


		def compose_from_pathcompreply(self, pathcomp_reply : PathCompReply, is_delete : bool = False) -> None:
		t0 = time.time()
		include_service = self._service_remove if is_delete else self._service_create
		@@ -186,28 +190,31 @@ class TasksScheduler:
		t1 = time.time()
		LOGGER.debug('[compose_from_pathcompreply] elapsed_time: {:f} sec'.format(t1-t0))


		def check_service_for_media_channel (self,connections:ConnectionList,item)->Tuple[bool,bool]:
		def check_service_for_media_channel(
		self, connections : ConnectionList, item
		)->Tuple[bool, bool]:
		service = item
		has_media_channel = False
		has_optical_band = False
		if (isinstance(item,ServiceId)):
		if isinstance(item, ServiceId):
		service = self._executor.get_service(item)
		class_service_handler = None
		service_handler_settings = {}
		for connection in connections.connections:
		connection_uuid = connection.connection_id.connection_uuid.uuid
		if class_service_handler is None:
		class_service_handler=self._executor.get_service_handler(connection, service,**service_handler_settings)

		class_service_handler = self._executor.get_service_handler(
		connection, service, **service_handler_settings
		)
		if class_service_handler.check_media_channel(connection_uuid):
		has_media_channel = True
		else :
		has_optical_band = True

		return (has_media_channel, has_optical_band)

		def compose_from_optical_service(self, service : Service,params:dict, is_delete : bool = False) -> None:
		def compose_from_optical_service(
		self, service : Service, params : dict, is_delete : bool = False
		) -> None:
		t0 = time.time()
		include_service = self._optical_service_remove if is_delete else self._service_create
		include_connection = self._optical_connection_deconfigure if is_delete else self._connection_configure
		@@ -224,109 +231,88 @@ class TasksScheduler:
		while not pending_items_to_explore.empty():
		try:
		item = pending_items_to_explore.get(block=False)

		except queue.Empty:
		break

		if isinstance(item, Service):

		str_item_key = grpc_message_to_json_string(item.service_id)
		if str_item_key in explored_items: continue
		connections = self._context_client.ListConnections(item.service_id)
		has_media_channel,has_optical_band=self.check_service_for_media_channel(connections=connections,item=item.service_id)
		has_media_channel, has_optical_band = self.check_service_for_media_channel(
		connections=connections, item=item.service_id
		)

		if len(service.service_config.config_rules) > 0:


		reply, code = delete_lightpath(
		params['src']
		,params ['dst']
		, params['bitrate']
		, params['ob_id']
		,delete_band=not has_media_channel
		, flow_id= params['flow_id']
		params['src'], params ['dst'], params['bitrate'], params['ob_id'],
		delete_band=not has_media_channel, flow_id= params['flow_id']
		)


		if code == 400 and reply_not_allowed in reply:
		MSG = 'Deleteion for the service is not Allowed , Served Lightpaths is not empty'
		raise Exception(MSG)

		include_service(item.service_id,has_media_channel=has_media_channel,has_optical_band=has_optical_band)
		include_service(
		item.service_id, has_media_channel=has_media_channel, has_optical_band=has_optical_band
		)
		self._add_service_to_executor_cache(item)


		for connection in connections.connections:
		self._add_connection_to_executor_cache(connection)
		pending_items_to_explore.put(connection)
		explored_items.add(str_item_key)


		elif isinstance(item, ServiceId):

		if code == 400 and reply_not_allowed in reply: break

		str_item_key = grpc_message_to_json_string(item)
		if str_item_key in explored_items: continue
		connections = self._context_client.ListConnections(item)
		has_media_channel,has_optical_band=self.check_service_for_media_channel(connections=connections,item=item)


		include_service(item,has_media_channel=has_media_channel,has_optical_band=has_optical_band)


		has_media_channel, has_optical_band = self.check_service_for_media_channel(
		connections=connections, item=item
		)
		include_service(
		item, has_media_channel=has_media_channel, has_optical_band=has_optical_band
		)
		self._executor.get_service(item)

		for connection in connections.connections:

		self._add_connection_to_executor_cache(connection)
		pending_items_to_explore.put(connection)




		explored_items.add(str_item_key)

		elif isinstance(item, Connection):


		if code == 400 and reply_not_allowed in reply:break

		str_item_key = grpc_message_to_json_string(item.connection_id)
		if str_item_key in explored_items: continue


		connection_key = include_connection(item.connection_id, item.service_id,has_media_channel=has_media_channel,has_optical_band=has_optical_band)
		connection_key = include_connection(
		item.connection_id, item.service_id, has_media_channel=has_media_channel,
		has_optical_band=has_optical_band
		)
		self._add_connection_to_executor_cache(connection)

		if include_service_config is not None :
		connections_list = ConnectionList()
		connections_list.connections.append(item)

		is_media_channel,_=self.check_service_for_media_channel(connections=connections_list,item=service)

		is_media_channel, _ = self.check_service_for_media_channel(
		connections=connections_list, item=service
		)
		if has_optical_band and is_media_channel:
		include_service_config(item.connection_id
		, item.service_id

		include_service_config(
		item.connection_id, item.service_id
		)


		self._executor.get_service(item.service_id)
		pending_items_to_explore.put(item.service_id)


		for sub_service_id in item.sub_service_ids:
		_,service_key_done = include_service(sub_service_id,has_media_channel=has_media_channel,has_optical_band=has_optical_band)
		_,service_key_done = include_service(
		sub_service_id, has_media_channel=has_media_channel,
		has_optical_band=has_optical_band
		)
		self._executor.get_service(sub_service_id)
		self._dag.add(service_key_done, connection_key)
		pending_items_to_explore.put(sub_service_id)



		explored_items.add(str_item_key)


		else:
		MSG = 'Unsupported item {:s}({:s})'
		raise Exception(MSG.format(type(item).__name__, grpc_message_to_json_string(item)))
		@@ -334,7 +320,6 @@ class TasksScheduler:
		t1 = time.time()
		LOGGER.debug('[compose_from_service] elapsed_time: {:f} sec'.format(t1-t0))


		def compose_from_service(self, service : Service, is_delete : bool = False) -> None:
		t0 = time.time()
		include_service = self._service_remove if is_delete else self._service_create
		@@ -357,12 +342,9 @@ class TasksScheduler:
		include_service(item.service_id)
		self._add_service_to_executor_cache(item)
		connections = self._context_client.ListConnections(item.service_id)

		for connection in connections.connections:

		self._add_connection_to_executor_cache(connection)
		pending_items_to_explore.put(connection)

		explored_items.add(str_item_key)

		elif isinstance(item, ServiceId):
		@@ -375,7 +357,6 @@ class TasksScheduler:
		for connection in connections.connections:
		self._add_connection_to_executor_cache(connection)
		pending_items_to_explore.put(connection)

		explored_items.add(str_item_key)

		elif isinstance(item, Connection):
		@@ -384,7 +365,6 @@ class TasksScheduler:

		connection_key = include_connection(item.connection_id, item.service_id)
		self._add_connection_to_executor_cache(connection)

		self._executor.get_service(item.service_id)
		pending_items_to_explore.put(item.service_id)

		@@ -413,16 +393,20 @@ class TasksScheduler:
		self._add_connection_to_executor_cache(new_connection)

		service_updating_key = self._add_task_if_not_exists(Task_ServiceSetStatus(
		self._executor, service.service_id, ServiceStatusEnum.SERVICESTATUS_UPDATING))
		self._executor, service.service_id, ServiceStatusEnum.SERVICESTATUS_UPDATING
		))

		old_connection_deconfigure_key = self._add_task_if_not_exists(Task_ConnectionDeconfigure(
		self._executor, old_connection.connection_id))
		self._executor, old_connection.connection_id
		))

		new_connection_configure_key = self._add_task_if_not_exists(Task_ConnectionConfigure(
		self._executor, new_connection.connection_id))
		self._executor, new_connection.connection_id
		))

		service_active_key = self._add_task_if_not_exists(Task_ServiceSetStatus(
		self._executor, service.service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE))
		self._executor, service.service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE
		))

		# the old connection deconfiguration depends on service being in updating state
		self._dag.add(old_connection_deconfigure_key, service_updating_key)

src/service/service/tools/OpticalTools.py

+44 −42

File changed.

Preview size limit exceeded, changes collapsed.

src/service/service/tools/object_uuid.py

+1 −1

File changed.

Contains only whitespace changes.