Merge branch 'feat/397-tid-e2e-orchestrator-with-pathcomp-for-p2mp-optical-slices' into 'develop' (e2b02e46) · Commits · TFS / controller

src/e2e_orchestrator/Dockerfile

+3 −0

Original line number	Diff line number	Diff line
		@@ -82,6 +82,9 @@ COPY src/context/client/. context/client/
		COPY src/context/service/database/uuids/. context/service/database/uuids/
		COPY src/service/__init__.py service/__init__.py
		COPY src/service/client/. service/client/
		COPY src/pathcomp/__init__.py pathcomp/__init__.py
		COPY src/pathcomp/frontend/__init__.py pathcomp/frontend/__init__.py
		COPY src/pathcomp/frontend/client/. pathcomp/frontend/client/
		COPY src/e2e_orchestrator/. e2e_orchestrator/

		# Start the service

src/e2e_orchestrator/service/E2EOrchestratorServiceServicerImpl.py

+15 −0

Original line number	Diff line number	Diff line
		@@ -19,6 +19,9 @@ from common.proto.context_pb2 import Empty, Connection, EndPointId
		from common.proto.e2eorchestrator_pb2_grpc import E2EOrchestratorServiceServicer
		from context.client.ContextClient import ContextClient
		from context.service.database.uuids.EndPoint import endpoint_get_uuid
		from common.proto.context_pb2 import ServiceTypeEnum
		from pathcomp.frontend.client.PathCompClient import PathCompClient
		from common.proto.pathcomp_pb2 import PathCompRequest

		LOGGER = logging.getLogger(__name__)

		@@ -33,6 +36,18 @@ class E2EOrchestratorServiceServicerImpl(E2EOrchestratorServiceServicer):
		def Compute(
		self, request: E2EOrchestratorRequest, context: grpc.ServicerContext
		) -> E2EOrchestratorReply:
		if request.service.service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY:
		LOGGER.info("E2E Orchestrator: Detected OPTICAL_CONNECTIVITY service. Calling PathComp.")
		pathcomp_client = PathCompClient()
		pathcomp_req = PathCompRequest()
		pathcomp_req.services.append(request.service)
		pathcomp_reply = pathcomp_client.Compute(pathcomp_req)

		e2e_reply = E2EOrchestratorReply()
		e2e_reply.services.extend(pathcomp_reply.services)
		e2e_reply.connections.extend(pathcomp_reply.connections)
		return e2e_reply

		endpoints_ids = [
		endpoint_get_uuid(endpoint_id)[2]
		for endpoint_id in request.service.service_endpoint_ids

src/pathcomp/frontend/service/OpticalPathComp.py

0 → 100644

+443 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2026 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 json
		import logging
		import uuid

		import requests
		from common.proto.context_pb2 import ConfigActionEnum, ConfigRule, Service
		from common.proto.pathcomp_pb2 import PathCompReply

		from .TopologyTools import get_device_with_driver

		LOGGER = logging.getLogger(__name__)

		def safe_get(d, keys, default=None):
		for key in keys:
		if isinstance(d, dict):
		d = d.get(key, default)
		else:
		return default
		return d


		def group_block(group, action, group_id_override=None, node=None):
		active = "true" if action == 'create' else "false"
		group_id = group_id_override if group_id_override is not None else group.get("group-id", group.get("digital_sub_carriers_group_id", 1))

		if node == "leaf":
		return {
		"digital_sub_carriers_group_id": group_id,
		"digital_sub_carrier_id": [
		{'sub_carrier_id': 1, 'active': active},
		{'sub_carrier_id': 2, 'active': active},
		{'sub_carrier_id': 3, 'active': active},
		{'sub_carrier_id': 4, 'active': active}
		]
		}
		else:
		return {
		"digital_sub_carriers_group_id": group_id,
		"digital_sub_carrier_id": [
		{
		"sub_carrier_id": sid,
		"active": active,
		}
		for sid in group.get("subcarrier-id", [])
		]
		}

		def compute_optical_path(service: Service) -> PathCompReply:
		# Extract intent from config rules
		intent_str = ""
		for cr in service.service_config.config_rules:
		if cr.WhichOneof('config_rule') == 'custom' and cr.custom.resource_key == "intent":
		intent_str = cr.custom.resource_value
		break

		intent = json.loads(intent_str) if intent_str else {}
		action = "create" # Default action

		# Extract src (sender) and dst (receivers) from intent
		services = intent.get("ietf-network-slice-service:network-slice-services", {}).get("slice-service", [])
		source = None
		destination = None

		for srv in services:
		c_groups = srv.get("connection-groups", {}).get("connection-group", [])
		for cg in c_groups:
		constructs = cg.get("connectivity-construct", [])
		for construct in constructs:
		source = construct.get("p2mp-sender-sdp")
		destination = construct.get("p2mp-receiver-sdp")
		if source and destination:
		break
		if source and destination:
		break
		if source and destination:
		break

		if not source or not destination:
		raise Exception("Missing p2mp-sender-sdp or p2mp-receiver-sdp parameters in the intent")

		if isinstance(source, str):
		sources_list = [source]
		else:
		sources_list = list(source)

		if isinstance(destination, str):
		destinations_list = [destination]
		else:
		destinations_list = list(destination)

		# In optical networks, the leaves are often the "sources" of light for the computation or vice-versa.
		# Based on user's instruction: sources = receivers, destinations = sender
		payload = {
		"sources": destinations_list,
		"destinations": sources_list,
		"bitrate": 100,
		"bidirectional": True,
		"band": 200,
		"subcarriers_per_source": [4] * len(destinations_list)
		}

		url = "http://10.30.7.66:31060/OpticalTFS/restconf/operations/tapi-path-computation:compute-p2mp"
		headers = {
		"Content-Type": "application/json",
		"Accept": "/"
		}

		# resp = requests.post(url, headers=headers, json=payload, timeout=15)
		# resp.raise_for_status()
		# resp_json = resp.json()

		# MOCK RESPONSE (If the Optical Controller is down)
		resp_json = {
		"tapi-connectivity:connectivity-service": {
		"connection": [
		{
		"optical-connection-attributes": {
		"central-frequency": 195000000000000,
		"Tx-power": 0,
		"modulation": {
		"modulation-technique": "DP-16QAM",
		"operational-mode": 9,
		"port": "port-1"
		},
		"digital-subcarrier-spacing": 300000000,
		"subcarrier-attributes": {
		"digital-subcarrier-group": [
		{
		"group-id": 1,
		"modulation-technique": "DP-QPSK",
		"central-frequency": 195006250,
		"operational-mode": 4,
		"Tx-power": -99,
		"group-size": 4,
		"port": "port-1",
		"subcarrier-id": [1, 2, 3, 4]
		},
		{
		"group-id": 2,
		"modulation-technique": "DP-QPSK",
		"central-frequency": 195018750,
		"operational-mode": 4,
		"Tx-power": -99,
		"group-size": 4,
		"port": "port-3",
		"subcarrier-id": [5, 6, 7, 8]
		},
		{
		"group-id": 3,
		"modulation-technique": "DP-QPSK",
		"central-frequency": 195031250,
		"operational-mode": 4,
		"Tx-power": -99,
		"group-size": 4,
		"port": "port-5",
		"subcarrier-id": [9, 10, 11, 12]
		}
		]
		}
		},
		"end-point": [
		{
		"direction": "BIDIRECTIONAL",
		"layer-protocol-name": "PHOTONIC_MEDIA",
		"layer-protocol-qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_MC",
		"local-id": "T1.1",
		"service-interface-point": {
		"service-interface-point-uuid": "T2.1"
		},
		"tapi-photonic-media:media-channel-connectivity-service-end-point-spec": {
		"mc-config": {
		"spectrum": {
		"center-frequency": 195000000000000,
		"frequency-constraint": {
		"adjustment-granularity": "G_6_25GHZ",
		"grid-type": "FLEX"
		}
		}
		}
		}
		},
		{
		"direction": "BIDIRECTIONAL",
		"layer-protocol-name": "PHOTONIC_MEDIA",
		"layer-protocol-qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_MC",
		"local-id": "T2.1",
		"service-interface-point": {
		"service-interface-point-uuid": "T1.1"
		},
		"tapi-photonic-media:media-channel-connectivity-service-end-point-spec": {
		"mc-config": {
		"spectrum": {
		"center-frequency": 195006250,
		"frequency-constraint": {
		"adjustment-granularity": "G_6_25GHZ",
		"grid-type": "FLEX"
		}
		}
		}
		}
		},
		{
		"direction": "BIDIRECTIONAL",
		"layer-protocol-name": "PHOTONIC_MEDIA",
		"layer-protocol-qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_MC",
		"local-id": "T1.2",
		"service-interface-point": {
		"service-interface-point-uuid": "T2.1"
		},
		"tapi-photonic-media:media-channel-connectivity-service-end-point-spec": {
		"mc-config": {
		"spectrum": {
		"center-frequency": 195018750,
		"frequency-constraint": {
		"adjustment-granularity": "G_6_25GHZ",
		"grid-type": "FLEX"
		}
		}
		}
		}
		},
		{
		"direction": "BIDIRECTIONAL",
		"layer-protocol-name": "PHOTONIC_MEDIA",
		"layer-protocol-qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_MC",
		"local-id": "T2.1",
		"service-interface-point": {
		"service-interface-point-uuid": "T1.2"
		},
		"tapi-photonic-media:media-channel-connectivity-service-end-point-spec": {
		"mc-config": {
		"spectrum": {
		"center-frequency": 195018750,
		"frequency-constraint": {
		"adjustment-granularity": "G_6_25GHZ",
		"grid-type": "FLEX"
		}
		}
		}
		}
		},
		{
		"direction": "BIDIRECTIONAL",
		"layer-protocol-name": "PHOTONIC_MEDIA",
		"layer-protocol-qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_MC",
		"local-id": "T1.3",
		"service-interface-point": {
		"service-interface-point-uuid": "T2.1"
		},
		"tapi-photonic-media:media-channel-connectivity-service-end-point-spec": {
		"mc-config": {
		"spectrum": {
		"center-frequency": 195031250,
		"frequency-constraint": {
		"adjustment-granularity": "G_6_25GHZ",
		"grid-type": "FLEX"
		}
		}
		}
		}
		},
		{
		"direction": "BIDIRECTIONAL",
		"layer-protocol-name": "PHOTONIC_MEDIA",
		"layer-protocol-qualifier": "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_MC",
		"local-id": "T2.1",
		"service-interface-point": {
		"service-interface-point-uuid": "T1.3"
		},
		"tapi-photonic-media:media-channel-connectivity-service-end-point-spec": {
		"mc-config": {
		"spectrum": {
		"center-frequency": 195031250,
		"frequency-constraint": {
		"adjustment-granularity": "G_6_25GHZ",
		"grid-type": "FLEX"
		}
		}
		}
		}
		}
		]
		}
		]
		}
		}

		# Generate Rules
		src_name = source # T2.1
		dest_list = destinations_list # T1.1, T1.2
		# Extract destinations from response end-points if available
		try:
		endpoints = resp_json.get("tapi-connectivity:connectivity-service", {}).get("connection", [{}])[0].get("end-point", [])
		extracted_dests = []
		for ep in endpoints:
		local_id = ep.get("local-id")
		if local_id and local_id != src_name and local_id not in extracted_dests:
		extracted_dests.append(local_id)
		if extracted_dests:
		dest_list = extracted_dests
		except Exception:
		pass

		dest_str = ",".join(dest_list)
		config_rules = []

		network_slice_uuid_str = f"{src_name}_to_{dest_str}"
		tunnel_uuid = str(uuid.uuid5(uuid.NAMESPACE_DNS, network_slice_uuid_str))

		provisionamiento = {
		"network-slice-uuid": network_slice_uuid_str,
		"viability": True,
		"actions": []
		}
		try:
		attributes = resp_json.get("tapi-connectivity:connectivity-service", {}).get("connection", [{}])[0].get("optical-connection-attributes", {})
		groups = attributes.get("subcarrier-attributes", {}).get("digital-subcarrier-group", [])
		operational_mode = attributes.get("modulation", {}).get("operational-mode", 9)
		except Exception:
		# Provide fallback if mock doesn't match perfectly
		groups = resp_json.get("digital-subcarrier-groups", [])
		operational_mode = resp_json.get("op-mode", 9)

		hub_groups = [
		group_block(group, action, group_id_override=index + 1)
		for index, group in enumerate(groups)
		]

		hub_freq_raw = attributes.get("central-frequency", 195000000000000)
		hub_freq = int(hub_freq_raw / 1e6) if hub_freq_raw > 1e10 else int(hub_freq_raw)

		hub = {
		"name": "channel-1",
		"frequency": hub_freq,
		"target_output_power": attributes.get("Tx-power", 0),
		"operational_mode": operational_mode,
		"operation": "merge",
		"digital_sub_carriers_group": hub_groups
		}

		leaves = []
		print("dest_list:", dest_list)
		print("groups:", groups)
		for dest, group in zip(dest_list, groups):
		port = group.get("port", "port-1")
		if port.startswith("port-"):
		name = f"channel-{port.split('-')[1]}"
		else:
		name = "channel-1"

		freq_raw = group.get("central-frequency", 195006250)
		freq = int(freq_raw / 1e6) if freq_raw > 1e10 else int(freq_raw)

		leaf = {
		"name": name,
		"frequency": freq,
		"target_output_power": group.get("Tx-power", 0),
		"operational_mode": int(group.get("operational-mode", operational_mode)),
		"operation": "merge",
		"digital_sub_carriers_group": [group_block(group, action, group_id_override=1, node="leaf")]
		}
		leaves.append(leaf)

		final_json = {"components": [hub] + leaves}

		# Add transceiver activation action
		provisionamiento["actions"].append({
		"type": "XR_AGENT_ACTIVATE_TRANSCEIVER",
		"layer": "OPTICAL",
		"content": final_json,
		"controller-uuid": "IPoWDM Controller"
		})

		# Extract IP configuration from intent for L3 VPN setup
		nodes = {}
		sdp_list = intent.get('ietf-network-slice-service:network-slice-services', {}).get('slice-service', [{}])[0].get('sdps', {}).get('sdp', [])

		for sdp in sdp_list:
		node = sdp.get('id')
		attachments = sdp.get('attachment-circuits', {}).get('attachment-circuit', [])
		for ac in attachments:
		ip = ac.get('ac-ipv4-address', None)
		prefix = ac.get('ac-ipv4-prefix-length', None)
		vlan = 500 # Fixed VLAN ID
		nodes[node] = {
		"ip-address": ip,
		"ip-mask": prefix,
		"vlan-id": vlan
		}

		# Add L3 VPN configuration action for P2MP topology
		if src_name in nodes:
		content = {
		"tunnel-uuid": tunnel_uuid,
		"src-node-uuid": src_name,
		"src-ip-address": nodes[src_name]["ip-address"],
		"src-ip-mask": str(nodes[src_name]["ip-mask"]),
		"src-vlan-id": nodes[src_name]["vlan-id"],
		}

		for i, dest in enumerate(dest_list):
		if dest in nodes:
		content[f"dest{i+1}-node-uuid"] = dest
		content[f"dest{i+1}-ip-address"] = nodes[dest]["ip-address"]
		content[f"dest{i+1}-ip-mask"] = str(nodes[dest]["ip-mask"])
		content[f"dest{i+1}-vlan-id"] = nodes[dest]["vlan-id"]
		controller_ip = get_device_with_driver()

		provisionamiento["actions"].append({
		"type": "CONFIG_VPNL3",
		"layer": "IP",
		"content": content,
		"controller_uuid": controller_ip
		})

		config_rules.append(provisionamiento)

		reply = PathCompReply()
		reply_svc = Service()
		reply_svc.CopyFrom(service)

		cr = ConfigRule()
		cr.action = ConfigActionEnum.CONFIGACTION_SET
		cr.custom.resource_key = "optical_path_result"
		cr.custom.resource_value = json.dumps(config_rules)
		reply_svc.service_config.config_rules.append(cr)
		reply.services.append(reply_svc)

		return reply

src/pathcomp/frontend/service/PathCompServiceServicerImpl.py

+7 −0

Original line number	Diff line number	Diff line
		@@ -28,6 +28,8 @@ from pathcomp.frontend.Config import is_forecaster_enabled
		#from context.client.ContextClient import ContextClient
		from pathcomp.frontend.service.TopologyTools import get_pathcomp_topology_details
		from pathcomp.frontend.service.algorithms.Factory import get_algorithm
		from pathcomp.frontend.service.OpticalPathComp import compute_optical_path
		from common.proto.context_pb2 import ServiceTypeEnum

		LOGGER = logging.getLogger(__name__)

		@@ -45,6 +47,11 @@ class PathCompServiceServicerImpl(PathCompServiceServicer):
		def Compute(self, request : PathCompRequest, context : grpc.ServicerContext) -> PathCompReply:
		LOGGER.debug('[Compute] begin ; request = {:s}'.format(grpc_message_to_json_string(request)))

		if len(request.services) > 0 and request.services[0].service_type == ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY:
		LOGGER.info('[Compute] Intercepting OPTICAL_CONNECTIVITY request...')
		return compute_optical_path(request.services[0])


		#context_client = ContextClient()
		# TODO: improve definition of topologies; for interdomain the current topology design might be not convenient
		#if (len(request.services) == 1) and is_inter_domain(context_client, request.services[0].service_endpoint_ids):

src/pathcomp/frontend/service/TopologyTools.py

+35 −1

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

		import logging, math
		import logging, math, os, socket, requests
		from typing import Dict, Optional
		from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME, ServiceNameEnum
		from common.Settings import ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, find_environment_variables, get_env_var_name
		@@ -25,6 +25,7 @@ from common.tools.grpc.Tools import grpc_message_to_json_string
		from context.client.ContextClient import ContextClient
		from forecaster.client.ForecasterClient import ForecasterClient


		LOGGER = logging.getLogger(__name__)

		def get_service_schedule(service : Service) -> Optional[Constraint_Schedule]:
		@@ -96,3 +97,36 @@ def get_pathcomp_topology_details(request : PathCompRequest, allow_forecasting :
		link.attributes.total_capacity_gbps = total_capacity_gbps

		return topology_details

		# The HOST IP of the TFS controller is needed to query the devices and get the
		# IP address of the controller to be used in the path provisioning.
		ENVVAR_TFS_API_HOST = '10.95.89.50'



		def get_device_with_driver(driver_name: str = 'DEVICEDRIVER_IETF_L3VPN', timeout: int = 10):
		"""Query the TFS controller REST API and return the _connect/address value of the first device with the given driver.

		Example: GET http://<vm-ip>/tfs-api/devices

		Returns the _connect/address string or None.
		"""
		url = f'http://{ENVVAR_TFS_API_HOST}:80/tfs-api/devices'
		try:
		logging.debug("Requesting devices from %s", url)
		resp = requests.get(url, timeout=timeout)
		resp.raise_for_status()
		data = resp.json()
		except Exception as e:
		logging.warning("Error fetching devices from %s: %s", url, e)
		return None

		devices = data.get('devices', []) if isinstance(data, dict) else []
		for dev in devices:
		drivers = dev.get('device_drivers') or []
		if isinstance(drivers, list) and driver_name in drivers:
		device_name = dev.get('name')
		return device_name

		logging.info("No device with driver %s found on %s", driver_name, url)
		return None