GSMA Validated (dde919cd) · Commits · OOP / code / transformation-function-sdk

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

+81 −45

Original line number	Diff line number	Diff line
		@@ -14,7 +14,7 @@ from requests import Response

		from sunrise6g_opensdk.edgecloud.adapters.aeros import config
		from sunrise6g_opensdk.edgecloud.adapters.aeros.utils import (
		urn_to_uuid, encode_app_instance_name)
		urn_to_uuid, encode_app_instance_name, map_aeros_service_status_to_gsma)
		from sunrise6g_opensdk.edgecloud.adapters.aeros.continuum_client import ContinuumClient
		from sunrise6g_opensdk.edgecloud.adapters.aeros.storageManagement import inMemoryStorage
		from sunrise6g_opensdk.edgecloud.adapters.aeros.converters import (
		@@ -967,6 +967,25 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		if not self.storage.get_app_gsma(app_id):
		raise ResourceNotFoundError(f"GSMA app '{app_id}' not found")

		# CHECKME: update for GSMA
		service_instances = self.storage.get_stopped_instances_gsma(
		app_id=app_id)
		if not service_instances:
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' cannot be deleted — please stop it first"
		)
		self.logger.debug(
		"Deleting application with id: %s and instances: %s",
		app_id,
		service_instances,
		)
		for service_instance in service_instances:
		self._purge_deployed_app_from_continuum_gsma(service_instance)
		self.logger.debug("successfully purged service instance: %s",
		service_instance)

		self.storage.remove_stopped_instances_gsma(app_id)

		self.storage.delete_app_gsma(app_id)

		return build_custom_http_response(
		@@ -983,6 +1002,23 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		app_id, e)
		raise EdgeCloudPlatformError(str(e))

		def _purge_deployed_app_from_continuum_gsma(self,
		app_instance_id: str) -> None:
		'''
		Purge the deployed application from aerOS continuum.
		:param app_id: The application ID to purge
		All instances of this app should be stopped
		'''
		aeros_client = ContinuumClient(self.base_url)
		response = aeros_client.purge_service(app_instance_id)
		if response:
		self.logger.debug("Purged deployed application with id: %s",
		app_instance_id)
		else:
		raise EdgeCloudPlatformError(
		f"Failed to purge service with id from the continuum '{app_instance_id}'"
		)

		# ------------------------------------------------------------------------
		# Application Deployment Management (GSMA)
		# ------------------------------------------------------------------------
		@@ -1009,8 +1045,9 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		f"GSMA app '{payload.appId}' not found")

		# 2. Generate unique service ID
		# (aerOS) service id <=> CAMARA appInstanceId
		service_id = self._generate_service_id(onboarded_app.appId)
		# (aerOS) service id <=> GSMA appInstanceId
		service_id = self._generate_aeros_service_id(
		self._generate_service_id(onboarded_app.appId))

		# 3. Create TOSCA (yaml str) from GSMA onboarded_app + connected artefacts
		# GSMA app corresponds to aerOS Service
		@@ -1028,7 +1065,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		aeros_response = aeros_client.onboard_and_deploy_service(
		service_id, tosca_str=tosca_yaml)

		if "serviceId" not in aeros_response:
		if "serviceId" not in aeros_response.json():
		raise EdgeCloudPlatformError(
		"Invalid response from onboard_service: missing 'serviceId'"
		)
		@@ -1040,8 +1077,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		appInstIdentifier=service_id,
		)
		status = gsma_schemas.AppInstanceStatus(
		appInstanceState=
		"DEPLOYED", # or "PENDING" if you simulate async
		appInstanceState="PENDING",
		accesspointInfo=[],
		)

		@@ -1050,21 +1086,15 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		status=status)

		# 6. Return expected format (deployment details)
		body = {
		"appId": payload.appId,
		"appVersion": payload.appVersion,
		"appProviderId": payload.appProviderId,
		"zoneId": payload.zoneInfo.zoneId,
		"appInstance": inst.model_dump(mode="json"),
		"status": status.model_dump(mode="json"),
		}
		body = inst.model_dump(mode="json")

		return build_custom_http_response(
		status_code=201,
		status_code=202,
		content=body,
		headers={"Content-Type": self.content_type_gsma},
		encoding=self.encoding_gsma,
		)
		url=aeros_response.json().get("url", ""),
		request=aeros_response.request)
		except EdgeCloudPlatformError as ex:
		self.logger.error("Failed to deploy app '%s': %s",
		onboarded_app.appId, str(ex))
		@@ -1088,24 +1118,34 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		if not self.storage.get_app_gsma(app_id):
		raise ResourceNotFoundError(f"GSMA app '{app_id}' not found")

		matches = self.storage.find_deployments_gsma(
		app_id=app_id,
		app_instance_id=app_instance_id,
		zone_id=zone_id,
		)
		if not matches:
		raise ResourceNotFoundError(
		f"Deployment not found (app_id={app_id}, instance={app_instance_id}, zone={zone_id})"
		# 4. Instantiate client and call continuum to deploy servic
		aeros_client = ContinuumClient(self.base_url)
		aeros_response = aeros_client.query_entity(
		entity_id=app_instance_id, ngsild_params='format=simplified')

		response_json = aeros_response.json()
		content = gsma_schemas.AppInstanceStatus(
		appInstanceState=map_aeros_service_status_to_gsma(
		response_json.get("actionType")),
		accesspointInfo=[{
		"service_status":
		f'{self.base_url}/entities/{app_instance_id}'
		}, {
		"serviceComponents_status":
		f'{self.base_url}/hlo_fe/services//{app_instance_id}'
		}],
		)

		inst = matches[0]
		body = inst.model_dump(mode="json")
		validated_data = gsma_schemas.AppInstanceStatus.model_validate(
		content)

		return build_custom_http_response(
		status_code=200,
		content=body,
		content=validated_data.model_dump(mode="json"),
		headers={"Content-Type": self.content_type_gsma},
		encoding=self.encoding_gsma,
		url=aeros_response.url,
		request=aeros_response.request,
		)
		except EdgeCloudPlatformError:
		raise
		@@ -1115,8 +1155,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		app_instance_id, app_id, zone_id, e)
		raise EdgeCloudPlatformError(str(e))

		def get_all_deployed_apps_gsma(self, app_id: str,
		app_provider: str) -> List:
		def get_all_deployed_apps_gsma(self) -> Response:
		"""
		Retrieves all instances for a given application of partner OP

		@@ -1125,18 +1164,10 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:return: List with application instances details
		"""
		try:
		app = self.storage.get_app_gsma(app_id)
		if not app:
		raise ResourceNotFoundError(f"GSMA app '{app_id}' not found")

		# Optional provider check (keep if you want extra validation)
		if app_provider and app.appProviderId != app_provider:
		raise ResourceNotFoundError(
		f"GSMA app '{app_id}' not found for provider '{app_provider}'"
		)

		insts = self.storage.find_deployments_gsma(app_id=app_id)
		insts = self.storage.find_deployments_gsma()
		body = [i.model_dump(mode="json") for i in insts]
		self.logger.info("All GSMA app instances retrieved successfully")
		self.logger.debug("Deployed GSMA applications: %s", body)

		return build_custom_http_response(
		status_code=200,
		@@ -1148,8 +1179,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		raise
		except Exception as e:
		self.logger.exception(
		"Unhandled error listing GSMA deployments for app '%s': %s",
		app_id, e)
		"Unhandled error listing GSMA deployments: '%s'", e)
		raise EdgeCloudPlatformError(str(e))

		def undeploy_app_gsma(self, app_id: str, app_instance_id: str,
		@@ -1177,8 +1207,14 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		f"Deployment not found (app_id={app_id}, instance={app_instance_id}, zone={zone_id})"
		)

		# Placeholder: call aerOS undeploy here (GSMA → aerOS conversion)
		# aeros_client.undeploy(instance_id=app_instance_id, zone_id=zone_id)
		# 2. Call the external undeploy_service
		aeros_client = ContinuumClient(self.base_url)
		try:
		aeros_response = aeros_client.undeploy_service(app_instance_id)
		except Exception as e:
		raise EdgeCloudPlatformError(
		f"Failed to undeploy app instance '{app_instance_id}': {str(e)}"
		) from e

		# Remove from deployed and mark as stopped so it can be purged later
		removed_app_id = self.storage.remove_deployment_gsma(
		@@ -1195,7 +1231,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"state": "TERMINATING",
		}
		return build_custom_http_response(
		status_code=202,
		status_code=aeros_response.status_code,
		content=body,
		headers={"Content-Type": self.content_type_gsma},
		encoding=self.encoding_gsma,

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

+2 −2

Original line number	Diff line number	Diff line
		@@ -51,7 +51,7 @@ class ContinuumClient:
		}

		@catch_requests_exceptions
		def query_entity(self, entity_id, ngsild_params) -> dict:
		def query_entity(self, entity_id, ngsild_params) -> requests.Response:
		"""
		Query entity with ngsi-ld params
		:input
		@@ -70,7 +70,7 @@ class ContinuumClient:
		self.logger.debug(
		"Query entity response: %s %s", response.status_code, response.text
		)
		return response.json()
		return response

		@catch_requests_exceptions
		def query_entities(self, ngsild_params) -> requests.Response:

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

+99 −31

Original line number	Diff line number	Diff line
		@@ -10,17 +10,30 @@ Notes:
		- Network ports are omitted for now (exposePorts = False).
		"""

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

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

		@@ -31,14 +44,28 @@ def generate_tosca_from_gsma_with_artefacts(
		artefact_resolver: Callable[[str], Optional[gsma_schemas.Artefact]],
		) -> str:
		"""
		Build TOSCA from GSMA ApplicationModel by resolving each component's artefactId.

		- One Node (under NodeTemplates) per AppComponentSpec (connects to Artefacts).
		- Image pulled from Artefact.componentSpec[i].images[0] (first image).
		- Optional ports read from Artefact.componentSpec[i].exposedInterfaces (if present).
		Expected dict fields (best-effort): {"protocol": "TCP", "port": 8080}
		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 = {}
		node_templates: Dict[str, NodeTemplate] = {}

		for comp in app_model.appComponentSpecs:
		artefact = artefact_resolver(comp.artefactId)
		@@ -46,22 +73,20 @@ def generate_tosca_from_gsma_with_artefacts(
		raise ResourceNotFoundError(
		f"GSMA artefact '{comp.artefactId}' not found")

		# We pick the FIRST componentSpec entry that matches componentName if present,
		# else fall back to the first componentSpec entry.
		# pick the componentSpec that matches componentName, else first
		comp_spec = None
		if artefact.componentSpec:
		# try exact match by name
		for c in artefact.componentSpec:
		if c.componentName == comp.componentName:
		comp_spec = c
		break
		if not comp_spec:
		if comp_spec is None:
		comp_spec = artefact.componentSpec[0]
		else:
		raise InvalidArgumentError(
		f"Artefact '{artefact.artefactId}' has no componentSpec")

		# Resolve image (first image in the list)
		# Resolve container image
		image = comp_spec.images[
		0] if comp_spec.images else "docker.io/library/nginx:stable"
		if "/" in image:
		@@ -70,19 +95,53 @@ def generate_tosca_from_gsma_with_artefacts(
		else:
		repository_url, image_file = "docker_hub", image

		# Build ports (best-effort read from exposedInterfaces)
		ports = {}
		# 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_id = f"if{idx}"
		ports[ports_id] = ExposedPort(properties=PortProperties(
		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},
		@@ -105,6 +164,17 @@ def generate_tosca_from_gsma_with_artefacts(
		))),
		]

		# 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,
		@@ -115,11 +185,9 @@ def generate_tosca_from_gsma_with_artefacts(
		file=image_file,
		type="tosca.artifacts.Deployment.Image.Container.Docker",
		repository=repository_url,
		is_private=(artefact.repoType == "PRIVATEREPO"),
		username=(artefact.artefactRepoLocation.userName
		if artefact.artefactRepoLocation else None),
		password=(artefact.artefactRepoLocation.password
		if artefact.artefactRepoLocation else None),
		is_private=is_private, # False for PUBLICREPO
		username=username, # None for PUBLICREPO
		password=password, # None for PUBLICREPO
		)
		},
		interfaces={
		@@ -127,16 +195,15 @@ def generate_tosca_from_gsma_with_artefacts(
		"create": {
		"implementation": "application_image",
		"inputs": {
		"cliArgs":
		[], # could map comp_spec.commandLineParams later
		"envVars":
		[], # could map comp_spec.compEnvParams later
		}
		"cliArgs": cli_args,
		"envVars": env_vars,
		},
		}
		}
		},
		)

		# Assemble and dump TOSCA
		tosca = TOSCA(
		tosca_definitions_version="tosca_simple_yaml_1_3",
		description=
		@@ -146,6 +213,7 @@ def generate_tosca_from_gsma_with_artefacts(
		)

		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

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

+24 −0

Original line number	Diff line number	Diff line
		@@ -91,6 +91,30 @@ def urn_to_uuid(urn: str) -> uuid.UUID:
		return uuid.uuid5(uuid.NAMESPACE_URL, urn)


		def map_aeros_service_status_to_gsma(status: str) -> str:
		"""
		Map aerOS service lifecycle states to GSMA-compliant status values.

		aerOS → GSMA
		DEPLOYING → PENDING
		DESTROYING → TERMINATING
		DEPLOYED → DEPLOYED
		FINISHED → No_Match
		No_Match → READY
		urn:ngsi-ld:null → No Match
		"""
		mapping = {
		"DEPLOYING": "PENDING",
		"DESTROYING": "TERMINATING",
		"DEPLOYED": "DEPLOYED",
		"FINISHED": "READY",
		# "urn:ngsi-ld:null": "READY",
		}
		if not status:
		return "FAILED"
		return mapping.get(status.strip().upper(), "FAILED")


		def catch_requests_exceptions(func):
		"""
		Decorator to catch and translate requests exceptions into custom app errors.

Admin message