implemented k8s gsma-compliant function calls (be5d6cee) · Commits · OOP / code / transformation-function-sdk

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

+116 −5

Original line number	Diff line number	Diff line
		@@ -18,6 +18,9 @@ from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.models.deploy_service_f
		from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.models.service_function_registration_request import (
		ServiceFunctionRegistrationRequest,
		)
		from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.models.required_resources import (
		RequiredResources
		)
		from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.utils.connector_db import (
		ConnectorDB,
		)
		@@ -536,6 +539,23 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):

		# ApplicationOnboardingManagement

		def _search_field_inside_body_gsma(self, keys: List[str], body: dict) -> List[str]:
		missing_fields : List[str] = []
		for key in keys:
		if "." in key:
		subfield = body
		for subkey in key.split("."):
		value = subfield.get(subkey)
		if not value:
		missing_fields.append(key)
		break

		subfield = value[0] if isinstance(value, list) else value
		else:
		if not body.get(key):
		missing_fields.append(key)
		return missing_fields

		def onboard_app_gsma(self, request_body: dict):
		"""
		Submits an application details to a partner OP.
		@@ -545,9 +565,80 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:param request_body: Payload with onboarding info.
		:return:
		"""
		pass
		needed_keys = [
		"appId",
		"appProviderId",
		"appMetaData.appName",
		"appMetaData.version",
		"artefactData.orig_am_artefact_repo_location_repo_url",
		"artefactData.orig_am_artefact_name",
		"artefactData.orig_am_artefact_version_info",
		]
		missing_fields: List[str] = self._search_field_inside_body_gsma(
		needed_keys, request_body
		)
		if len(missing_fields) != 0:
		error = {
		"title": "Invalid request",
		"detail": ",".join(missing_fields) + " are required",
		"cause": "MISSING_FIELD",
		"invalidParams": [
		{"param": "or".join(missing_fields), "reason": "field is missing"}
		],
		}

		return build_custom_http_response(
		status_code=400,
		content=error,
		headers={"Content-Type": "application/problem+json"},
		encoding="utf-8",
		url=None,
		request=None,
		)


		app_id = request_body.get("appId")
		app_provider_id = request_body.get("appProviderId")
		app_name = request_body.get("appMetaData").get("appName")
		version = request_body.get("appMetaData").get("version")

		artefact = request_body.get("artefactData")
		repo_url = artefact.get("orig_am_artefact_repo_location_repo_url")
		image_name = artefact.get("orig_am_artefact_name")
		image_tag = artefact.get("orig_am_artefact_version_info") or "latest"

		insert_doc = ServiceFunctionRegistrationRequest(
		service_function_id=app_id,
		service_function_image=repo_url + "/" + image_name + ":" + image_tag,
		service_function_name=app_name,
		service_function_type="package_type",
		application_ports= [],
		required_resources= RequiredResources(),
		app_provider=app_provider_id,
		version=version,
		)

		result = self.connector_db.insert_document_service_function(
		insert_doc.to_dict()
		)
		if type(result) is str:
		status_code = 409
		submitted_app = {"message": "App already exists"}
		else:
		submitted_app = camara_schemas.SubmittedApp(
		appId=camara_schemas.AppId(result.inserted_id)
		).model_dump(mode="json")
		status_code = 200
		return build_custom_http_response(
		status_code=status_code,
		content=submitted_app,
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)

		def get_onboarded_app_gsma(self, app_id: str) -> Dict:
		def get_onboarded_app_gsma(self, app_id: str) -> dict:
		"""
		Retrieves application details from partner OP

		@@ -565,16 +656,36 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
		:param request_body: Payload with updated onboarding info.
		:return:
		"""
		pass
		# None of the sent data has a 1-1 mapping to the saved object.
		return build_custom_http_response(
		status_code=200,
		content=None,
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)

		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:
		"""
		return self.delete_onboarded_app(app_id=app_id)
		result, code = self.connector_db.delete_document_service_function(_id=app_id)
		print(f"Removing application metadata: {app_id}")
		content = None
		if code != 200:
		status_code = 404
		content = {"status": 404, "code": "NOT_FOUND", "message": "Resource does not exist"}
		return build_custom_http_response(
		status_code=status_code,
		content=content,
		headers={"Content-Type": "application/json"},
		encoding="utf-8",
		url=None,
		request=None,
		)

		# ApplicationDeploymentManagement

src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/lib/utils/kubernetes_connector.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -191,9 +191,9 @@ class KubernetesConnector:
		pop_ = {}
		pop_["name"] = node.metadata.name
		pop_["uid"] = node.metadata.uid
		pop_["location"] = node.metadata.labels.get("location")
		pop_["location"] = node.metadata.labels.get("location") or ""
		pop_["serial"] = node.status.addresses[0].address
		pop_["node_type"] = node.metadata.labels.get("node_type")
		pop_["node_type"] = node.metadata.labels.get("node_type") or ""
		pop_["status"] = (
		"active" if node.status.conditions[-1].status == "True" else "inactive"
		)

Admin message