Resolved conflicts (3ae87ee4) · Commits · OOP / code / open-exposure-gateway

.coverage

(52 KiB)

File changed.

No diff preview for this file type.

README.md

+34 −3

Original line number	Diff line number	Diff line
		# Open Exposure Gateway

		A Server Implementation of the CAMARA Edge Application Management API as specified [here](https://github.com/camaraproject/EdgeCloud/blob/main/code/API_definitions/Edge-Application-Management.yaml)

		The Open Exposure Gateway (OEG) is a Python web service implementation. It implements the Open Exposure Gateway role of the Operator Platform, defined by the [GSMA Operator Platform Group (OPG)](https://www.gsma.com/solutions-and-impact/technologies/networks/gsma-operator-platform-group-september-2024-publications/).

		### Overview

		@@ -12,7 +11,39 @@ This server was generated by the [swagger-codegen](https://github.com/swagger-ap
		It uses the [Connexion](https://github.com/zalando/connexion) library on top of Flask.
		It propagates requests to the [Sunrise 6G OP service resource manager](https://github.com/OpenOperatorPlatform/service_resource_manager), as configured in the `.env`.

		### Usage
		OEG implements the Northbound CAMARA APIs for application providers to interact with different infrastructures, including:

		- Application metadata lifecycle management (registering, retrieving, deleting apps)
		- Application instantiation management (instantation, and termination)
		- Federation Management APIs to onboard applications accross partners' operator platforms

		OEG acts as a middleware layer between the CAMARA APIs and the Service Resource Manager (SRM), by enabling a common exposure of heteregeneous infrastructure resources.

		<br>

		\| Edge Cloud Management API \| Federation Management API \| Network Exposure API (QoD & Traffic Influence)\|
		\| ------------- \| ------------- \| ------------- \|
		\| Application Metadata registration \| Create one direction federation \| Create QoD Session \|
		\| App Metadata Removal \| Retrieve details about federation with the partner OP \| Remove QoD Session \|
		\| App Metadata Retrieval \| Retrieve the existing federationContextId with partner \| Retrieve QoD Session \|
		\| Application Instantiation \| Remove existing federation with partner OP \| Create TrafficeInfluence Resource \|
		\| Application Instance Retrieval \| \| Remove TrafficeInfluence Resource \|
		\| Application Instance Removal \| \| Retrieve TrafficeInfluence Resource \|


		## Deployment

		OEG can be deployed in a Kubernetes cluster by executing the file _oeg-deployment.yaml_ located in the root folder. This file will create a OEG Deployment resource and its supporting native K8s Service. The following table contains the necesssary environment variables for the Kubernetes adapter.

		\| Variable \| Description \|
		\|-----------------------------\|-------------------------------------------------------------\|
		\| `SRM_HOST` \| Base URL of the Service Resource Manager \|
		\| `FEDERATION_MANAGER_HOST` \| Base URL of the Federation Manager \|



		You can also run the OEG outside of k8s environment:

		Before running the server you need to `mv env.sample .env` and update variables to properly link to a running OP service resource manager instance.

		To run the server, please execute the following from the root directory:

coverage.xml

+327 −163

File changed.

Preview size limit exceeded, changes collapsed.

edge_cloud_management_api/controllers/federation_manager_controller.py

+47 −57

Original line number	Diff line number	Diff line
		from flask import request, jsonify
		from flask import request, jsonify, Response
		import json
		from edge_cloud_management_api.services.federation_services import FederationManagerClientFactory

		# Factory pattern
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()

		def handle_response(status_code, title, detail):
		return Response(
		json.dumps({
		"title": title,
		"detail": detail,
		"status": status_code
		}),
		status=status_code,
		content_type="application/problem+json"
		)


		def create_federation():
		"""
		POST /partner
		Forwards the federation creation request to Federation Manager.
		"""
		"""POST /partner - Create federation with partner OP."""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		body = request.get_json()
		result = federation_client.post_partner(body)
		return jsonify(result), 200 if "error" not in result else 502
		if "error" in result:
		return handle_response(502, "Federation Error", result["error"])
		return jsonify(result), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 400
		return handle_response(400, "Bad Request", str(e))


		def get_federation(federationContextId):
		"""
		GET /{federationContextId}/partner
		Forwards the GET federation info request.
		"""
		"""GET /{federationContextId}/partner - Get federation info."""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		result = federation_client.get_partner(federationContextId)
		return jsonify(result), 200 if "error" not in result else 502
		if "error" in result:
		return handle_response(502, "Federation Error", result["error"])
		return jsonify(result), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 400
		return handle_response(400, "Bad Request", str(e))


		def delete_federation(federationContextId):
		"""
		DELETE /{federationContextId}/partner
		Forwards the DELETE federation request.
		"""
		"""DELETE /{federationContextId}/partner - Delete federation."""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		result = federation_client.delete_partner(federationContextId)
		return jsonify(result), 200 if "error" not in result else 502
		if "error" in result:
		return handle_response(502, "Federation Error", result["error"])
		return jsonify(result), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 400
		return handle_response(400, "Bad Request", str(e))


		def get_federation_context_ids():
		"""
		GET /fed-context-id
		Forwards the request to fetch federation context IDs.
		"""
		"""GET /fed-context-id - Fetch federationContextId(s)."""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		result = federation_client.get_federation_context_ids()
		return jsonify(result), 200 if "error" not in result else 502
		if "error" in result:
		return handle_response(502, "Federation Error", result["error"])
		return jsonify(result), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 400
		return handle_response(400, "Bad Request", str(e))


		def onboard_application_to_partner(federationContextId):
		"""
		POST /{federationContextId}/application/onboarding
		Forwards the onboarding request to the Federation Manager.
		"""
		"""POST /{federationContextId}/application/onboarding - Onboard app."""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		body = request.get_json()
		result = federation_client.onboard_application(federationContextId, body)
		return jsonify(result), 202 if "error" not in result else 502
		if "error" in result:
		return handle_response(502, "Onboarding Error", result["error"])
		return jsonify(result), 202
		except Exception as e:
		return jsonify({"error": str(e)}), 400
		return handle_response(400, "Bad Request", str(e))


		def get_onboarded_app(federationContextId, appId):
		"""
		GET /{federationContextId}/application/onboarding/app/{appId}
		Retrieves onboarding info of a federated app from a partner OP.
		"""
		"""GET /{federationContextId}/application/onboarding/app/{appId}"""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		result = federation_client.get_onboarded_app(federationContextId, appId)
		if "error" in result:
		return jsonify(result), result.get("status_code", 502)
		return handle_response(result.get("status_code", 502), "Retrieval Error", result["error"])
		return jsonify(result), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 500
		return handle_response(500, "Internal Server Error", str(e))


		def delete_onboarded_app(federationContextId, appId):
		"""
		DELETE /{federationContextId}/application/onboarding/app/{appId}
		Deboards the application and deletes it from the partner OP.
		"""
		"""DELETE /{federationContextId}/application/onboarding/app/{appId}"""
		try:
		factory = FederationManagerClientFactory()
		federation_client = factory.create_federation_client()
		result = federation_client.delete_onboarded_app(federationContextId, appId)
		if "error" in result:
		return jsonify(result), result.get("status_code", 502)
		return handle_response(result.get("status_code", 502), "Deletion Error", result["error"])
		return jsonify({"message": f"App {appId} successfully deleted from partner"}), 200
		except Exception as e:
		return jsonify({"error": str(e)}), 500
		return handle_response(500, "Internal Server Error", str(e))

edge_cloud_management_api/controllers/network_functions_controller.py

+2 −2

Original line number	Diff line number	Diff line
		from flask import jsonify
		from pydantic import Field, ValidationError
		from pydantic import ValidationError #Field
		from edge_cloud_management_api.managers.log_manager import logger
		from edge_cloud_management_api.services.pi_edge_services import PiEdgeAPIClientFactory