Add deprecated i2edge contract adapter, flag i2cat_ric as deprecated (b44c7cf2) · Commits · OOP / code / transformation-function-sdk

src/openop_tfsdk/contract/adapters/init.py

+2 −0

Changes for src/openop_tfsdk/contract/adapters/__init__.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ the `adapter_class` path in the YAML, which names another distribution.
		"""
		from typing import Optional

		from openop_tfsdk.contract.adapters.edgecloud import I2edgeCapabilityAdapter
		from openop_tfsdk.contract.adapters.kubernetes import (
		KubernetesCapabilityAdapter,
		)
		@@ -27,6 +28,7 @@ ADAPTERS_BY_TYPE = {
		Open5gsCapabilityAdapter,
		Open5gcoreCapabilityAdapter,
		I2catRicCapabilityAdapter,
		I2edgeCapabilityAdapter,
		)
		}

src/openop_tfsdk/contract/adapters/edgecloud.py

0 → 100644

+139 −0

Changes for src/openop_tfsdk/contract/adapters/edgecloud.py: 139 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		##
		# This file is part of the TF SDK
		##
		"""
		Contract adapter for the i2Edge platform of the previous codebase. It deploys an
		application that is already onboarded on i2Edge.

		Deprecated: unsupported in this release (no test access).
		"""
		import time
		from typing import Optional

		import pydantic
		import structlog

		from openop_tfsdk.contract.errors import (
		DomainExecutionFailed,
		ResourceNotFound,
		ValidationError,
		)
		from openop_tfsdk.contract.execution_models import (
		CapabilityLifecycleRequest,
		CapabilityLifecycleResult,
		CapabilityRuntimeState,
		CapabilityRuntimeStateRequest,
		LifecycleState,
		RuntimeState,
		)
		from openop_tfsdk.contract.models import SupportedAction
		from openop_tfsdk.edgecloud.adapters.i2edge.client import (
		EdgeApplicationManager as I2edgeClient,
		)
		from openop_tfsdk.edgecloud.adapters.i2edge.common import I2EdgeError

		log = structlog.get_logger(__name__)

		WORKLOAD_ACTIONS = [
		SupportedAction.ACTIVATE,
		SupportedAction.TERMINATE,
		SupportedAction.READ_RUNTIME_STATE,
		]

		RUNTIME_KINDS = ["container"]

		POLL_SECONDS = 5


		class I2edgeCapabilityAdapter:
		"""Executes workload on i2Edge through its EdgeApplicationManager client."""

		adapter_type = "i2edge"

		def __init__(self, infrastructure: Optional[dict] = None):
		infrastructure = infrastructure or {}
		self._client = I2edgeClient(
		base_url=infrastructure.get("base_url"),
		flavour_id=infrastructure.get("flavour_id"),
		)
		self._deploy_timeout = infrastructure.get("deploy_timeout_seconds", 300)

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		if capability_kind == "workload":
		return list(WORKLOAD_ACTIONS), list(RUNTIME_KINDS)
		return [], []

		def activate(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		if request.capability_kind != "workload":
		raise ValidationError(f"i2edge does not activate {request.capability_kind}")
		unit = request.deployment_unit
		if unit is None or not unit.artifact_ref:
		raise ValidationError("activate needs a deployment_unit with an artifact_ref")
		# The i2Edge zone is the domain SRM placed the workload on.
		zones = [{"EdgeCloudZone": {"edgeCloudZoneId": request.target.domain_ref}}]
		response = self._call("deploy_app", unit.artifact_ref, zones)
		app_instance_id = response.json()["appInstanceId"]
		self._wait_until_ready(app_instance_id)
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		state=LifecycleState.ACTIVE,
		external_id=app_instance_id,
		)

		def read_runtime_state(self, request: CapabilityRuntimeStateRequest) -> CapabilityRuntimeState:
		if not request.external_id:
		raise ValidationError("read_runtime_state needs the external_id returned by activate")
		try:
		status = self._status(request.external_id)
		except ResourceNotFound:
		status = None
		return CapabilityRuntimeState(
		external_id=request.external_id,
		capability_kind=request.capability_kind,
		state=RuntimeState.ACTIVE if status == "ready" else RuntimeState.UNKNOWN,
		)

		def terminate(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		if not request.external_id:
		raise ValidationError("terminate needs the external_id returned by activate")
		self._call("undeploy_app", request.external_id)
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		state=LifecycleState.DEACTIVATED,
		external_id=request.external_id,
		)

		def _wait_until_ready(self, app_instance_id: str):
		# SRM requires an active workload, so wait like helm --wait and undo like --atomic.
		deadline = time.monotonic() + self._deploy_timeout
		try:
		while self._status(app_instance_id) != "ready":
		if time.monotonic() >= deadline:
		raise DomainExecutionFailed("i2edge deployment did not become ready in time")
		time.sleep(POLL_SECONDS)
		except Exception:
		self._call("undeploy_app", app_instance_id)
		raise

		def _status(self, app_instance_id: str) -> str:
		return self._call("get_deployed_app", app_instance_id).json()["appInstance"]["status"]

		def _call(self, method: str, *args):
		try:
		return getattr(self._client, method)(*args)
		except pydantic.ValidationError as error:
		log.error("%s rejected the request: %s", method, error)
		raise ValidationError(f"{method}: invalid request for i2edge") from error
		except I2EdgeError as error:
		log.error("%s failed: %s", method, error)
		# The client has no status code, only a message: "got 404" or "... not found ...".
		if "got 404" in str(error) or "not found" in str(error):
		raise ResourceNotFound(f"{method}: not found") from error
		raise DomainExecutionFailed(f"{method} failed") from error
		except Exception as error:
		log.error("%s failed: %s", method, error)
		raise DomainExecutionFailed(f"{method} failed") from error

src/openop_tfsdk/contract/adapters/oran.py

+2 −0

Changes for src/openop_tfsdk/contract/adapters/oran.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,6 +5,8 @@
		"""
		Contract adapter for the i2CAT O-RAN RIC of the previous codebase. Its client has the
		same QoD methods as the network cores, so it reuses their translation.

		Deprecated: unsupported in this release (closed source, no test access).
		"""
		from typing import Optional