feat(fm): deploy Helm charts on partners, both ways (d7af1aa6) · Commits · OOP / code / federation-manager

pyproject.toml

+1 −1

Changes for pyproject.toml: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ dependencies = [
		"nats-py>=2.6",
		"pydantic-settings>=2.4",
		"pyjwt[crypto]>=2.9",
		"pyyaml>=6.0",
		"sqlalchemy[asyncio]>=2.0",
		"structlog>=25.1",
		]
		@@ -28,7 +29,6 @@ dev = [
		"pytest>=8.0",
		"pytest-asyncio>=0.24",
		"pytest-cov>=6.0",
		"pyyaml>=6.0",
		"ruff>=0.6",
		]

src/federation_manager/adapters/http/chart_inspector.py

0 → 100644

+103 −0

Changes for src/federation_manager/adapters/http/chart_inspector.py: 103 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import io
		import tarfile
		from typing import Any

		import httpx
		import yaml

		from federation_manager.domain.errors import OnboardingNotSupported

		# charts are small; anything this big is not a chart we can read in memory
		_MAX_CHART_BYTES = 20 * 1024 * 1024


		class HttpxChartInspector:
		"""Reads which container images a public Helm chart runs, from its values.yaml.

		GSMA needs every component of a Helm artefact to name its images as uploaded files, and
		a CAMARA Helm app names only the chart. Charts conventionally declare their images as
		`image: {registry?, repository, tag?}` blocks in values.yaml, with Chart.yaml's
		appVersion as the default tag, so that is what is read; the chart is not rendered.
		"""

		def __init__(self, client: httpx.AsyncClient) -> None:
		self._client = client

		async def container_images(self, chart_url: str) -> list[str]:
		archive = await self._download(chart_url)
		try:
		values, chart = _top_level_files(archive)
		except (tarfile.TarError, yaml.YAMLError, UnicodeDecodeError):
		raise _unreadable(chart_url, "it is not a readable chart archive") from None
		images = _images(values, default_tag=str(chart.get("appVersion") or ""))
		if not images:
		raise _unreadable(chart_url, "its values.yaml declares no image repository")
		return images

		async def _download(self, chart_url: str) -> bytes:
		try:
		async with self._client.stream(
		"GET", chart_url, follow_redirects=True, timeout=30.0
		) as response:
		if response.status_code != 200:
		raise _unreadable(chart_url, f"fetching it answered {response.status_code}")
		body = bytearray()
		async for chunk in response.aiter_bytes():
		body.extend(chunk)
		if len(body) > _MAX_CHART_BYTES:
		raise _unreadable(chart_url, "it is larger than 20 MB")
		return bytes(body)
		except httpx.HTTPError:
		raise _unreadable(chart_url, "it could not be fetched") from None


		def _unreadable(chart_url: str, reason: str) -> OnboardingNotSupported:
		return OnboardingNotSupported(
		f"The Helm chart {chart_url} can't be onboarded at a partner: {reason}."
		)


		def _top_level_files(archive: bytes) -> tuple[Any, dict[str, Any]]:
		values: Any = None
		chart: dict[str, Any] = {}
		with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as tar:
		for member in tar.getmembers():
		parts = member.name.split("/")
		# <chart>/values.yaml, not a subchart's under <chart>/charts/
		if len(parts) != 2 or not member.isfile():
		continue
		extracted = tar.extractfile(member)
		if extracted is None:
		continue
		if parts[1] == "values.yaml":
		values = yaml.safe_load(extracted.read().decode())
		elif parts[1] == "Chart.yaml":
		chart = yaml.safe_load(extracted.read().decode()) or {}
		return values, chart


		def _images(values: Any, default_tag: str) -> list[str]:
		found: list[str] = []

		def walk(node: Any) -> None:
		if isinstance(node, dict):
		repository = node.get("repository")
		if isinstance(repository, str) and repository:
		registry = node.get("registry")
		name = (
		f"{registry}/{repository}"
		if isinstance(registry, str) and registry
		else repository
		)
		tag = str(node.get("tag") or default_tag)
		image = f"{name}:{tag}" if tag else name
		if image not in found:
		found.append(image)
		for child in node.values():
		walk(child)
		elif isinstance(node, list):
		for child in node:
		walk(child)

		walk(values)
		return found

src/federation_manager/application/inbound_onboarding.py

+68 −29

Changes for src/federation_manager/application/inbound_onboarding.py: 68 added lines, 29 removed lines.
Original line number	Diff line number	Diff line
		@@ -71,12 +71,14 @@ class InboundOnboardingService:
		self, partner: PartnerOP, federation_context_id: str, upload: ArtefactUpload
		) -> None:
		await self._live_inbound(partner, federation_context_id)
		if (
		upload.artefact_descriptor_type != "COMPONENTSPEC"
		or upload.artefact_virt_type != "CONTAINER_TYPE"
		chart = _helm_chart(upload)
		if upload.artefact_virt_type != "CONTAINER_TYPE" or (
		upload.artefact_descriptor_type != "COMPONENTSPEC" and chart is None
		):
		raise OnboardingNotSupported(
		"Only container artefacts described by componentSpec are supported."
		"Only container artefacts are supported, described by componentSpec or by a "
		"Helm chart in a public repository (repoType PUBLICREPO with "
		"artefactRepoLocation.repoURL)."
		)
		images = {str(image) for spec in upload.component_spec for image in spec.images}
		missing = sorted(
		@@ -95,6 +97,7 @@ class InboundOnboardingService:
		{
		"artefact_name": upload.artefact_name,
		"version": upload.artefact_version_info,
		**({"helm_chart": chart} if chart is not None else {}),
		"component_spec": [
		spec.model_dump(mode="json", by_alias=True) for spec in upload.component_spec
		],
		@@ -123,11 +126,14 @@ class InboundOnboardingService:
		if missing:
		raise OnboardingResourceUnknown("artefact", missing)

		units = [
		await self._deployment_unit(partner, ComponentSpec.model_validate(spec))
		for artefact in artefacts
		for spec in artefact["component_spec"]
		]
		units: list[dict[str, Any]] = []
		for artefact in artefacts:
		specs = [ComponentSpec.model_validate(spec) for spec in artefact["component_spec"]]
		if "helm_chart" in artefact:
		# the chart deploys every component itself: one unit, one release
		units.append(_helm_unit(artefact, specs))
		else:
		units.extend([await self._deployment_unit(partner, spec) for spec in specs])
		# derived, so a retried onboarding meets the specification an earlier try created
		specification_id = uuid5(
		ONBOARDING_NAMESPACE, f"spec:{partner.id}:{request.app_id}:{version}"
		@@ -142,7 +148,9 @@ class InboundOnboardingService:
		"version": version,
		"descriptor": {
		"source_api": "gsma-opg04-onboarding",
		"artifact_type": "container",
		"artifact_type": "helm"
		if any(unit["runtime_kind"] == "helm" for unit in units)
		else "container",
		},
		"metadata": {"federation_partner": partner.mcc_mnc},
		},
		@@ -186,20 +194,64 @@ class InboundOnboardingService:
		async def _deployment_unit(self, partner: PartnerOP, spec: ComponentSpec) -> dict[str, Any]:
		image_file = await self._catalogue.get_resource(partner.id, FILE, str(spec.images[0]))
		assert image_file is not None # checked when the artefact was uploaded
		cpu = _millicores(spec.compute_resource_profile.num_cpu)
		memory = spec.compute_resource_profile.memory
		return {
		"ref": spec.component_name,
		"name": spec.component_name,
		"runtime_kind": "container",
		"artifact_ref": image_file["image"],
		"resource_requirements": {
		"resource_requirements": _resource_requirements([spec]),
		"metadata": {"repo": {"type": "PUBLICREPO"}},
		}

		async def _live_inbound(self, partner: PartnerOP, federation_context_id: str) -> None:
		context = await self._contexts.find_inbound(partner.id, federation_context_id)
		if context is None or context.is_terminated():
		raise FederationContextUnknown(partner.id)


		async def onboarded_specification(
		catalogue: CatalogueRepositoryPort, partner_id: UUID, app_id: str, version: str
		) -> UUID \| None:
		found = await catalogue.get_resource(partner_id, APPLICATION, _application_key(app_id, version))
		return None if found is None else UUID(str(found["service_specification_id"]))


		def _helm_chart(upload: ArtefactUpload) -> str \| None:
		location = upload.artefact_repo_location
		if (
		upload.artefact_descriptor_type != "HELM"
		or upload.repo_type != "PUBLICREPO"
		or location is None
		or not location.repo_url
		):
		return None
		return str(location.repo_url)


		def _helm_unit(artefact: dict[str, Any], specs: list[ComponentSpec]) -> dict[str, Any]:
		return {
		"ref": artefact["artefact_name"],
		"name": artefact["artefact_name"],
		"runtime_kind": "helm",
		"artifact_ref": artefact["helm_chart"],
		"resource_requirements": _resource_requirements(specs),
		"metadata": {"repo": {"type": "PUBLICREPO"}},
		}


		def _resource_requirements(specs: list[ComponentSpec]) -> dict[str, Any]:
		# several components in one unit (a Helm chart) need their resources added up
		cpu = sum(_millicores(spec.compute_resource_profile.num_cpu) for spec in specs)
		memory = sum(spec.compute_resource_profile.memory for spec in specs)
		return {
		"schema_version": "srm.compute/v1",
		"compute": {"cpu_millicores": cpu, "memory_mb": memory, "storage": []},
		"topology": {
		"min_nodes": 1,
		"min_node_cpu_millicores": cpu,
		"min_node_memory_mb": memory,
		"min_node_cpu_millicores": max(
		_millicores(spec.compute_resource_profile.num_cpu) for spec in specs
		),
		"min_node_memory_mb": max(spec.compute_resource_profile.memory for spec in specs),
		},
		"interfaces": [
		{
		@@ -211,25 +263,12 @@ class InboundOnboardingService:
		if interface.visibility_type == "VISIBILITY_EXTERNAL"
		else "internal",
		}
		for spec in specs
		for interface in spec.exposed_interfaces or []
		],
		"standalone": False,
		},
		"metadata": {"repo": {"type": "PUBLICREPO"}},
		}

		async def _live_inbound(self, partner: PartnerOP, federation_context_id: str) -> None:
		context = await self._contexts.find_inbound(partner.id, federation_context_id)
		if context is None or context.is_terminated():
		raise FederationContextUnknown(partner.id)


		async def onboarded_specification(
		catalogue: CatalogueRepositoryPort, partner_id: UUID, app_id: str, version: str
		) -> UUID \| None:
		found = await catalogue.get_resource(partner_id, APPLICATION, _application_key(app_id, version))
		return None if found is None else UUID(str(found["service_specification_id"]))


		def _application_key(app_id: str, version: str) -> str:
		return f"{app_id}:{version}"

src/federation_manager/application/outbound_deploy.py

+44 −11

Changes for src/federation_manager/application/outbound_deploy.py: 44 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -47,6 +47,7 @@ from federation_manager.domain.onboarding import (
		)
		from federation_manager.domain.ports import (
		CatalogueRepositoryPort,
		ChartInspectorPort,
		EwbiClientPort,
		FederationContextRepositoryPort,
		PartnerRepositoryPort,
		@@ -97,10 +98,12 @@ class OutboundDeploymentService:
		transactions: TransactionRepositoryPort,
		local: LocalOperator,
		*,
		charts: ChartInspectorPort \| None = None,
		clock: Callable[[], datetime] = _utcnow,
		id_factory: Callable[[], UUID] = uuid4,
		) -> None:
		self._partners = partners
		self._charts = charts
		self._contexts = contexts
		self._catalogue = catalogue
		self._srm = srm
		@@ -220,27 +223,46 @@ class OutboundDeploymentService:
		artefact_refs: list[AppComponentSpecRef] = []
		for unit in units:
		unit_ref = str(unit["ref"])
		image = file_id(specification_id, unit_ref)
		helm = unit.get("runtime_kind") == "helm"
		images = (
		await self._chart_images(str(unit["artifact_ref"]))
		if helm
		else [str(unit["artifact_ref"])]
		)
		image_files: list[UUID] = []
		for image in images:
		# one file per image; a container unit keeps its id from before Helm support
		image_file = file_id(specification_id, f"{unit_ref}:{image}" if helm else unit_ref)
		self._check(
		partner,
		await self._ewbi.post_form(
		partner,
		files_path(ctx),
		{
		"fileId": str(image),
		"fileId": str(image_file),
		"appProviderId": provider,
		"fileName": gsma_name(unit_ref),
		"fileName": gsma_name(image.rsplit("/", 1)[-1] if helm else unit_ref),
		"fileVersionInfo": version,
		"fileType": "DOCKER",
		"imgOSType": _UNSPECIFIED_OS,
		"imgInsSetArch": "ISA_X86_64",
		"repoType": "PUBLICREPO",
		"fileRepoLocation": {"repoURL": unit["artifact_ref"]},
		"fileRepoLocation": {"repoURL": image},
		},
		),
		)
		image_files.append(image_file)
		artefact = artefact_id(specification_id, unit_ref)
		component = _component_spec(unit, image)
		component = _component_spec(unit, image_files)
		descriptor: dict[str, object] = (
		{
		"artefactDescriptorType": "HELM",
		"repoType": "PUBLICREPO",
		"artefactRepoLocation": {"repoURL": unit["artifact_ref"]},
		}
		if helm
		else {"artefactDescriptorType": "COMPONENTSPEC"}
		)
		self._check(
		partner,
		await self._ewbi.post_form(
		@@ -252,7 +274,7 @@ class OutboundDeploymentService:
		"artefactName": gsma_name(unit_ref),
		"artefactVersionInfo": version,
		"artefactVirtType": "CONTAINER_TYPE",
		"artefactDescriptorType": "COMPONENTSPEC",
		**descriptor,
		"componentSpec": [component.model_dump(mode="json", by_alias=True)],
		},
		),
		@@ -286,6 +308,11 @@ class OutboundDeploymentService:
		"version": version,
		}

		async def _chart_images(self, chart_url: str) -> list[str]:
		if self._charts is None:
		raise OnboardingNotSupported("Helm charts can't be onboarded at a partner here.")
		return await self._charts.container_images(chart_url)

		@staticmethod
		def _check(partner: PartnerOP, response: EwbiResponse) -> None:
		if response.status_code not in _DONE:
		@@ -297,10 +324,16 @@ def _require_supported(units: list[dict[str, Any]]) -> None:
		raise OnboardingNotSupported("The service specification has no deployment unit.")
		for unit in units:
		repo = (unit.get("metadata") or {}).get("repo") or {}
		if unit.get("runtime_kind") != "container" or not unit.get("artifact_ref"):
		kind, artifact = unit.get("runtime_kind"), str(unit.get("artifact_ref") or "")
		if kind not in ("container", "helm") or not artifact:
		raise OnboardingNotSupported(
		f"Deployment unit {unit.get('ref')} is neither a container image nor a Helm "
		"chart; only those can be deployed on a partner in this release."
		)
		if kind == "helm" and not artifact.startswith(("https://", "http://")):
		raise OnboardingNotSupported(
		f"Deployment unit {unit.get('ref')} is not a container image; only container "
		"units can be deployed on a partner in this release."
		f"Deployment unit {unit.get('ref')} names its Helm chart as {artifact!r}; only a "
		"chart archive at an http(s) URL can be deployed on a partner in this release."
		)
		if repo.get("type", "PUBLICREPO") != "PUBLICREPO":
		raise OnboardingNotSupported(
		@@ -309,12 +342,12 @@ def _require_supported(units: list[dict[str, Any]]) -> None:
		)


		def _component_spec(unit: dict[str, Any], image: UUID) -> ComponentSpec:
		def _component_spec(unit: dict[str, Any], images: list[UUID]) -> ComponentSpec:
		requirements = unit.get("resource_requirements") or {}
		compute = requirements.get("compute") or {}
		return ComponentSpec(
		component_name=gsma_name(str(unit["ref"]), max_length=64),
		images=[image],
		images=images,
		num_of_instances=1,
		restart_policy="RESTART_POLICY_ALWAYS",
		compute_resource_profile=ComputeResourceInfo(

src/federation_manager/dependencies.py

+15 −1

Changes for src/federation_manager/dependencies.py: 15 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -36,6 +36,7 @@ from federation_manager.core.config import get_settings
		from federation_manager.domain.ports import (
		AgreementRepositoryPort,
		CatalogueRepositoryPort,
		ChartInspectorPort,
		DataBusHealthPort,
		DataBusPublisherPort,
		EwbiClientPort,
		@@ -151,6 +152,11 @@ def get_srm_client(request: Request) -> SrmQueryPort:
		return client


		def get_chart_inspector(request: Request) -> ChartInspectorPort \| None:
		inspector: ChartInspectorPort \| None = getattr(request.app.state, "chart_inspector", None)
		return inspector


		def get_inbound_onboarding_service(
		context_repo: Annotated[FederationContextRepositoryPort, Depends(get_federation_context_repo)],
		srm_client: Annotated[SrmQueryPort, Depends(get_srm_client)],
		@@ -201,9 +207,17 @@ def get_outbound_deployment_service(
		ewbi_client: Annotated[EwbiClientPort, Depends(get_ewbi_client)],
		transaction_repo: Annotated[TransactionRepositoryPort, Depends(get_transaction_repo)],
		local: Annotated[LocalOperator, Depends(get_local_operator)],
		charts: Annotated[ChartInspectorPort \| None, Depends(get_chart_inspector)],
		) -> OutboundDeploymentService:
		return OutboundDeploymentService(
		partner_repo, context_repo, catalogue, srm_client, ewbi_client, transaction_repo, local
		partner_repo,
		context_repo,
		catalogue,
		srm_client,
		ewbi_client,
		transaction_repo,
		local,
		charts=charts,
		)