Loading src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/client.py +199 −30 Original line number Diff line number Diff line Loading @@ -580,44 +580,213 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): def deploy_app_gsma(self, request_body: dict) -> Response: """ Instantiates an application on a partner OP zone. :param request_body: Payload with deployment info. :return: Dictionary with deployment details. GSMA - Instantiates an application in a specific partner OP zone. """ pass def get_deployed_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Dict: """ Retrieves an application instance details from partner OP. logging.info("GSMA deploy app request received") try: # Validate and extract required fields app_id = request_body.get("appId") zone_info = request_body.get("zoneInfo") callback = request_body.get("appInstCallbackLink") if not app_id or not zone_info: error = { "title": "Invalid request", "detail": "appId and zoneInfo are required", "cause": "MISSING_FIELD", "invalidParams": [ {"param": "appId or zoneInfo", "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, ) zone_id = zone_info.get("zoneId") # Retrieve the application record from DB app_list = self.connector_db.get_documents_from_collection( "service_functions", input_type="_id", input_value=app_id ) if len(app_list) < 1: error = { "title": "Application not found", "detail": f"App with id {app_id} not found", "cause": "NOT_FOUND", } return build_custom_http_response(404, error) :param app_id: Identifier of the app. :param app_instance_id: Identifier of the deployed instance. :param zone_id: Identifier of the zone :return: Dictionary with application instance details app = app_list[0] # Prepare deployment descriptor sf = DeployServiceFunction( service_function_name=app.get("name"), service_function_instance_name=app.get("name"), ) # Call KubernetesConnector to deploy result = deploy_service_function( service_function=sf, connector_db=self.connector_db, kubernetes_connector=self.k8s_connector, ) # Success → Kubernetes returned a Deployment object from kubernetes.client import V1Deployment if isinstance(result, V1Deployment): deployment_uid = result.metadata.uid response = { "zoneId": zone_id, "appInstIdentifier": deployment_uid, } validated = gsma_schemas.AppInstanceCreatedResponse.model_validate(response) return build_custom_http_response( status_code=202, content=validated.model_dump(), headers={"Content-Type": "application/json"}, encoding="utf-8", url=None, request=None, ) # Conflict or error if "Conflict" in str(result): error = { "title": "Conflict", "detail": "Application already instantiated in this zone", "cause": "CONFLICT", } return build_custom_http_response(409, error) # Unknown error error = { "title": "Deployment failed", "detail": str(result), "cause": "UNKNOWN_ERROR", } return build_custom_http_response(500, error) except Exception as e: error = { "title": "Internal error", "detail": str(e), "cause": "EXCEPTION", } return build_custom_http_response(500, error) def get_deployed_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Response: """ GSMA - Retrieve details of a specific deployed application instance. """ pass logging.info(f"GSMA: get deployed app. appId={app_id}, instance={app_instance_id}") deployments = self.k8s_connector.get_deployed_service_functions(self.connector_db) for item in deployments: if item.get("appId") == app_id and item.get("appInstanceId") == app_instance_id: response = { "zoneId": zone_id, "appInstIdentifier": app_instance_id, "status": item.get("status", "unknown"), } validated = gsma_schemas.AppInstanceInfo.model_validate(response) return build_custom_http_response( 200, validated.model_dump(), headers={"Content-Type": "application/json"}, encoding="utf-8", ) error = { "title": "Not found", "detail": "Application instance does not exist", "cause": "NOT_FOUND", } return build_custom_http_response(404, error) def get_all_deployed_apps_gsma(self) -> Response: """ Retrieves all instances for a given application of partner OP :param app_id: Identifier of the app. :param app_provider: App provider :return: List with application instances details GSMA - Retrieve all deployed application instances on the OP. """ pass logging.info("GSMA: list all deployed app instances") deployments = self.k8s_connector.get_deployed_service_functions(self.connector_db) response_list = [] for item in deployments: response_list.append( { "zoneId": item.get("edgeCloudZoneId", ""), "appInstIdentifier": item.get("appInstanceId"), "status": item.get("status", "unknown"), } ) validated = gsma_schemas.AppInstanceList.model_validate(response_list) return build_custom_http_response( 200, validated.model_dump(), headers={"Content-Type": "application/json"}, encoding="utf-8", ) def undeploy_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str): """ Terminate an application instance on a partner OP zone. :param app_id: Identifier of the app. :param app_instance_id: Identifier of the deployed app. :param zone_id: Identifier of the zone :return: GSMA - Terminate an application instance. """ pass logging.info(f"GSMA undeploy appInstance {app_instance_id}") try: deployments = self.k8s_connector.get_deployed_service_functions(self.connector_db) for sf in deployments: if sf.get("appInstanceId") == app_instance_id: self.k8s_connector.delete_service_function( self.connector_db, sf["service_function_instance_name"], ) return build_custom_http_response( 204, None, headers={"Content-Type": "application/json"}, ) error = { "title": "Not found", "detail": "App instance does not exist", "cause": "NOT_FOUND", } return build_custom_http_response(404, error) except Exception as e: error = { "title": "Internal error", "detail": str(e), "cause": "EXCEPTION", } return build_custom_http_response(500, error) def __get_zone_details_gsma(self, node_details): gsma_details = {} Loading Loading
src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/client.py +199 −30 Original line number Diff line number Diff line Loading @@ -580,44 +580,213 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): def deploy_app_gsma(self, request_body: dict) -> Response: """ Instantiates an application on a partner OP zone. :param request_body: Payload with deployment info. :return: Dictionary with deployment details. GSMA - Instantiates an application in a specific partner OP zone. """ pass def get_deployed_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Dict: """ Retrieves an application instance details from partner OP. logging.info("GSMA deploy app request received") try: # Validate and extract required fields app_id = request_body.get("appId") zone_info = request_body.get("zoneInfo") callback = request_body.get("appInstCallbackLink") if not app_id or not zone_info: error = { "title": "Invalid request", "detail": "appId and zoneInfo are required", "cause": "MISSING_FIELD", "invalidParams": [ {"param": "appId or zoneInfo", "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, ) zone_id = zone_info.get("zoneId") # Retrieve the application record from DB app_list = self.connector_db.get_documents_from_collection( "service_functions", input_type="_id", input_value=app_id ) if len(app_list) < 1: error = { "title": "Application not found", "detail": f"App with id {app_id} not found", "cause": "NOT_FOUND", } return build_custom_http_response(404, error) :param app_id: Identifier of the app. :param app_instance_id: Identifier of the deployed instance. :param zone_id: Identifier of the zone :return: Dictionary with application instance details app = app_list[0] # Prepare deployment descriptor sf = DeployServiceFunction( service_function_name=app.get("name"), service_function_instance_name=app.get("name"), ) # Call KubernetesConnector to deploy result = deploy_service_function( service_function=sf, connector_db=self.connector_db, kubernetes_connector=self.k8s_connector, ) # Success → Kubernetes returned a Deployment object from kubernetes.client import V1Deployment if isinstance(result, V1Deployment): deployment_uid = result.metadata.uid response = { "zoneId": zone_id, "appInstIdentifier": deployment_uid, } validated = gsma_schemas.AppInstanceCreatedResponse.model_validate(response) return build_custom_http_response( status_code=202, content=validated.model_dump(), headers={"Content-Type": "application/json"}, encoding="utf-8", url=None, request=None, ) # Conflict or error if "Conflict" in str(result): error = { "title": "Conflict", "detail": "Application already instantiated in this zone", "cause": "CONFLICT", } return build_custom_http_response(409, error) # Unknown error error = { "title": "Deployment failed", "detail": str(result), "cause": "UNKNOWN_ERROR", } return build_custom_http_response(500, error) except Exception as e: error = { "title": "Internal error", "detail": str(e), "cause": "EXCEPTION", } return build_custom_http_response(500, error) def get_deployed_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str) -> Response: """ GSMA - Retrieve details of a specific deployed application instance. """ pass logging.info(f"GSMA: get deployed app. appId={app_id}, instance={app_instance_id}") deployments = self.k8s_connector.get_deployed_service_functions(self.connector_db) for item in deployments: if item.get("appId") == app_id and item.get("appInstanceId") == app_instance_id: response = { "zoneId": zone_id, "appInstIdentifier": app_instance_id, "status": item.get("status", "unknown"), } validated = gsma_schemas.AppInstanceInfo.model_validate(response) return build_custom_http_response( 200, validated.model_dump(), headers={"Content-Type": "application/json"}, encoding="utf-8", ) error = { "title": "Not found", "detail": "Application instance does not exist", "cause": "NOT_FOUND", } return build_custom_http_response(404, error) def get_all_deployed_apps_gsma(self) -> Response: """ Retrieves all instances for a given application of partner OP :param app_id: Identifier of the app. :param app_provider: App provider :return: List with application instances details GSMA - Retrieve all deployed application instances on the OP. """ pass logging.info("GSMA: list all deployed app instances") deployments = self.k8s_connector.get_deployed_service_functions(self.connector_db) response_list = [] for item in deployments: response_list.append( { "zoneId": item.get("edgeCloudZoneId", ""), "appInstIdentifier": item.get("appInstanceId"), "status": item.get("status", "unknown"), } ) validated = gsma_schemas.AppInstanceList.model_validate(response_list) return build_custom_http_response( 200, validated.model_dump(), headers={"Content-Type": "application/json"}, encoding="utf-8", ) def undeploy_app_gsma(self, app_id: str, app_instance_id: str, zone_id: str): """ Terminate an application instance on a partner OP zone. :param app_id: Identifier of the app. :param app_instance_id: Identifier of the deployed app. :param zone_id: Identifier of the zone :return: GSMA - Terminate an application instance. """ pass logging.info(f"GSMA undeploy appInstance {app_instance_id}") try: deployments = self.k8s_connector.get_deployed_service_functions(self.connector_db) for sf in deployments: if sf.get("appInstanceId") == app_instance_id: self.k8s_connector.delete_service_function( self.connector_db, sf["service_function_instance_name"], ) return build_custom_http_response( 204, None, headers={"Content-Type": "application/json"}, ) error = { "title": "Not found", "detail": "App instance does not exist", "cause": "NOT_FOUND", } return build_custom_http_response(404, error) except Exception as e: error = { "title": "Internal error", "detail": str(e), "cause": "EXCEPTION", } return build_custom_http_response(500, error) def __get_zone_details_gsma(self, node_details): gsma_details = {} Loading
