fix: observe container endpoints with the existing Kubernetes client (90cd985a) · Commits · OOP / code / transformation-function-sdk

src/openop_tfsdk/contract/adapters/kubernetes.py

+48 −13

Changes for src/openop_tfsdk/contract/adapters/kubernetes.py: 48 added lines, 13 removed lines.
Original line number	Diff line number	Diff line
		@@ -8,12 +8,17 @@ 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.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.errors import (
		@@ -56,6 +61,26 @@ def _helm(args: list):
		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:
		return client.CoreV1Api(api).read_namespaced_service(
		release, namespace, _request_timeout=(3, 10)
		)


		def _chart_args(unit) -> list:
		if unit.runtime_type == "helm":
		return [unit.artifact_ref]
		@@ -153,7 +178,7 @@ class KubernetesCapabilityAdapter:
		endpoints=self._endpoints(unit, release, namespace),
		)

		def _endpoints(self, unit, release: str, namespace: str) -> Optional[list]:
		def _endpoints(self, unit, release: str, namespace: str) -> Optional[list[ServiceEndpoint]]:
		# Kubernetes picks the node ports at install; the host reachable from outside is configured.
		host = self._infrastructure.get("endpoint_host")
		if not host or unit.runtime_type != "container" or not unit.resource_requirements:
		@@ -163,19 +188,11 @@ class KubernetesCapabilityAdapter:
		if not external:
		return None
		try:
		result = subprocess.run(
		["kubectl", "get", "service", release, "--namespace", namespace, "-o", "json"],
		capture_output=True,
		text=True,
		)
		except FileNotFoundError:
		log.warning("endpoints_unavailable", reason="kubectl_not_installed")
		return None
		if result.returncode != 0:
		log.warning("endpoints_unavailable", reason="kubectl_failed")
		return None
		service = _read_service(release, namespace)
		node_ports = {
		port["name"]: port["nodePort"] for port in json.loads(result.stdout)["spec"]["ports"]
		port.name: port.node_port
		for port in service.spec.ports or []
		if port.node_port is not None
		}
		return [
		ServiceEndpoint(
		@@ -185,6 +202,24 @@ class KubernetesCapabilityAdapter:
		)
		for interface in external
		]
		except (
		OSError,
		ValueError,
		KeyError,
		TypeError,
		AttributeError,
		ConfigException,
		ApiException,
		HTTPError,
		) as error:
		# Metadata observation must preserve the handle of an installed release.
		log.warning(
		"endpoints_unavailable",
		reason="service_observation_failed",
		error_type=type(error).__name__,
		status=getattr(error, "status", None),
		)
		return None

		def read_runtime_state(self, request: CapabilityRuntimeStateRequest) -> CapabilityRuntimeState:
		if not request.external_id:

tests/contract/test_kubernetes_container.py

+25 −21

Changes for tests/contract/test_kubernetes_container.py: 25 added lines, 21 removed lines.
Original line number	Diff line number	Diff line
		@@ -5,6 +5,8 @@ import os
		from types import SimpleNamespace

		import pytest
		from kubernetes import client
		from kubernetes.client.exceptions import ApiException

		from openop_tfsdk.contract.adapters.kubernetes import (
		CONTAINER_CHART,
		@@ -114,38 +116,36 @@ def test_mixed_visibility_is_rejected_before_exposing_internal_ports(helm_calls,
		assert helm_calls == []


		def _cluster(monkeypatch, kubectl):
		def _cluster(monkeypatch, observation):
		calls = []

		def run(args, **kwargs):
		calls.append(args)
		if args[0] == "helm":
		assert args[0] == "helm"
		return SimpleNamespace(returncode=0, stdout="", stderr="")
		return kubectl()

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

		monkeypatch.setattr("openop_tfsdk.contract.adapters.kubernetes.subprocess.run", run)
		monkeypatch.setattr("openop_tfsdk.contract.adapters.kubernetes._read_service", read)
		return calls


		def _service():
		ports = [{"name": "p80-tcp", "port": 80, "nodePort": 31234}]
		return SimpleNamespace(returncode=0, stdout=json.dumps({"spec": {"ports": ports}}), stderr="")
		return client.V1Service(
		spec=client.V1ServiceSpec(
		ports=[client.V1ServicePort(name="p80-tcp", port=80, node_port=31234)]
		)
		)


		def test_container_reports_its_node_ports_as_endpoints(monkeypatch):
		calls = _cluster(monkeypatch, _service)
		request = _request("container", interfaces=[_interface(80, visibility="external")])
		result = KubernetesCapabilityAdapter(HOST).activate(request)
		assert calls[1] == [
		"kubectl",
		"get",
		"service",
		"video-app",
		"--namespace",
		"default",
		"-o",
		"json",
		]
		assert calls[1] == ["service", "video-app", "default"]
		assert result.endpoints == [ServiceEndpoint(interface_id="api", fqdn="192.0.2.10", port=31234)]


		@@ -156,16 +156,20 @@ def test_no_endpoints_without_a_configured_host(monkeypatch):
		assert len(calls) == 1


		def _missing_kubectl():
		raise FileNotFoundError
		def _unavailable_service():
		raise OSError("connection failed")


		def _forbidden_service():
		raise ApiException(status=403)


		@pytest.mark.parametrize(
		"kubectl",
		[lambda: SimpleNamespace(returncode=1, stdout="", stderr="forbidden"), _missing_kubectl],
		"observation",
		[_forbidden_service, _unavailable_service],
		)
		def test_unreadable_service_still_activates(monkeypatch, kubectl):
		_cluster(monkeypatch, kubectl)
		def test_unreadable_service_still_activates(monkeypatch, observation):
		_cluster(monkeypatch, observation)
		request = _request("container", interfaces=[_interface(80, visibility="external")])
		result = KubernetesCapabilityAdapter(HOST).activate(request)
		assert result.state == LifecycleState.ACTIVE

tests/contract/test_kubernetes_endpoint_observation.py

0 → 100644

+108 −0

Changes for tests/contract/test_kubernetes_endpoint_observation.py: 108 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Endpoint observation must work in the SRM image without a kubectl binary."""

		from types import SimpleNamespace

		import pytest
		from kubernetes import client, config
		from kubernetes.client.exceptions import ApiException
		from urllib3.exceptions import HTTPError

		from openop_tfsdk.contract.adapters.kubernetes import KubernetesCapabilityAdapter

		from .test_kubernetes_container import _interface, _request


		@pytest.fixture
		def service_api(monkeypatch, tmp_path):
		calls = []

		def command(args, **kwargs):
		if args[0] != "helm":
		raise FileNotFoundError("The SRM runtime has no kubectl")
		return SimpleNamespace(returncode=0, stdout="", stderr="")

		monkeypatch.setattr("openop_tfsdk.contract.adapters.kubernetes.subprocess.run", command)
		monkeypatch.delenv("KUBECONFIG", raising=False)
		monkeypatch.setattr(
		config.kube_config, "KUBE_CONFIG_DEFAULT_LOCATION", str(tmp_path / "absent")
		)

		def load(**kwargs):
		calls.append(("incluster", kwargs["client_configuration"]))
		kwargs["client_configuration"].host = "https://lab-cluster.invalid"

		monkeypatch.setattr(config, "load_incluster_config", load)
		service = client.V1Service(
		spec=client.V1ServiceSpec(
		ports=[
		client.V1ServicePort(name="p80-tcp", port=80, node_port=31234),
		client.V1ServicePort(name="p9000-udp", port=9000, node_port=31235),
		]
		)
		)

		def read(self, name, namespace, **kwargs):
		calls.append(("read", name, namespace, kwargs))
		if isinstance(api.response, Exception):
		raise api.response
		return api.response

		api = SimpleNamespace(response=service, calls=calls)
		monkeypatch.setattr(client.CoreV1Api, "read_namespaced_service", read)
		return api


		def test_srm_without_kubectl_reports_each_interface_with_a_bounded_api_read(service_api):
		default_host = client.Configuration.get_default_copy().host
		tcp, udp = _interface(80, visibility="external"), _interface(9000, "udp", "external")
		udp.interface_id = "metrics"
		result = KubernetesCapabilityAdapter(
		{"endpoint_host": "192.0.2.10", "namespace": "app-lab"}
		).activate(_request("container", interfaces=[tcp, udp]))
		assert [(e.interface_id, e.fqdn, e.port) for e in result.endpoints] == [
		("api", "192.0.2.10", 31234),
		("metrics", "192.0.2.10", 31235),
		]
		assert service_api.calls[-1] == ("read", "video-app", "app-lab", {"_request_timeout": (3, 10)})
		assert client.Configuration.get_default_copy().host == default_host


		@pytest.mark.parametrize(
		"response",
		[
		ApiException(status=403),
		HTTPError("timeout"),
		client.V1Service(spec=client.V1ServiceSpec(ports=[])),
		client.V1Service(
		spec=client.V1ServiceSpec(ports=[client.V1ServicePort(name="p80-tcp", port=80)])
		),
		],
		)
		def test_failed_or_incomplete_observation_preserves_the_created_release(service_api, response):
		service_api.response = response
		result = KubernetesCapabilityAdapter({"endpoint_host": "192.0.2.10"}).activate(
		_request("container", interfaces=[_interface(80, visibility="external")])
		)
		assert result.state == "active"
		assert result.external_id == "video-app"
		assert result.endpoints is None
		assert any(call[0] == "read" for call in service_api.calls)


		def test_explicit_kubeconfig_uses_an_isolated_client_and_helm_context(
		service_api, monkeypatch, tmp_path
		):
		path = str(tmp_path / "selected-kubeconfig")
		monkeypatch.setenv("KUBECONFIG", path)
		monkeypatch.setenv("HELM_KUBECONTEXT", "selected-context")
		loaded = []
		monkeypatch.setattr(config, "load_kube_config", lambda **kwargs: loaded.append(kwargs))
		result = KubernetesCapabilityAdapter({"endpoint_host": "192.0.2.10"}).activate(
		_request("container", interfaces=[_interface(80, visibility="external")])
		)
		assert result.endpoints
		assert len(loaded) == 1
		assert loaded[0]["config_file"] == path
		assert loaded[0]["context"] == "selected-context"
		assert isinstance(loaded[0]["client_configuration"], client.Configuration)
		assert not any(call[0] == "incluster" for call in service_api.calls)