GSMA edge zones validated (3 tests) (5a788b90) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/edgecloud/adapters/aeros/client.py

+18 −21

Original line number	Diff line number	Diff line
		@@ -5,10 +5,8 @@
		# - Vasilis Pitsilis (vpitsilis@dat.demokritos.gr, vpitsilis@iit.demokritos.gr)
		# - Andreas Sakellaropoulos (asakellaropoulos@iit.demokritos.gr)
		##
		from time import sleep
		import uuid
		import json
		import re
		from typing import Any, Dict, List, Optional
		from collections import defaultdict
		from pydantic import ValidationError
		@@ -18,9 +16,9 @@ from sunrise6g_opensdk.edgecloud.adapters.aeros import config
		from sunrise6g_opensdk.edgecloud.adapters.aeros.utils import (
		urn_to_uuid, encode_app_instance_name)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_client import ContinuumClient
		from sunrise6g_opensdk.edgecloud.adapters.aeros.converters import gsma2aeros_converter
		from sunrise6g_opensdk.edgecloud.adapters.aeros.storageManagement import inMemoryStorage
		from sunrise6g_opensdk.edgecloud.adapters.aeros.converters import camara2aeros_converter
		from sunrise6g_opensdk.edgecloud.adapters.aeros.converters import (
		camara2aeros_converter, gsma2aeros_converter, aeros2gsma_zone_details)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.storageManagement.appStorageManager import (
		AppStorageManager, )
		from sunrise6g_opensdk.edgecloud.adapters.errors import EdgeCloudPlatformError
		@@ -246,12 +244,12 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		domain,
		"reservedComputeResources": [{
		"cpuArchType": "ISA_X86_64",
		"numCPU": str(total_cpu),
		"numCPU": int(total_cpu),
		"memory": total_ram,
		}],
		"computeResourceQuotaLimits": [{
		"cpuArchType": "ISA_X86_64",
		"numCPU": str(total_cpu * 2), # Assume quota is 2x total?
		"numCPU": int(total_cpu * 2), # Assume quota is 2x total?
		"memory": total_ram * 2,
		}],
		"flavoursSupported":
		@@ -609,15 +607,12 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		aeros_response = aeros_client.query_entities(ngsild_params)
		aeros_domains = aeros_response.json()
		zone_list = [{
		"zoneId":
		domain["id"],
		"status":
		domain["domainStatus"].split(":")[-1].lower(),
		"geographyDetails":
		"NOT_USED",
		"zoneId": domain["id"],
		"geolocation": "NOT_Available",
		"geographyDetails": domain["description"],
		} for domain in aeros_domains]
		return build_custom_http_response(
		status_code=aeros_domains.status_code,
		status_code=aeros_response.status_code,
		content=zone_list,
		headers={"Content-Type": self.content_type_gsma},
		encoding=self.encoding_gsma,
		@@ -658,15 +653,15 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		# Transform the IEs to required format
		# per domain and append to response list
		camara_response = []
		gsma_response = []
		for domain, ies in grouped_by_domain.items():
		result = self.transform_infrastructure_elements(domain_ies=ies,
		result = aeros2gsma_zone_details.transformer(domain_ies=ies,
		domain=domain)
		camara_response.append(result)
		gsma_response.append(result)
		# Return the transformed response
		return build_custom_http_response(
		status_code=aeros_response.status_code,
		content=camara_response,
		content=gsma_response,
		headers={"Content-Type": self.content_type_gsma},
		encoding=self.encoding_gsma,
		url=aeros_response.url,
		@@ -704,14 +699,16 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		aeros_domain_ies = aeros_response.json()
		# Transform the infrastructure elements into the required format
		# and return the details of the edge cloud zone
		camara_response = self.transform_infrastructure_elements(
		# camara_response = self.transform_infrastructure_elements(
		# domain_ies=aeros_domain_ies, domain=zone_id)
		gsma_response = aeros2gsma_zone_details.transformer(
		domain_ies=aeros_domain_ies, domain=zone_id)
		if config.DEBUG:
		self.logger.debug("Transformed response: %s", camara_response)
		self.logger.debug("Transformed response: %s", gsma_response)
		# Return the transformed response
		return build_custom_http_response(
		status_code=aeros_response.status_code,
		content=camara_response,
		content=gsma_response,
		headers={"Content-Type": "application/json"},
		encoding=aeros_response.encoding,
		url=aeros_response.url,

src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/aeros2gsma_zone_details.py

0 → 100644

+154 −0

Original line number	Diff line number	Diff line
		'''
		aeros2gsma_zone_details.py
		'''
		from typing import List, Dict, Any


		def transformer(domain_ies: List[Dict[str, Any]],
		domain: str) -> Dict[str, Any]:
		"""
		Transform aerOS InfrastructureElements into GSMA ZoneRegisteredData structure.
		:param domain_ies: List of aerOS InfrastructureElement dicts
		:param domain: The ID of the edge cloud zone (zoneId)
		:return: Dict matching gsma_schemas.ZoneRegisteredData (JSON-serializable)
		"""

		def map_cpu_arch_to_isa(urn: str) -> str:
		"""
		Map aerOS cpuArchitecture URN to GSMA ISA_* literal.
		Examples:
		'urn:ngsi-ld:CpuArchitecture:x64' -> 'ISA_X86_64'
		'urn:ngsi-ld:CpuArchitecture:arm64' -> 'ISA_ARM_64'
		'urn:ngsi-ld:CpuArchitecture:arm32' -> 'ISA_ARM_64' (closest)
		'urn:ngsi-ld:CpuArchitecture:x86' -> 'ISA_X86'
		Fallback: 'ISA_X86_64'
		"""
		if not isinstance(urn, str):
		return "ISA_X86_64"
		tail = urn.split(":")[-1].lower()
		if tail in ("x64", "x86_64", "amd64"):
		return "ISA_X86_64"
		if tail in ("x86", "i386", "i686"):
		return "ISA_X86"
		if tail in ("arm64", "aarch64"):
		return "ISA_ARM_64"
		if tail in ("arm32", "arm"):
		# GSMA only has ARM_64 vs X86/X86_64; pick closest
		return "ISA_ARM_64"
		return "ISA_X86_64"

		def map_cpu_arch_to_ostype_arch(urn: str) -> str:
		"""
		Map aerOS cpuArchitecture URN to OSType.architecture literal: 'x86_64' or 'x86'.
		Use 'x86_64' for x64/arm64 (closest allowed), and 'x86' for x86/arm32.
		"""
		if not isinstance(urn, str):
		return "x86_64"
		tail = urn.split(":")[-1].lower()
		if tail in ("x64", "x86_64", "amd64", "arm64", "aarch64"):
		return "x86_64"
		if tail in ("x86", "i386", "i686", "arm32", "arm"):
		return "x86"
		return "x86_64"

		def map_os_distribution(_urn: str) -> str:
		"""
		aerOS uses 'urn:ngsi-ld:OperatingSystem:Linux' etc.
		map Linux -> UBUNTU (assume), else OTHER.
		"""
		if isinstance(_urn, str) and _urn.split(":")[-1].lower() == "linux":
		return "UBUNTU"
		return "OTHER"

		def default_os_version(dist: str) -> str:
		# You asked to assume Ubuntu 22.04 LTS for Linux
		return "OS_VERSION_UBUNTU_2204_LTS" if dist == "UBUNTU" else "OTHER"

		# Totals (aggregate over elements)
		total_cpu = 0
		total_ram = 0
		total_disk = 0
		total_available_ram = 0
		total_available_disk = 0

		flavours_supported: List[Dict[str, Any]] = []
		seen_cpu_isas: set[str] = set()

		for element in domain_ies:
		cpu_cores = int(element.get("cpuCores", 0) or 0)
		ram_cap = int(element.get("ramCapacity", 0) or 0) # MB?
		avail_ram = int(element.get("availableRam", 0) or 0) # MB?
		disk_cap = int(element.get("diskCapacity", 0)
		or 0) # MB/GB? (pass-through)
		avail_disk = int(element.get("availableDisk", 0) or 0)

		total_cpu += cpu_cores
		total_ram += ram_cap
		total_available_ram += avail_ram
		total_disk += disk_cap
		total_available_disk += avail_disk

		cpu_arch_urn = element.get("cpuArchitecture", "")
		os_urn = element.get("operatingSystem", "")

		isa = map_cpu_arch_to_isa(cpu_arch_urn)
		seen_cpu_isas.add(isa)
		ost_arch = map_cpu_arch_to_ostype_arch(cpu_arch_urn)
		dist = map_os_distribution(os_urn)
		ver = default_os_version(dist)

		# Create a flavour per machine
		flavour = {
		"flavourId":
		f"{element.get('hostname', 'host')}-{element.get('containerTechnology', 'CT')}",
		"cpuArchType":
		isa, # Literal ISA_*
		"supportedOSTypes": [{
		"architecture": ost_arch, # 'x86_64' or 'x86'
		"distribution": dist, # 'UBUNTU' or 'OTHER'
		"version": ver, # 'OS_VERSION_UBUNTU_2204_LTS' or 'OTHER'
		"license": "OS_LICENSE_TYPE_FREE",
		}],
		"numCPU":
		cpu_cores,
		"memorySize":
		ram_cap,
		"storageSize":
		disk_cap,
		}
		flavours_supported.append(flavour)

		# Decide a single ISA for the aggregate reserved/quota entries
		# Preference order: X86_64, ARM_64, X86
		def pick_aggregate_isa() -> str:
		if "ISA_X86_64" in seen_cpu_isas:
		return "ISA_X86_64"
		if "ISA_ARM_64" in seen_cpu_isas:
		return "ISA_ARM_64"
		if "ISA_X86" in seen_cpu_isas:
		return "ISA_X86"
		# fallback
		return "ISA_X86_64"

		agg_isa = pick_aggregate_isa()

		result = {
		"zoneId":
		domain,
		"reservedComputeResources": [{
		"cpuArchType": agg_isa,
		"numCPU": int(
		total_cpu
		), # Same as Quotas untill we have somem policy or data to differentiate
		"memory": total_ram, # ditto
		}],
		"computeResourceQuotaLimits": [{
		"cpuArchType": agg_isa,
		"numCPU": int(total_cpu),
		"memory": total_ram,
		}],
		"flavoursSupported":
		flavours_supported,
		}

		return result

Admin message