Final commit before leaving the company (f2491c46) · Commits · OOP / code / transformation-function-sdk

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

+103 −37

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ from typing import Dict, List, Optional
		from kubernetes.client import V1Deployment
		from requests import Response

		from sunrise6g_opensdk.edgecloud.core import gsma_schemas
		from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.core.piedge_encoder import (
		deploy_service_function,
		)
		@@ -372,40 +373,28 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		def get_edge_cloud_zones_details(
		self, zone_id: str, flavour_id: Optional[str] = None
		) -> Response:
		nodes = self.k8s_connector.get_node_details()
		node_details = None
		nodes = self.k8s_connector.get_node_details()
		for item in nodes.get("items"):
		# TODO: Fix uid stuff
		if item.get("metadata").get("uid") == zone_id:
		node_details = item
		break
		labels = node_details.get("metadata").get("labels")
		status = node_details.get("status")
		arch_type = labels.get("beta.kubernetes.io/arch")
		computeResourceQuotaLimits = [
		{
		"cpuArchType": arch_type,
		"numCPU": status.get("capacity").get("cpu"),
		"memory": status.get("capacity").get("memory"),
		# "memory": int(status.get("capacity").get("memory")) / (1024 * 1024),
		}
		]
		reservedComputeResources = [
		{
		"cpuArchType": arch_type,
		"numCPU": status.get("allocatable").get("cpu"),
		"memory": status.get("allocatable").get("memory"),
		# "memory": int(status.get("allocatable").get("memory")) / (1024 * 1024),
		}
		]
		flavoursSupported = []
		node_details["computeResourceQuotaLimits"] = computeResourceQuotaLimits
		node_details["reservedComputeResources"] = reservedComputeResources
		node_details["flavoursSupported"] = flavoursSupported
		node_details["zoneId"] = zone_id
		if node_details is None:
		return build_custom_http_response(
		status_code=404,
		content='Zone not found',
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)
		else:
		gsma_zone_details = self.__get_zone_details_gsma(node_details)
		gsma_zone_details["zoneId"] = zone_id
		return build_custom_http_response(
		status_code=200,
		content=node_details,
		content=gsma_zone_details,
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		@@ -438,25 +427,56 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		# FederationManagement

		def get_edge_cloud_zones_list_gsma(self) -> List:
		def get_edge_cloud_zones_list_gsma(self) -> Response:
		"""
		Retrieves list of all Zones

		:return: List.
		"""
		pass
		nodes_response = self.k8s_connector.get_PoPs()
		zone_list = []
		for node in nodes_response:
		zone = {}
		zone["zoneId"] = node.get("uid")
		zone["geographyDetails"] = node.get("location")
		zone_list.append(zone)
		validated_data = gsma_schemas.ZonesList.model_validate(zone_list)
		return build_custom_http_response(
		status_code=200,
		content=[zone.model_dump_json() for zone in validated_data.root],
		headers={"Content-Type": 'application/json'},
		encoding="utf-8",
		url=None,
		request=None,
		)

		# AvailabilityZoneInfoSynchronization

		def get_edge_cloud_zones_gsma(self) -> List:
		def get_edge_cloud_zones_gsma(self) -> Response:
		"""
		Retrieves details of all Zones

		:return: List.
		"""
		pass
		zone_list = []
		nodes = self.k8s_connector.get_node_details()
		for node in nodes.get("items"):
		zone_id = node.get("metadata").get("uid")
		gsma_details = self.__get_zone_details_gsma(node)
		gsma_details["zoneId"] = zone_id
		zone_list.append(gsma_details)
		validated_data = gsma_schemas.ZoneRegisteredDataList.model_validate(zone_list)
		return build_custom_http_response(
		status_code=200,
		content=validated_data.model_dump(),
		headers={"Content-Type": 'application/json'},
		encoding="utf-8",
		url=None,
		request=None,
		)

		def get_edge_cloud_zone_details_gsma(self, zone_id: str) -> Dict:

		def get_edge_cloud_zone_details_gsma(self, zone_id: str) -> Response:
		"""
		Retrieves details of a specific Edge Cloud Zone reserved
		for the specified zone by the partner OP.
		@@ -464,7 +484,24 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:param zone_id: Unique identifier of the Edge Cloud Zone.
		:return: Dictionary with Edge Cloud Zone details.
		"""
		pass
		node_details = None
		nodes = self.k8s_connector.get_node_details()
		for item in nodes.get("items"):
		# TODO: Fix uid stuff
		if item.get("metadata").get("uid") == zone_id:
		node_details = item
		break
		gsma_zone_details = self.__get_zone_details_gsma(node_details)
		gsma_zone_details["zoneId"] = zone_id
		validated_data = gsma_schemas.ZoneRegisteredData.model_validate(gsma_zone_details)
		return build_custom_http_response(
		status_code=200,
		content=validated_data,
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)

		# ArtefactManagement

		@@ -530,18 +567,18 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		"""
		pass

		def delete_onboarded_app_gsma(self, app_id: str):
		def delete_onboarded_app_gsma(self, app_id: str) -> Response:
		"""
		Deboards an application from specific partner OP zones

		:param app_id: Identifier of the application onboarded.
		:return:
		"""
		pass
		return self.delete_onboarded_app(app_id=app_id)

		# ApplicationDeploymentManagement

		def deploy_app_gsma(self, request_body: dict) -> Dict:
		def deploy_app_gsma(self, request_body: dict) -> Response:
		"""
		Instantiates an application on a partner OP zone.

		@@ -581,3 +618,32 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:return:
		"""
		pass

		def __get_zone_details_gsma(self, node_details):
		gsma_details = {}
		labels = node_details.get("metadata").get("labels")
		status = node_details.get("status")
		arch_type = labels.get("beta.kubernetes.io/arch")
		computeResourceQuotaLimits = [
		{
		"cpuArchType": arch_type,
		"numCPU": status.get("capacity").get("cpu"),
		"memory": status.get("capacity").get("memory"),
		# "memory": int(status.get("capacity").get("memory")) / (1024 * 1024),
		}
		]
		reservedComputeResources = [
		{
		"cpuArchType": arch_type,
		"numCPU": status.get("allocatable").get("cpu"),
		"memory": status.get("allocatable").get("memory"),
		# "memory": int(status.get("allocatable").get("memory")) / (1024 * 1024),
		}
		]
		flavoursSupported = []
		gsma_details["computeResourceQuotaLimits"] = computeResourceQuotaLimits
		gsma_details["reservedComputeResources"] = reservedComputeResources
		gsma_details["flavoursSupported"] = flavoursSupported
		gsma_details["networkResources"] = {"egressBandWidth": 0, "dedicatedNIC": 0, "supportSriov": False,"supportDPDK": False}
		gsma_details["zoneServiceLevelObjsInfo"] = {"latencyRanges": {"minLatency": 0,"maxLatency": 0}, "jitterRanges": {"minJitter": 0,"maxJitter": 0}, "throughputRanges": {"minThroughput": 0, "maxThroughput": 0}}
		return gsma_details

Admin message