Merge branch 'feat/tid-media-channel-service' of... (61276e61) · Commits · TFS / controller

src/device/service/drivers/ietf_l3vpn/IetfL3VpnDriver.py

+6 −1

Original line number	Diff line number	Diff line
		@@ -189,10 +189,15 @@ class IetfL3VpnDriver(_Driver):
		if len(resources) == 0: return results
		with self.__lock:
		if 'ipowdm' in str(resources):
		# Build controller URL from device address and port
		scheme = self.settings.get('scheme', 'http')
		controller_url = f"{scheme}://{self.address}:{self.port}"
		LOGGER.info("IPoWDM: Using controller URL: %s", controller_url)

		for resource in resources:
		if 'ipowdm' in str(resource):
		try:
		create_request(resource)
		create_request(resource, controller_url)
		LOGGER.info('Request created successfully')
		results.append((resource, True))
		except Exception as e:

src/device/service/drivers/ietf_l3vpn/TfsApiClient.py

+3 −2

Original line number	Diff line number	Diff line
		@@ -144,7 +144,8 @@ class TfsApiClient(RestApiClient):
		cr_rk : str = cr['custom']['resource_key']
		if not cr_rk.startswith('/endpoints/endpoint['): continue
		settings = json.loads(cr['custom']['resource_value'])
		ep_name = settings['name']
		ep_name = settings.get('name', settings.get('uuid'))
		if ep_name:
		config_rule_dict[ep_name] = settings

		for json_endpoint in json_device['device_endpoints']:

src/device/service/drivers/ietf_l3vpn/templates/tools.py

+391 −72

Original line number	Diff line number	Diff line
		@@ -17,80 +17,399 @@ import json
		import logging

		import requests
		import uuid
		from typing import Dict, List, Optional, Tuple
		from concurrent.futures import ThreadPoolExecutor

		LOGGER = logging.getLogger(__name__)

		def create_request(resource_value):
		""" Create and send HTTP request based on a JSON template and provided resource value.
		The JSON template is expected to be in the same directory as this script, named 'ipowdm.json'.
		Example resource_value:
		{"rule_set": {
		"uuid": "unique-service-uuid",
		"bw": 100,
		"src": [{"uuid": "src-device-uuid", "ip_address": "192.168.1.1", "ip_mask": "24", "vlan_id": 100, "power": 10, "frequency": 193100}],
		"dst": [{"uuid": "dst-device-uuid", "ip_address": "192.168.3.3", "ip_mask": "24", "vlan_id": 100, "power": 10, "frequency": 193100}]
		}}
		The src and dst fields are lists to accommodate future extensions for multi-endpoint scenarios.
		The request is sent to a predefined URL with appropriate headers.
		Returns a response-like object with status_code and text attributes.
		In case of error, returns a SimpleNamespace with status_code 500 and the error message in text.

		Note: The actual HTTP request sending is currently mocked for testing purposes.
		The URL and headers are hardcoded for demonstration and should be adapted as needed.
		"""

		LOGGER.info("Creating request for resource_value: %s", resource_value)
		LOGGER.info("Resource value type: %s", type(resource_value))
		BaseDir = os.path.dirname(os.path.abspath(__file__))
		json_path = os.path.join(BaseDir, 'ipowdm.json')
		with open(json_path, 'r', encoding='utf-8') as f:
		template = json.load(f)

		LOGGER.info("Sending POST DSTINATION request with payload: %s", json.dumps(template, indent=2))

		response = FakeResponse()
		# tfs_post(template)
		# Default configuration - URLs will be passed from driver with actual controller IP
		DEFAULT_CONTROLLER_URL = 'http://127.0.0.1:80'

		HEADERS = {
		"Accept": "application/yang-data+json",
		"Content-Type": "application/yang-data+json"
		}

		executor = ThreadPoolExecutor()


		# =============================================================================
		# L3NM Service Generation
		# =============================================================================

		def generate_l3vpn_site(endpoint: Dict, vpn_id: str, role: str = "spoke-role") -> Dict:
		"""
		Generate a single L3VPN site configuration from an endpoint.

		Args:
		endpoint: Dict with uuid, ip_address, ip_mask, vlan_id
		vpn_id: VPN identifier
		role: "spoke-role" or "hub-role"

		Returns:
		Dict with L3VPN site configuration
		"""
		return {
		"site-id": endpoint["uuid"],
		"management": {"type": "ietf-l3vpn-svc:provider-managed"},
		"locations": {"location": [{"location-id": f"location-{endpoint['uuid']}"}]},
		"devices": {"device": [{
		"device-id": endpoint.get("ip_address", "127.0.0.1"),
		"location": f"location-{endpoint['uuid']}"
		}]},
		"routing-protocols": {"routing-protocol": [{
		"type": "ietf-l3vpn-svc:static",
		"static": {
		"cascaded-lan-prefixes": {
		"ipv4-lan-prefixes": [{
		"lan": f"{endpoint['ip_address']}{endpoint['ip_mask']}",
		"lan-tag": f"vlan{endpoint['vlan_id']}",
		"next-hop": endpoint.get("gateway", endpoint['ip_address'])
		}]
		}
		}
		}]},
		"site-network-accesses": {
		"site-network-access": [{
		"site-network-access-id": str(endpoint['vlan_id']),
		"site-network-access-type": "ietf-l3vpn-svc:multipoint",
		"device-reference": endpoint.get("ip_address", "127.0.0.1"),
		"vpn-attachment": {
		"vpn-id": vpn_id,
		"site-role": f"ietf-l3vpn-svc:{role}"
		},
		"ip-connection": {
		"ipv4": {
		"address-allocation-type": "ietf-l3vpn-svc:static-address",
		"addresses": {
		"provider-address": endpoint.get("provider_ip", endpoint['ip_address']),
		"customer-address": endpoint['ip_address'],
		"prefix-length": int(endpoint['ip_mask'].replace('/', '')) if isinstance(endpoint['ip_mask'], str) else endpoint['ip_mask']
		}
		}
		},
		"service": {
		"svc-mtu": 1500,
		"svc-input-bandwidth": 1000000000,
		"svc-output-bandwidth": 1000000000,
		"qos": {
		"qos-profile": {
		"classes": {
		"class": [{
		"class-id": "qos-realtime",
		"direction": "ietf-l3vpn-svc:both",
		"latency": {"latency-boundary": 10},
		"bandwidth": {"guaranteed-bw-percent": 100}
		}]
		}
		}
		}
		}
		}]
		}
		}


		def generate_l3vpn_service(src: Dict, dst: Dict, vpn_id: str) -> Dict:
		"""
		Generate a complete L3VPN service with source and destination sites.

		Args:
		src: Source endpoint configuration
		dst: Destination endpoint configuration
		vpn_id: VPN service identifier

		Returns:
		Dict with complete L3VPN service configuration
		"""
		return {
		"ietf-l3vpn-svc:l3vpn-svc": {
		"vpn-services": {
		"vpn-service": [{"vpn-id": vpn_id}]
		},
		"sites": {
		"site": [
		generate_l3vpn_site(src, vpn_id, "spoke-role"),
		generate_l3vpn_site(dst, vpn_id, "hub-role")
		]
		}
		}
		}


		def send_l3vpn_service(service_config: Dict, controller_url: str, service_uuid: str) -> requests.Response:
		"""
		Send L3VPN/L3NM service configuration to the NBI.

		Args:
		service_config: L3VPN service configuration dict
		controller_url: Controller URL (required)
		service_uuid: Service UUID for the endpoint

		Returns:
		requests.Response object or status code
		"""
		url = f"{controller_url}/restconf/ipowdm/v1/l3nm/{service_uuid}"
		headers = {
		'accept': 'application/json',
		'Content-Type': 'application/json'
		}

		LOGGER.info("Sending L3NM service to %s", url)
		LOGGER.debug("L3NM payload: %s", json.dumps(service_config, indent=2))

		try:
		response = requests.post(url=url, headers=headers, json=service_config, timeout=30)
		LOGGER.info("L3NM response: %s - %s", response.status_code, response.text)
		return response
		except Exception as e:
		LOGGER.error("Failed to send L3NM service: %s", str(e))
		raise


		class FakeResponse:
		"""_Fake response object for testing purposes."""
		def __init__(self):
		self.ok = True
		self.status_code = 200
		self.text = '{"message": "OK"}'
		# =============================================================================
		# Pluggables/Transceiver Configuration
		# =============================================================================

		def generate_pluggable_config(component: Dict, device_id: str) -> Dict:
		"""
		Generate pluggable/transceiver configuration from a component.

		Args:
		component: Dict with name, frequency, target_output_power, operational_mode, operation
		device_id: Device identifier

		Returns:
		Dict with pluggable configuration in format {device, config}
		"""
		return {
		"device": device_id,
		"config": {
		"frequency": component.get("frequency", 193100000),
		"target-output-power": component.get("target_output_power", 0.0),
		"operational-mode": component.get("operational_mode", 0),
		"name": component.get("name", ""),
		"operation": component.get("operation", "activate")
		}
		}

		def json(self):
		"""Return a sample JSON response."""
		return {"message": "OK"}

		def tfs_post(request):
		def configure_pluggable(component: Dict, device_id: str, controller_url: str, service_uuid: str):
		"""
		Send a POST request to the TeraFlow Service Orchestrator.
		Configure a pluggable/transceiver via NBI.

		Args:
		ip (str): IP address of the TeraFlow Service Orchestrator.
		request (dict): The request payload to be sent.
		component: Component configuration dict
		device_id: Device identifier
		controller_url: Controller URL (required)
		service_uuid: Service UUID for the endpoint

		Returns:
		dict: The response from the TeraFlow Service Orchestrator.
		"""
		user="admin"
		password="admin"
		token=""
		session = requests.Session()
		session.auth = (user, password)
		url=f'http://10.95.86.62/webui'
		response=session.get(url=url)
		for item in response.iter_lines():
		if"csrf_token" in str(item):
		string=str(item).split('<input id="csrf_token" name="csrf_token" type="hidden" value=')[1]
		token=string.split(">")[0].strip('"')
		logging.debug("csrf token %s",token)

		files = {'descriptors': ("data.json", json.dumps(request).encode(
		"utf-8"), "application/json")}
		token={'csrf_token':token}
		response = session.post(url,files=files,data=token,timeout=60)
		logging.debug("Http response: %s",response.text)
		Response object or status code
		"""
		url = f"{controller_url}/restconf/ipowdm/v1/pluggables/{service_uuid}"

		config = generate_pluggable_config(component, device_id)

		LOGGER.info("Configuring pluggable for service %s on device %s", service_uuid, device_id)
		LOGGER.debug("Pluggable config: %s", json.dumps(config, indent=2))

		try:
		response = requests.post(url, json=config, headers=HEADERS, timeout=30)
		LOGGER.info("Pluggable response: %s - %s", response.status_code, response.text)
		return response
		except Exception as e:
		LOGGER.error("Failed to configure pluggable on %s: %s", device_id, str(e))
		return None


		def configure_all_pluggables(components: List[Dict], device_ids: List[str], controller_url: str = None) -> List[Optional[requests.Response]]:
		"""
		Configure all pluggables/transceivers in parallel.

		Args:
		components: List of component configurations
		device_ids: List of device identifiers (same order as components)
		controller_url: Optional controller URL override

		Returns:
		List of response objects
		"""
		responses = []
		for component, device_id in zip(components, device_ids):
		response = configure_pluggable(component, device_id, controller_url)
		responses.append(response)
		return responses


		# =============================================================================
		# IPoWDM Service Processing (Main Entry Point)
		# =============================================================================

		def parse_ipowdm_data(resource_value) -> Tuple[List[Dict], List[Dict], List[Dict], str]:
		"""
		Parse IPoWDM resource data into structured components.

		Args:
		resource_value: Raw resource value (tuple or dict)

		Returns:
		Tuple of (src_endpoints, dst_endpoints, transceiver_components, service_uuid)
		"""
		# Handle different input formats
		if isinstance(resource_value, tuple):
		data = resource_value[1] if isinstance(resource_value[1], dict) else json.loads(resource_value[1])
		elif isinstance(resource_value, dict):
		data = resource_value
		else:
		data = json.loads(resource_value)

		# Extract from rule_set if present, otherwise use direct structure
		if 'rule_set' in data:
		data = data['rule_set']

		src_endpoints = data.get('src', [])
		dst_endpoints = data.get('dst', [])
		components = data.get('transceiver', {}).get('components', [])
		service_uuid = data.get('uuid', 'unknown')

		return src_endpoints, dst_endpoints, components, service_uuid


		def process_ipowdm_service(resource_value, controller_url: str) -> Dict:
		"""
		Process an IPoWDM service request, generating L3NM services and configuring pluggables.

		This is the main entry point for IPoWDM processing. It:
		1. Parses the incoming IPoWDM data
		2. Configures all pluggables/transceivers
		3. Generates and sends L3VPN services for each src-dst pair

		Args:
		resource_value: IPoWDM resource data
		controller_url: Controller URL for L3VPN and pluggable configuration

		Returns:
		Dict with processing results
		"""
		LOGGER.info("Processing IPoWDM service request")

		# Parse the incoming data
		src_endpoints, dst_endpoints, components, service_uuid = parse_ipowdm_data(resource_value)

		LOGGER.info("IPoWDM Service UUID: %s", service_uuid)
		LOGGER.info("Source endpoints: %d, Destination endpoints: %d, Components: %d",
		len(src_endpoints), len(dst_endpoints), len(components))

		results = {
		"service_uuid": service_uuid,
		"pluggable_results": [],
		"l3vpn_results": []
		}

		# Step 1: Configure pluggables/transceivers
		if components:
		all_endpoints = src_endpoints + dst_endpoints
		device_ids = [ep.get('uuid', f'device-{i}') for i, ep in enumerate(all_endpoints)]

		LOGGER.info("Configuring %d pluggables", len(components))
		for i, (component, device_id) in enumerate(zip(components, device_ids)):
		LOGGER.debug("Configuring pluggable %d on device %s: %s", i, device_id, component)
		try:
		# Generate unique UUID for each pluggable service
		pluggable_uuid = f"{service_uuid}-pluggable-{device_id}"
		response = configure_pluggable(component, device_id, controller_url, pluggable_uuid)
		# Handle both mock (int) and real response objects
		status_code = response if isinstance(response, int) else (response.status_code if response else 0)
		results["pluggable_results"].append({
		"device_id": device_id,
		"status": status_code,
		"success": status_code in [200, 201, 204]
		})
		except Exception as e:
		LOGGER.error("Pluggable configuration failed for %s: %s", device_id, str(e))
		results["pluggable_results"].append({
		"device_id": device_id,
		"status": "error",
		"success": False,
		"error": str(e)
		})

		# Step 2: Generate and send L3VPN services
		for src in src_endpoints:
		for dst in dst_endpoints:
		vpn_id = f"L3VPN_{src['uuid']}_{dst['uuid']}"
		LOGGER.info("Creating L3VPN service: %s", vpn_id)

		try:
		service_config = generate_l3vpn_service(src, dst, vpn_id)
		# Generate unique UUID for each L3NM service
		l3nm_uuid = f"{service_uuid}-l3nm-{src['uuid']}-{dst['uuid']}"
		response = send_l3vpn_service(service_config, controller_url, l3nm_uuid)
		# Handle both mock (int) and real response objects
		status_code = response if isinstance(response, int) else (response.status_code if response else 0)
		results["l3vpn_results"].append({
		"vpn_id": vpn_id,
		"src": src['uuid'],
		"dst": dst['uuid'],
		"status": status_code,
		"success": status_code in [200, 201, 204]
		})
		except Exception as e:
		LOGGER.error("L3VPN creation failed for %s: %s", vpn_id, str(e))
		results["l3vpn_results"].append({
		"vpn_id": vpn_id,
		"src": src['uuid'],
		"dst": dst['uuid'],
		"status": "error",
		"success": False,
		"error": str(e)
		})

		LOGGER.info("IPoWDM processing complete. Results: %s", json.dumps(results, indent=2))
		return results


		# =============================================================================
		# Legacy Entry Point (for backward compatibility)
		# =============================================================================

		def create_request(resource_value, controller_url: str = None):
		"""
		Legacy entry point for IPoWDM processing.
		Maintained for backward compatibility with existing driver code.

		Args:
		resource_value: IPoWDM resource data
		controller_url: Controller URL (IP:port of the L3VPN controller device)

		Returns:
		Processing results dict
		"""
		LOGGER.info("create_request called")
		LOGGER.info("Resource value: %s", resource_value)
		LOGGER.info("Controller URL: %s", controller_url)

		url = controller_url or DEFAULT_CONTROLLER_URL
		return process_ipowdm_service(resource_value, url)


		# =============================================================================
		# Utility Functions
		# =============================================================================

		def generate_l3vpn_template_pair(src: Dict, dst: Dict, vpn_id: str) -> Dict:
		"""
		Legacy function for generating L3VPN template.
		Wrapper around generate_l3vpn_service for backward compatibility.
		"""
		return generate_l3vpn_service(src, dst, vpn_id)


		def patch_optical_channel_frequency(data: Dict, device_id: str) -> Optional[requests.Response]:
		"""
		Legacy function for patching optical channel frequency.
		Wrapper around configure_pluggable for backward compatibility.
		"""
		return configure_pluggable(data, device_id)

src/device/service/drivers/transport_api/TfsApiClient.py

0 → 100644

+135 −0

File added.

Preview size limit exceeded, changes collapsed.

src/device/service/drivers/transport_api/TransportApiDriver.py

+82 −24

File changed.

Preview size limit exceeded, changes collapsed.