updated TFs for CAMARA payloads (aec4787f) · Commits · OOP / code / transformation-function-sdk

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

+88 −84

Original line number	Diff line number	Diff line
		@@ -30,6 +30,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		self.logger = setup_logger(__name__, is_debug=True, file_name=config.LOG_FILE)
		self._app_store: Dict[str, Dict] = {}
		self._deployed_services: Dict[str, List[str]] = {}
		self._stopped_services: Dict[str, List[str]] = {}

		# Overwrite config values if provided via kwargs
		if "aerOS_API_URL" in kwargs:
		@@ -47,13 +48,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		raise ValueError("Missing 'aerOS_HLO_TOKEN'")

		def onboard_app(self, app_manifest: Dict) -> Dict:
		# HLO-FE POST with TOSCA and app_id (service_id)
		# service_id = app_manifest.get("serviceId")
		# tosca_str = app_manifest.get("tosca")
		# aeros_client = ContinuumClient(self.base_url)
		# onboard_response = aeros_client.onboard_service(
		# service_id=service_id, tosca_str=tosca_str
		# )
		app_id = app_manifest.get("appId")
		if not app_id:
		raise EdgeCloudPlatformError("Missing 'appId' in app manifest")
		@@ -68,23 +62,15 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return {"appId": app_id}

		def get_all_onboarded_apps(self) -> List[Dict]:
		# aeros_client = ContinuumClient(self.base_url)
		# ngsild_params = "type=Service&format=simplified"
		# aeros_apps = aeros_client.query_entities(ngsild_params)
		# return [
		# {"appId": service["id"], "name": service["name"]} for service in aeros_apps
		# ]
		self.logger.debug("Onboarded applications: %s", list(self._app_store.keys()))
		return list(self._app_store.values())

		def get_onboarded_app(self, app_id: str) -> Dict:
		# aeros_client = ContinuumClient(self.base_url)
		# ngsild_params = "format=simplified"
		# aeros_app = aeros_client.query_entity(app_id, ngsild_params)
		# return {"appId": aeros_app["id"], "name": aeros_app["name"]}
		if app_id not in self._app_store:
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' does not exist"
		)
		self.logger.debug("Retrieved application with id: %s", app_id)
		return self._app_store[app_id]

		def delete_onboarded_app(self, app_id: str) -> None:
		@@ -92,9 +78,19 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		raise EdgeCloudPlatformError(
		f"Application with id '{app_id}' does not exist"
		)
		del self._app_store[app_id]
		# TBD: Purge from continuum (make all ngsil-ld calls for servieId connected entities)
		# Should check if undeployed first
		service_instances = self._stopped_services.get(app_id, [])
		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(service_instance)
		self.logger.debug(
		"successfully purged service instance: %s", service_instance
		)
		del self._stopped_services[app_id] # Clean up stopped services
		del self._app_store[app_id] # Remove from onboarded apps

		def _generate_service_id(self, app_id: str) -> str:
		return f"urn:ngsi-ld:Service:{app_id}-{uuid.uuid4().hex[:4]}"
		@@ -106,16 +102,23 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		component_name = component.get("componentName", "application")

		image_path = app_manifest.get("appRepo", {}).get("imagePath", "")
		# Extract image_file
		image_file = image_path.split("/")[-1]
		# Extract repository_url
		if "/" in image_path:
		repository_url = "/".join(image_path.split("/")[:-1])
		else:
		repository_url = "docker_hub"
		repository_url = (
		"/".join(image_path.split("/")[:-1]) if "/" in image_path else "docker_hub"
		)
		zone_id = (
		app_zones[0].get("EdgeCloudZone", {}).get("edgeCloudZoneId", "default-zone")
		)

		# Extract minNodeMemory
		min_node_memory = (
		app_manifest.get("requiredResources", {})
		.get("applicationResources", {})
		.get("cpuPool", {})
		.get("topology", {})
		.get("minNodeMemory", 1024)
		)

		ports = {}
		for iface in component.get("networkInterfaces", []):
		interface_id = iface.get("interfaceId", "default")
		@@ -124,6 +127,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		ports[interface_id] = {
		"properties": {"protocol": [protocol], "source": port}
		}

		expose_ports = any(
		iface.get("visibilityType") == "VISIBILITY_EXTERNAL"
		for iface in component.get("networkInterfaces", [])
		@@ -132,9 +136,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		yaml_dict = {
		"tosca_definitions_version": "tosca_simple_yaml_1_3",
		"description": f"TOSCA for {app_manifest.get('name', 'application')}",
		"serviceOverlay": False,
		"node_templates": {
		component_name: {
		"type": "tosca.nodes.Container.Application",
		"isJob": False,
		"requirements": [
		{
		"network": {
		@@ -144,12 +150,38 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		}
		}
		},
		{"host": {"node_filter": {"properties": {"id": zone_id}}}},
		{
		"host": {
		"node_filter": {
		"capabilities": [
		{
		"host": {
		"properties": {
		"cpu_arch": {"equal": "x64"},
		"realtime": {"equal": False},
		"cpu_usage": {
		"less_or_equal": "0.1"
		},
		"mem_size": {
		"greater_or_equal": str(
		min_node_memory
		)
		},
		"domain_id": {"equal": zone_id},
		}
		}
		}
		],
		"properties": None,
		}
		}
		},
		],
		"artifacts": {
		"application_image": {
		"file": image_file,
		"type": "tosca.artifacts.Deployment.Image.Container.Docker",
		"is_private": False,
		"repository": repository_url,
		}
		},
		@@ -168,10 +200,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		return yaml_dict

		def deploy_app(self, app_id: str, app_zones: List[Dict]) -> Dict:
		# HLO-FE PUT with app_id (service_id)
		# aeros_client = ContinuumClient(self.base_url)
		# deploy_response = aeros_client.deploy_service(app_id)
		# return {"deploy_name": deploy_response["serviceId"]}
		# 1. Get app CAMARA manifest
		app_manifest = self._app_store.get(app_id)
		if not app_manifest:
		@@ -181,35 +209,28 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		# 2. Generate unique service ID
		service_id = self._generate_service_id(app_id)
		# Dev test
		# service_id = "service-my-id"

		# yaml_dict = {
		# "serviceId": service_id,
		# "tosca": app_manifest.get("tosca", "")
		# }

		# 5. Convert dict to YAML string
		# 3. Convert dict to YAML string
		yaml_dict = self._generate_tosca_yaml_dict(app_manifest, app_zones)
		tosca_yaml = yaml.dump(yaml_dict, sort_keys=False)
		print("Generated TOSCA YAML:")
		print(tosca_yaml)
		response = {"serviceId": service_id} # Mocked response
		# 6. Instantiate client and call onboard_service
		self.logger.info("Generated TOSCA YAML:")
		self.logger.info(tosca_yaml)

		# 4. Instantiate client and call continuum to deploy service
		aeros_client = ContinuumClient(self.base_url)
		response = aeros_client.onboard_service(service_id, tosca_yaml)
		response = aeros_client.onboard_and_deploy_service(service_id, tosca_yaml)

		if "serviceId" not in response:
		raise EdgeCloudPlatformError(
		"Invalid response from onboard_service: missing 'serviceId'"
		)

		# 7. Track deployment
		# 5. Track deployment
		if app_id not in self._deployed_services:
		self._deployed_services[app_id] = []
		self._deployed_services[app_id].append(service_id)

		# 8. Return expected format
		# 6. Return expected format
		return {"appInstanceId": response["serviceId"]}

		def get_all_deployed_apps(
		@@ -218,45 +239,21 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		app_instance_id: Optional[str] = None,
		region: Optional[str] = None,
		) -> List[Dict]:
		# # FIXME: Get services in deployed state
		# aeros_client = ContinuumClient(self.base_url)
		# ngsild_params = 'type=Service&format=simplified&q=actionType=="DEPLOYED"'
		# if app_id:
		# ngsild_params += f'&q=service=="{app_id}"'
		# aeros_apps = aeros_client.query_entities(ngsild_params)
		# return [
		# {
		# "appInstanceId": service["id"],
		# "status":
		# # scomponent["serviceComponentStatus"].split(":")[-1].lower()
		# service["actionType"],
		# }
		# for service in aeros_apps
		# ]
		# # return [{"appInstanceId": "abcd-efgh", "status": "ready"}]
		deployed = []
		for stored_app_id, instance_ids in self._deployed_services.items():
		for instance_id in instance_ids:
		deployed.append({"appId": stored_app_id, "appInstanceId": instance_id})
		return deployed

		# def get_all_deployed_apps(self,
		# app_id: Optional[str] = None,
		# app_instance_id: Optional[str] = None,
		# region: Optional[str] = None) -> List[Dict]:
		# # FIXME: Get services in deployed state
		# aeros_client = ContinuumClient(self.base_url)
		# ngsild_params = "type=ServiceComponent&format=simplified"
		# if app_id:
		# ngsild_params += f'&q=service=="{app_id}"'
		# aeros_apps = aeros_client.query_entities(ngsild_params)
		# return [{
		# "appInstanceId":
		# scomponent["id"],
		# "status":
		# scomponent["serviceComponentStatus"].split(":")[-1].lower()
		# } for scomponent in aeros_apps]
		# # return [{"appInstanceId": "abcd-efgh", "status": "ready"}]
		def _purge_deployed_app_from_continuum(self, app_id: str) -> None:
		aeros_client = ContinuumClient(self.base_url)
		response = aeros_client.purge_service(app_id)
		if response:
		self.logger.debug("Purged deployed application with id: %s", app_id)
		else:
		raise EdgeCloudPlatformError(
		f"Failed to purg service with id from the continuum '{app_id}'"
		)

		def undeploy_app(self, app_instance_id: str) -> None:
		# 1. Locate app_id corresponding to this instance
		@@ -278,10 +275,19 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		except Exception as e:
		raise EdgeCloudPlatformError(
		f"Failed to undeploy app instance '{app_instance_id}': {str(e)}"
		)
		) from e

		# 3. Clean up internal tracking
		# We could do it here with a little wait but better all instances in the same app are purged at once
		# 3. Purge the deployed app from continuum
		# self._purge_deployed_app_from_continuum(app_instance_id)

		# 4. Clean up internal tracking
		self._deployed_services[found_app_id].remove(app_instance_id)
		# Add instance to _stopped_services to purge it later
		if found_app_id not in self._stopped_services:
		self._stopped_services[found_app_id] = []
		self._stopped_services[found_app_id].append(app_instance_id)
		# If app has no instances left, remove it from deployed services
		if not self._deployed_services[found_app_id]:
		del self._deployed_services[found_app_id]

		@@ -300,8 +306,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		for domain in aeros_domains
		]

		# return [{"edgeCloudZoneId": "zone-1", "status": "active"}]

		def get_edge_cloud_zones_details(
		self, zone_id: str, flavour_id: Optional[str] = None
		) -> Dict:

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

+27 −1

Original line number	Diff line number	Diff line
		@@ -139,7 +139,7 @@ class ContinuumClient:
		return response.json()

		@catch_requests_exceptions
		def onboard_service(self, service_id: str, tosca_str: str) -> dict:
		def onboard_and_deploy_service(self, service_id: str, tosca_str: str) -> dict:
		"""
		Onboard (& deploy) service on aerOS continuum
		:input
		@@ -168,3 +168,29 @@ class ContinuumClient:
		response.text,
		)
		return response.json()

		@catch_requests_exceptions
		def purge_service(self, service_id: str) -> bool:
		"""
		Purge service from aerOS continuum
		:input
		@param service_id: the id of the service to be purged
		:output
		the purge result message from aerOS continuum
		"""
		purge_url = f"{self.api_url}/hlo_fe/services/{service_id}/purge"
		response = requests.delete(purge_url, headers=self.hlo_headers, timeout=15)
		if response is None:
		return False
		else:
		if config.DEBUG:
		self.logger.debug("Purge service URL: %s", purge_url)
		self.logger.debug(
		"Purge service response: %s %s",
		response.status_code,
		response.text,
		)
		if response.status_code != 200:
		self.logger.error("Failed to purge service: %s", response.text)
		return False
		return True

tests/edgecloud/test_cases.py

+11 −11

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		test_cases = [
		{
		"edgecloud": {
		"client_name": "i2edge",
		"base_url": "http://192.168.123.48:30769/",
		}
		},
		# {
		# "edgecloud": {
		# "client_name": "aeros",
		# "base_url": "http://test-aeros.url",
		# "aerOS_API_URL": "http://fake.api.url",
		# "aerOS_ACCESS_TOKEN": "fake-access",
		# "aerOS_HLO_TOKEN": "fake-hlo"
		# "client_name": "i2edge",
		# "base_url": "http://192.168.123.48:30769/",
		# }
		# },
		{
		"edgecloud": {
		"client_name": "aeros",
		"base_url": "https://ncsrd-mvp-domain.aeros-project.eu",
		"aerOS_API_URL": "https://ncsrd-mvp-domain.aeros-project.eu",
		"aerOS_ACCESS_TOKEN": "",
		"aerOS_HLO_TOKEN": "",
		}
		},
		# {
		# "edgecloud": {
		# "client_name": "piedge",

tests/edgecloud/test_config.py

+3 −96

Original line number	Diff line number	Diff line
		@@ -23,7 +23,7 @@ the EdgeCloud Platform integration across different clients.
		ZONE_ID = "urn:ngsi-ld:Domain:NCSRD"

		# Artefact
		ARTEFACT_ID = "aeros-id-1"
		ARTEFACT_ID = "aeros-app-1"
		ARTEFACT_NAME = "i2edgechart"
		REPO_NAME = "github-cesar"
		REPO_TYPE = "PUBLICREPO"
		@@ -32,13 +32,13 @@ REPO_URL = "https://cesarcajas.github.io/helm-charts-examples/"
		# Onboarding: CAMARA /app payload (only mandatory fields)
		APP_ONBOARD_MANIFEST = {
		"appId": ARTEFACT_ID,
		"name": "aeros-SDK",
		"name": "aeros-SDK-app",
		"version": "1.0.0",
		"appProvider": "ncsrd",
		"packageType": "CONTAINER",
		"appRepo": {
		"type": "PUBLICREPO",
		"imagePath": "https://example.com/nginx:latest",
		"imagePath": "docker.io/library/nginx:latest",
		},
		"requiredResources": {
		"infraKind": "kubernetes",
		@@ -94,97 +94,4 @@ APP_ZONES = [
		######################
		# aerOS variables
		######################
		AEROS_APP_ID = "urn:ngsi-ld:Service:sunriseapp2"
		AEROS_TOSCA_DESCRIPTOR = {
		"serviceId": AEROS_APP_ID,
		"tosca": """
		tosca_definitions_version: tosca_simple_yaml_1_3
		description: TOSCA for network performance
		node_templates:
		influxdb:
		type: tosca.nodes.Container.Application
		requirements:
		- network:
		properties:
		ports:
		fastapi:
		properties:
		protocol: [tcp]
		source: 8086
		exposePorts: true
		- host:
		node_filter:
		properties:
		id: "urn:ngsi-ld:InfrastructureElement:NCSRD:cebf2bd4d0ba"
		artifacts:
		influxdb-image:
		file: p4lik4ri/influxdb
		type: tosca.artifacts.Deployment.Image.Container.Docker
		repository: docker_hub
		interfaces:
		Standard:
		create:
		implementation: influxdb-image
		inputs:
		envVars:
		- INFLUXDB_BUCKET: some-bucket
		- INFLUXDB_ORG: NCSRD
		- INFLUXDB_USER: vpitsilis
		- INFLUXDB_USER_PASSWORD: mypassword
		""",
		}

		AEROS_ZONE_ID = "urn:ngsi-ld:Domain:NCSRD"

		AEROS_TOSCA_DESCRIPTOR_2 = """
		tosca_definitions_version: tosca_simple_yaml_1_3
		description: A test service for testing TOSCA generation
		node_templates:
		auto-component:
		artifacts:
		nginx-image:
		file: nginx
		type: tosca.artifacts.Deployment.Image.Container.Docker
		repository: docker_hub
		interfaces:
		Standard:
		create:
		implementation: application_image
		inputs:
		cliArgs: []
		envVars: []
		requirements:
		- network:
		properties:
		ports:
		port1:
		properties:
		protocol:
		- tcp
		source: 80
		port2:
		properties:
		protocol:
		- tcp
		source: 443
		exposePorts: false
		- host:
		node_filter:
		capabilities:
		- host:
		properties:
		cpu_arch:
		equal: x64
		realtime:
		equal: false
		cpu_usage:
		less_or_equal: '0.4'
		mem_size:
		greater_or_equal: '1'
		energy_efficiency:
		greater_or_equal: '10'
		green:
		greater_or_equal: '10'
		properties: null
		type: tosca.nodes.Container.Application
		"""

tests/edgecloud/test_e2e.py

+14 −13

Original line number	Diff line number	Diff line
		@@ -112,11 +112,6 @@ def test_onboard_app(edgecloud_client):
		pytest.fail(f"App onboarding failed unexpectedly: {e}")


		# @pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		# def test_timer_wait_15_seconds(edgecloud_client):
		# time.sleep(15)


		@pytest.fixture(scope="module")
		def app_instance_id(edgecloud_client):
		try:
		@@ -134,8 +129,8 @@ def test_deploy_app(app_instance_id):


		@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		def test_timer_wait_30_seconds(edgecloud_client):
		time.sleep(30)
		def test_timer_wait_60_seconds(edgecloud_client):
		time.sleep(60)


		@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		@@ -146,12 +141,18 @@ def test_undeploy_app(edgecloud_client, app_instance_id):
		pytest.fail(f"App undeployment failed unexpectedly: {e}")


		# @pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		# def test_delete_onboarded_app(edgecloud_client):
		# try:
		# edgecloud_client.delete_onboarded_app(app_id=APP_ONBOARD_MANIFEST["appId"])
		# except EdgeCloudPlatformError as e:
		# pytest.fail(f"App onboarding deletion failed unexpectedly: {e}")
		@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		def test_timer_wait_30_seconds(edgecloud_client):
		time.sleep(30)


		@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		def test_delete_onboarded_app(edgecloud_client):
		try:
		edgecloud_client.delete_onboarded_app(app_id=APP_ONBOARD_MANIFEST["appId"])
		except EdgeCloudPlatformError as e:
		pytest.fail(f"App onboarding deletion failed unexpectedly: {e}")


		# @pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
		# def test_delete_artefact(edgecloud_client):