fix: bind Kubernetes operations to each control path (11a35fe8) · Commits · OOP / code / transformation-function-sdk

README.md

+32 −0

Changes for README.md: 32 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -186,3 +186,35 @@ NEF ->> 5GS: QoS session creation
		## License

		Apache 2.0 License – see [`LICENSE`](LICENSE) file for details.
		# Kubernetes control-path connections

		Supply the selected control path's connection in the infrastructure/config
		dictionary passed to `build_adapter("kubernetes", config)`. For example:

		```python
		adapter = build_adapter("kubernetes", {
		"kubeconfig_path": "/etc/openop/kubeconfigs/athens.yaml",
		"kube_context": "athens",
		"namespace": "applications",
		"endpoint_host": "192.0.2.10",
		})
		```

		Helm lifecycle operations and Kubernetes Service reads use that same connection.
		`api_server` and `tls_verify`, when supplied, override the kubeconfig consistently
		for both clients. Construction performs no file or network access; an invalid
		explicit kubeconfig fails before Helm contacts a backend and never selects
		another cluster. Kubeconfigs are loaded without persisting credential updates.

		Use `in_cluster: True` for the current Pod's service account, without a
		kubeconfig path or context. Helm receives that identity through its child
		environment; no token is placed in command arguments. Operator tooling must
		resolve any `kubeconfig_ref` and provide the mounted `kubeconfig_path`; the SDK
		does not implement a `secret://` resolver. Configuration and credentials stay
		outside topology models and responses.

		For compatibility, omitted connection settings capture KUBECONFIG,
		HELM_KUBECONTEXT or the in-cluster identity environment when constructing the
		adapter. Later environment changes do not redirect it. Explicit per-path
		configuration is required to distinguish clusters. The adapter does not mutate
		the process environment or the Kubernetes client's global default configuration.

src/openop_tfsdk/contract/adapters/kubernetes.py

+25 −25

Changes for src/openop_tfsdk/contract/adapters/kubernetes.py: 25 added lines, 25 removed lines.
Original line number	Diff line number	Diff line
		@@ -8,19 +8,19 @@ the bootstrap YAML.
		"""

		import json
		import os
		import re
		import subprocess
		from pathlib import Path
		from typing import Optional

		import structlog
		from kubernetes import client, config
		from kubernetes import client
		from kubernetes.client.exceptions import ApiException
		from kubernetes.config.config_exception import ConfigException
		from urllib3.exceptions import HTTPError

		from openop_tfsdk._diagnostics import helm_diagnostics
		from openop_tfsdk.contract.adapters.kubernetes_connection import KubernetesConnection
		from openop_tfsdk.contract.errors import (
		DomainExecutionFailed,
		ResourceNotFound,
		@@ -53,29 +53,24 @@ RUNTIME_KINDS = ["helm", "container"]
		CONTAINER_CHART = str(Path(__file__).parent / "charts" / "container")


		def _helm(args: list):
		def _helm(args: list, connection: KubernetesConnection):
		try:
		return subprocess.run(["helm", *args], capture_output=True, text=True)
		options, environment = connection.helm_options()
		return subprocess.run(
		["helm", args, options], env=environment, capture_output=True, text=True
		)
		except ConfigException as error:
		raise DomainExecutionFailed(
		"Kubernetes control-path configuration could not be loaded"
		) from error
		except FileNotFoundError as error:
		log.error("helm_not_installed", error_code=DomainExecutionFailed.error_code)
		raise DomainExecutionFailed("helm is not installed") from error


		def _read_service(release: str, namespace: str):
		"""Use the ordinary Helm kubeconfig or in-cluster identity without global mutation."""
		configuration = client.Configuration()
		kubeconfig = os.environ.get("KUBECONFIG")
		default = config.kube_config.KUBE_CONFIG_DEFAULT_LOCATION
		if kubeconfig or Path(default).expanduser().is_file():
		config.load_kube_config(
		config_file=kubeconfig or default,
		context=os.environ.get("HELM_KUBECONTEXT"),
		client_configuration=configuration,
		)
		else:
		config.load_incluster_config(client_configuration=configuration)
		configuration.retries = 0
		with client.ApiClient(configuration) as api:
		def _read_service(release: str, namespace: str, connection: KubernetesConnection):
		"""Observe the Service through the same control-path connection as Helm."""
		with client.ApiClient(connection.configuration()) as api:
		return client.CoreV1Api(api).read_namespaced_service(
		release, namespace, _request_timeout=(3, 10)
		)
		@@ -134,7 +129,8 @@ class KubernetesCapabilityAdapter:
		adapter_type = "kubernetes"

		def __init__(self, infrastructure: Optional[dict] = None):
		self._infrastructure = infrastructure or {}
		self._infrastructure = dict(infrastructure or {})
		self._connection = KubernetesConnection(self._infrastructure)

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		@@ -159,7 +155,8 @@ class KubernetesCapabilityAdapter:
		"--create-namespace",
		"--atomic",
		"--wait",
		]
		],
		self._connection,
		)

		if result.returncode != 0:
		@@ -188,7 +185,7 @@ class KubernetesCapabilityAdapter:
		if not external:
		return None
		try:
		service = _read_service(release, namespace)
		service = _read_service(release, namespace, self._connection)
		node_ports = {
		port.name: port.node_port
		for port in service.spec.ports or []
		@@ -235,7 +232,8 @@ class KubernetesCapabilityAdapter:
		namespace,
		"--output",
		"json",
		]
		],
		self._connection,
		)

		if result.returncode != 0:
		@@ -278,7 +276,8 @@ class KubernetesCapabilityAdapter:
		"--namespace",
		namespace,
		"--wait",
		]
		],
		self._connection,
		)

		if result.returncode != 0:
		@@ -313,7 +312,8 @@ class KubernetesCapabilityAdapter:
		namespace,
		"--atomic",
		"--wait",
		]
		],
		self._connection,
		)

		if result.returncode != 0:

src/openop_tfsdk/contract/adapters/kubernetes_connection.py

0 → 100644

+111 −0

Changes for src/openop_tfsdk/contract/adapters/kubernetes_connection.py: 111 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Connection settings belong to one adapter, shared by Helm and API reads."""

		import os
		from pathlib import Path

		import yaml
		from kubernetes import client, config
		from kubernetes.config.config_exception import ConfigException

		from openop_tfsdk.contract.errors import ValidationError


		class KubernetesConnection:
		def __init__(self, settings: dict):
		# Compatibility defaults are captured once; execution never reads os.environ.
		self._environment = dict(os.environ)
		path = settings.get("kubeconfig_path")
		context = settings.get("kube_context")
		if any(
		value is not None and (not isinstance(value, str) or not value.strip())
		for value in (path, context)
		):
		raise ValidationError(
		"Kubernetes configuration path and context must be nonempty strings"
		)
		if settings.get("kubeconfig_ref") and not path:
		raise ValidationError(
		"Resolve kubeconfig_ref to kubeconfig_path before building the adapter"
		)
		self.in_cluster = settings.get(
		"in_cluster",
		not (path or context or self._environment.get("KUBECONFIG"))
		and bool(self._environment.get("KUBERNETES_SERVICE_HOST")),
		)
		if not isinstance(self.in_cluster, bool) or (self.in_cluster and (path or context)):
		raise ValidationError("Choose in_cluster or kubeconfig_path/kube_context")
		self._explicit = bool(path or context or settings.get("api_server"))
		self.path = (
		None
		if self.in_cluster
		else (path or self._environment.get("KUBECONFIG") or str(Path.home() / ".kube/config"))
		)
		self.context = (
		None
		if self.in_cluster
		else (context if path else (context or self._environment.get("HELM_KUBECONTEXT")))
		)
		self.server = settings.get("api_server")
		if self.server is not None and (
		not isinstance(self.server, str) or not self.server.startswith(("https://", "http://"))
		):
		raise ValidationError("api_server must be an HTTP or HTTPS URL")
		self.tls_verify = settings.get("tls_verify")
		if self.tls_verify is not None and not isinstance(self.tls_verify, bool):
		raise ValidationError("tls_verify must be a boolean")

		def configuration(self):
		configuration = client.Configuration()
		try:
		if self.in_cluster:
		# The pinned client's loader accepts an explicit environment snapshot.
		config.incluster_config.InClusterConfigLoader(
		token_filename=config.incluster_config.SERVICE_TOKEN_FILENAME,
		cert_filename=config.incluster_config.SERVICE_CERT_FILENAME,
		environ=self._environment,
		).load_and_set(configuration)
		else:
		config.load_kube_config(
		config_file=self.path,
		context=self.context,
		client_configuration=configuration,
		persist_config=False,
		)
		except (OSError, ValueError, yaml.YAMLError) as error:
		raise ConfigException(
		"Kubernetes control-path configuration could not be loaded"
		) from error
		if self.server:
		configuration.host = self.server
		if self.tls_verify is not None:
		configuration.verify_ssl = self.tls_verify
		configuration.retries = 0
		return configuration

		def helm_options(self):
		environment = {
		key: value
		for key, value in self._environment.items()
		if key != "KUBECONFIG" and not key.startswith("HELM_KUBE")
		}
		args = ["--kubeconfig", self.path or os.devnull]
		if self.path and os.pathsep in self.path:
		# Helm merges KUBECONFIG lists only through its child environment.
		environment["KUBECONFIG"] = self.path
		args = []
		if self.context:
		args += ["--kube-context", self.context]
		# Validate explicit credentials locally before Helm can contact a backend.
		if self.in_cluster or self._explicit:
		configuration = self.configuration()
		if self.in_cluster:
		environment.update(
		HELM_KUBEAPISERVER=configuration.host,
		HELM_KUBECAFILE=configuration.ssl_ca_cert,
		HELM_KUBETOKEN=configuration.api_key["authorization"].split(" ", 1)[-1],
		)
		if self.server:
		args += ["--kube-apiserver", self.server]
		if self.tls_verify is not None:
		args += ["--kube-insecure-skip-tls-verify=" + str(not self.tls_verify).lower()]
		return args, environment

tests/contract/bootstrap_example.yaml

+3 −0

Changes for tests/contract/bootstrap_example.yaml: 3 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -24,6 +24,9 @@ control_paths:
		infrastructure:
		api_server: https://k8s-athens-edge.example.local:6443
		kubeconfig_ref: secret://openop/k8s-athens-edge/kubeconfig
		# Ops resolves the reference and mounts this file before adapter execution.
		kubeconfig_path: /etc/openop/kubeconfigs/athens-edge.yaml
		kube_context: athens-edge
		namespace: openop-services
		default_storage_class: fast-edge-ssd
		registry_mirror: registry.athens-edge.example.local

tests/contract/test_kubernetes_container.py

+1 −1

Changes for tests/contract/test_kubernetes_container.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -124,7 +124,7 @@ def _cluster(monkeypatch, observation):
		assert args[0] == "helm"
		return SimpleNamespace(returncode=0, stdout="", stderr="")

		def read(release, namespace):
		def read(release, namespace, connection):
		calls.append(["service", release, namespace])
		return observation()