aerOS cherry pick (!189) · Merge requests · OOP / code / transformation-function-sdk

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

0 → 100644

+0 −0

Empty file added.

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

0 → 100644

+152 −0

Changes for src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/aeros2gsma_zone_details.py: 152 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		aeros2gsma_zone_details.py
		"""

		from typing import Any, Dict, List


		def transformer(domain_ies: List[Dict[str, Any]], domain: str) -> Dict[str, Any]: # noqa: C901
		"""
		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

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

0 → 100644

+143 −0

Changes for src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/camara2aeros_converter.py: 143 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Module: converter.py
		This module provides functions to convert application manifests into TOSCA models.
		It includes the `generate_tosca` function that constructs a TOSCA model based on
		the application manifest and associated app zones.
		"""

		from typing import List

		import yaml

		from sunrise6g_opensdk.edgecloud.adapters.aeros import config
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		TOSCA,
		ArtifactModel,
		CustomRequirement,
		DomainIdOperator,
		ExposedPort,
		HostCapability,
		HostRequirement,
		NetworkProperties,
		NetworkRequirement,
		NodeFilter,
		NodeTemplate,
		PortProperties,
		)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		Property as HostProperty,
		)
		from sunrise6g_opensdk.edgecloud.core.camara_schemas import AppManifest, VisibilityType
		from sunrise6g_opensdk.logger import setup_logger

		logger = setup_logger(__name__, is_debug=True, file_name=config.LOG_FILE)


		def generate_tosca(app_manifest: AppManifest, app_zones: List[str]) -> str:
		"""
		Generate a TOSCA model from the application manifest and app zones.
		Args:
		app_manifest (AppManifest): The application manifest containing details about the app.
		app_zones (List[Dict[str, Any]]): List of app zones where the app will be deployed.
		Returns:
		TOSCA yaml as string which can be used in a POST request with applcation type yaml
		"""
		component = app_manifest.componentSpec[0]
		image_path = app_manifest.appRepo.imagePath.root
		image_file = image_path.split("/")[-1]
		repository_url = "/".join(image_path.split("/")[:-1]) if "/" in image_path else "docker_hub"

		zone_id = app_zones[0]
		logger.info("DEBUG : %s", app_manifest.requiredResources.root)
		# Extract minNodeMemory (fallback = 1024 MB)

		res = app_manifest.requiredResources.root
		if hasattr(res, "applicationResources") and hasattr(
		res.applicationResources.cpuPool.topology, "minNodeMemory"
		):
		min_node_memory = res.applicationResources.cpuPool.topology.minNodeMemory
		else:
		min_node_memory = 1024

		# Build exposed network ports
		ports = {
		iface.interfaceId: ExposedPort(
		properties=PortProperties(protocol=[iface.protocol.value.lower()], source=iface.port)
		)
		for iface in component.networkInterfaces
		}

		expose_ports = any(
		iface.visibilityType == VisibilityType.VISIBILITY_EXTERNAL
		for iface in component.networkInterfaces
		)

		# Define host property constraints
		host_props = HostProperty(
		cpu_arch={"equal": "x64"},
		realtime={"equal": False},
		cpu_usage={"less_or_equal": "0.4"},
		mem_size={"greater_or_equal": str(min_node_memory)},
		energy_efficiency={"greater_or_equal": "0"},
		green={"greater_or_equal": "0"},
		domain_id=DomainIdOperator(equal=zone_id),
		)

		# Create Node compute and network requirements
		requirements = [
		CustomRequirement(
		network=NetworkRequirement(
		properties=NetworkProperties(ports=ports, exposePorts=expose_ports)
		)
		),
		CustomRequirement(
		host=HostRequirement(
		node_filter=NodeFilter(
		capabilities=[{"host": HostCapability(properties=host_props)}], properties=None
		)
		)
		),
		]
		# Define the NodeTemplate
		node_template = NodeTemplate(
		type="tosca.nodes.Container.Application",
		isJob=False,
		requirements=requirements,
		artifacts={
		"application_image": ArtifactModel(
		file=image_file,
		type="tosca.artifacts.Deployment.Image.Container.Docker",
		repository=repository_url,
		is_private=app_manifest.appRepo.type == "PRIVATEREPO",
		username=app_manifest.appRepo.userName,
		password=app_manifest.appRepo.credentials,
		)
		},
		interfaces={
		"Standard": {
		"create": {
		"implementation": "application_image",
		"inputs": {"cliArgs": [], "envVars": []},
		}
		}
		},
		)

		# Assemble full TOSCA object
		tosca = TOSCA(
		tosca_definitions_version="tosca_simple_yaml_1_3",
		description=f"TOSCA for {app_manifest.name}",
		serviceOverlay=False,
		node_templates={component.componentName: node_template},
		)

		tosca_dict = tosca.model_dump(by_alias=True, exclude_none=True)

		for template in tosca_dict.get("node_templates", {}).values():
		template["requirements"] = [
		{k: v for k, v in req.items() if v is not None}
		for req in template.get("requirements", [])
		]

		yaml_str = yaml.dump(tosca_dict, sort_keys=False)
		return yaml_str

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

0 → 100644

+226 −0

Changes for src/sunrise6g_opensdk/edgecloud/adapters/aeros/converters/gsma2aeros_converter.py: 226 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Module: gsm2aeros_converter.py
		Initial GSMA -> TOSCA generator.

		Notes:
		- GSMA ApplicationModel does not include container image or ports directly.
		(Those usually come from Artefacts, which we're ignoring for now.)
		- We provide an `image_map` hook to resolve artefactId -> image string.
		- Defaults to a public nginx image if nothing is provided.
		- Network ports are omitted for now (exposePorts = False).
		"""

		from typing import Callable, Dict, List, Optional

		import yaml

		from sunrise6g_opensdk.edgecloud.adapters.aeros import config
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		TOSCA,
		ArtifactModel,
		CustomRequirement,
		DomainIdOperator,
		ExposedPort,
		HostCapability,
		HostRequirement,
		NetworkProperties,
		NetworkRequirement,
		NodeFilter,
		NodeTemplate,
		PortProperties,
		)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_models import (
		Property as HostProperty,
		)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.errors import (
		InvalidArgumentError,
		ResourceNotFoundError,
		)
		from sunrise6g_opensdk.edgecloud.core import gsma_schemas
		from sunrise6g_opensdk.logger import setup_logger

		logger = setup_logger(__name__, is_debug=True, file_name=config.LOG_FILE)


		def generate_tosca_from_gsma_with_artefacts( # noqa: C901
		app_model: gsma_schemas.ApplicationModel,
		zone_id: str,
		artefact_resolver: Callable[[str], Optional[gsma_schemas.Artefact]],
		) -> str:
		"""
		Build a TOSCA YAML from a GSMA `ApplicationModel` by resolving each component's `artefactId`.

		Rules/assumptions:
		- One node_template per `AppComponentSpec` in the application model.
		- Container image is taken from the first entry of `artefact.componentSpec[i].images`.
		- Ports come (best-effort) from `exposedInterfaces` items in the matching componentSpec, e.g. {"protocol": "TCP", "port": 8080}.
		- Host filter includes domain_id == `zone_id` and basic CPU/mem constraints.
		- For PUBLICREPO artefacts: set `is_private=False` and omit credentials entirely.
		For PRIVATEREPO artefacts: set `is_private=True` and include non-empty username/password if present.
		- `cliArgs` are derived from `commandLineParams` dict:
		- bool True -> "flag"
		- key/value -> "key=value"
		`envVars` are derived from `compEnvParams` list:
		- [{"name": "KEY", "value": "VAL"}] -> [{"KEY": "VAL"}, ...]
		- If a component name mismatch occurs between app and artefact, fall back to the first artefact componentSpec.

		:param app_model: GSMA ApplicationModel (already validated)
		:param zone_id: Target aerOS domain id/zone urn for host node filter
		:param artefact_resolver: Callable that returns an Artefact for a given artefactId
		:return: TOSCA YAML string (tosca_simple_yaml_1_3)
		"""
		node_templates: Dict[str, NodeTemplate] = {}

		for comp in app_model.appComponentSpecs:
		artefact = artefact_resolver(comp.artefactId)
		if not artefact:
		raise ResourceNotFoundError(f"GSMA artefact '{comp.artefactId}' not found")

		# pick the componentSpec that matches componentName, else first
		comp_spec = None
		if artefact.componentSpec:
		for c in artefact.componentSpec:
		if c.componentName == comp.componentName:
		comp_spec = c
		break
		if comp_spec is None:
		comp_spec = artefact.componentSpec[0]
		else:
		raise InvalidArgumentError(f"Artefact '{artefact.artefactId}' has no componentSpec")

		# Resolve container image
		image = comp_spec.images[0] if comp_spec.images else "docker.io/library/nginx:stable"
		if "/" in image:
		repository_url = "/".join(image.split("/")[:-1])
		image_file = image.split("/")[-1]
		else:
		repository_url, image_file = "docker_hub", image

		# Ports (best-effort) from exposedInterfaces
		ports: Dict[str, ExposedPort] = {}
		expose_ports = False
		if comp_spec.exposedInterfaces:
		for idx, iface in enumerate(comp_spec.exposedInterfaces):
		protocol = str(iface.get("protocol", "TCP")).lower()
		port = iface.get("port")
		if isinstance(port, int):
		ports[f"if{idx}"] = ExposedPort(
		properties=PortProperties(protocol=[protocol], source=port)
		)
		expose_ports = True

		# Build cliArgs as a list of dicts: [{"KEY": "VAL"}, {"FLAG": ""}, ...]
		cli_args: List[Dict[str, str]] = []
		cmd = getattr(comp_spec, "commandLineParams", None)

		if isinstance(cmd, dict):
		for k, v in cmd.items():
		if v is True:
		cli_args.append({str(k): ""}) # flag without value
		elif v is False or v is None:
		continue
		else:
		cli_args.append({str(k): str(v)})
		elif isinstance(cmd, list):
		# if someone passes ["--flag", "--opt=1"] style
		for item in cmd:
		if isinstance(item, str):
		if "=" in item:
		k, v = item.split("=", 1)
		cli_args.append({k: v})
		else:
		cli_args.append({item: ""})

		# Build envVars from compEnvParams list of {"name": "...", "value": "..."}
		env_vars: List[Dict[str, str]] = []
		if isinstance(getattr(comp_spec, "compEnvParams", None), list):
		for item in comp_spec.compEnvParams:
		if isinstance(item, dict):
		if "name" in item and "value" in item:
		env_vars.append({str(item["name"]): str(item["value"])})
		elif len(item) == 1: # already mapping-like {"KEY": "VAL"}
		k, v = next(iter(item.items()))
		env_vars.append({str(k): str(v)})

		# Host filter (basic example)
		host_props = HostProperty(
		cpu_arch={"equal": "x64"},
		realtime={"equal": False},
		cpu_usage={"less_or_equal": "0.4"},
		mem_size={"greater_or_equal": "1024"},
		energy_efficiency={"greater_or_equal": "0"},
		green={"greater_or_equal": "0"},
		domain_id=DomainIdOperator(equal=zone_id),
		)

		requirements = [
		CustomRequirement(
		network=NetworkRequirement(
		properties=NetworkProperties(ports=ports, exposePorts=expose_ports)
		)
		),
		CustomRequirement(
		host=HostRequirement(
		node_filter=NodeFilter(
		capabilities=[{"host": HostCapability(properties=host_props)}],
		properties=None,
		)
		)
		),
		]

		# PUBLICREPO => is_private=False and omit credentials
		repo_type = getattr(artefact, "repoType", None)
		is_private = bool(repo_type == "PRIVATEREPO")
		username = None
		password = None
		if is_private and artefact.artefactRepoLocation:
		u = artefact.artefactRepoLocation.userName
		p = artefact.artefactRepoLocation.password
		username = u if u else None
		password = p if p else None

		node_templates[comp.componentName] = NodeTemplate(
		type="tosca.nodes.Container.Application",
		isJob=False,
		requirements=requirements,
		artifacts={
		"application_image": ArtifactModel(
		file=image_file,
		type="tosca.artifacts.Deployment.Image.Container.Docker",
		repository=repository_url,
		is_private=is_private, # False for PUBLICREPO
		username=username, # None for PUBLICREPO
		password=password, # None for PUBLICREPO
		)
		},
		interfaces={
		"Standard": {
		"create": {
		"implementation": "application_image",
		"inputs": {
		"cliArgs": cli_args,
		"envVars": env_vars,
		},
		}
		}
		},
		)

		# Assemble and dump TOSCA
		tosca = TOSCA(
		tosca_definitions_version="tosca_simple_yaml_1_3",
		description=f"GSMA->TOSCA for {app_model.appMetaData.appName} ({app_model.appId})",
		serviceOverlay=False,
		node_templates=node_templates,
		)

		tosca_dict = tosca.model_dump(by_alias=True, exclude_none=True)
		# Clean requirements lists from None entries
		for template in tosca_dict.get("node_templates", {}).values():
		template["requirements"] = [
		{k: v for k, v in req.items() if v is not None}
		for req in template.get("requirements", [])
		]

		return yaml.dump(tosca_dict, sort_keys=False)

src/sunrise6g_opensdk/edgecloud/adapters/aeros/storageManagement/init.py

0 → 100644

+6 −0

Changes for src/sunrise6g_opensdk/edgecloud/adapters/aeros/storageManagement/__init__.py: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		This module contains the storage management implementations for aerOS.
		"""

		from .inMemoryStorage import InMemoryAppStorage
		from .sqlite_storage import SQLiteAppStorage

aerOS cherry pick

Merge request reports