Return Helm workload NodePort endpoints for Release 2.0.0

from typing import Optional

import structlog

import yaml

from kubernetes import client

from kubernetes.client.exceptions import ApiException

from kubernetes.config.config_exception import ConfigException

        )

def _release_services(release: str, namespace: str, connection: KubernetesConnection):

    """Read only NodePort Services declared by this Helm release in its namespace."""

    result = _helm(["get", "manifest", release, "--namespace", namespace], connection)

    if result.returncode != 0:

        raise ValueError("Helm release manifest is unavailable")

    names = set()

    for resource in yaml.safe_load_all(result.stdout):

        if not isinstance(resource, dict) or resource.get("kind") != "Service":

            continue

        metadata = resource.get("metadata") or {}

        if metadata.get("namespace", namespace) != namespace:

            continue

        if (resource.get("spec") or {}).get("type") != "NodePort":

            continue

        name = metadata.get("name")

        if isinstance(name, str) and name:

            names.add(name)

    return [_read_service(name, namespace, connection) for name in sorted(names)]

def _helm_node_port(services, interface) -> int:

    """Match the requested application port/protocol; never pick an ambiguous Service."""

    matches = []

    for service in services:

        if service.spec.type != "NodePort":

            continue

        for port in service.spec.ports or []:

            target = port.target_port if port.target_port is not None else port.port

            if (

                type(target) is int

                and target == interface.port

                and (port.protocol or "TCP").upper() == interface.protocol.upper()

                and port.node_port is not None

            ):

                matches.append(port.node_port)

    if len(matches) != 1:

        raise ValueError("Helm interface has no unique observed NodePort")

    return matches[0]

def _chart_args(unit) -> list:

    if unit.runtime_type == "helm":

        return [unit.artifact_ref]

    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:

        if not host or not unit.resource_requirements:

            return None

        interfaces = unit.resource_requirements.interfaces

        external = [interface for interface in interfaces if interface.visibility == "external"]

        if not external:

            return None

        try:

            if unit.runtime_type == "helm":

                services = _release_services(release, namespace, self._connection)

                ports = [_helm_node_port(services, interface) for interface in external]

            elif unit.runtime_type == "container":

                service = _read_service(release, namespace, self._connection)

                node_ports = {

                    port.name: port.node_port

                    for port in service.spec.ports or []

                    if port.node_port is not None

                }

                ports = [

                    node_ports[f"p{interface.port}-{interface.protocol.lower()}"]

                    for interface in external

                ]

            else:

                return None

            return [

                ServiceEndpoint(

                    interface_id=interface.interface_id,

                    fqdn=host,

                    port=node_ports[f"p{interface.port}-{interface.protocol.lower()}"],

                    port=port,

                )

                for interface in external

                for interface, port in zip(external, ports)

            ]

        except (

            OSError,

            TypeError,

            AttributeError,

            ConfigException,

            DomainExecutionFailed,

            ApiException,

            HTTPError,

            yaml.YAMLError,

        ) as error:

            # Metadata observation must preserve the handle of an installed release.

            log.warning(

"""Helm workload endpoints come from the exact release and its control path."""

from concurrent.futures import ThreadPoolExecutor

from pathlib import Path

from threading import Barrier

from types import SimpleNamespace

import pytest

import yaml

from kubernetes import client

from kubernetes.client.exceptions import ApiException

from openop_tfsdk.contract.adapters.kubernetes import KubernetesCapabilityAdapter

from .test_kubernetes_container import _interface, _request

from .test_kubernetes_endpoint_observation import service_api as _service_api

@pytest.fixture

def service_api(monkeypatch, tmp_path):

    return _service_api.__wrapped__(monkeypatch, tmp_path)

def document(name="web-svc", namespace=None, kind="Service", service_type="NodePort"):

    return {

        "apiVersion": "v1",

        "kind": kind,

        "metadata": {"name": name, **({"namespace": namespace} if namespace else {})},

        "spec": {"type": service_type},

    }

def service(port=80, target=8080, node_port=31234, protocol="TCP", service_type="NodePort"):

    return client.V1Service(

        spec=client.V1ServiceSpec(

            type=service_type,

            ports=[

                client.V1ServicePort(

                    name="http",

                    port=port,

                    target_port=target,

                    node_port=node_port,

                    protocol=protocol,

                )

            ],

        )

    )

@pytest.fixture

def helm(monkeypatch):

    state = SimpleNamespace(documents=[document()], calls=[], status=0, raw=None)

    def command(args, connection):

        state.calls.append((args, connection))

        observation = args[:2] == ["get", "manifest"]

        return SimpleNamespace(

            returncode=state.status if observation else 0,

            stdout=(

                (state.raw if state.raw is not None else yaml.safe_dump_all(state.documents))

                if observation

                else ""

            ),

            stderr="PRIVATE_BACKEND_TEXT",

        )

    monkeypatch.setattr("openop_tfsdk.contract.adapters.kubernetes._helm", command)

    return state

def activate(settings, interfaces=None):

    return KubernetesCapabilityAdapter({**settings, "endpoint_host": "192.0.2.10"}).activate(

        _request(

            "helm",

            artifact_ref="example/chart",

            interfaces=interfaces or [_interface(8080, visibility="external")],

        )

    )

def test_nonstandard_service_name_and_allocated_port_use_the_same_connection(

    kubernetes_settings, helm, service_api

):

    service_api.response = service()

    result = activate(kubernetes_settings)

    assert result.state == "active" and result.external_id == "video-app"

    assert [(e.interface_id, e.fqdn, e.port) for e in result.endpoints] == [

        ("api", "192.0.2.10", 31234)

    ]

    assert helm.calls[1][0] == ["get", "manifest", "video-app", "--namespace", "default"]

    assert helm.calls[0][1] is helm.calls[1][1]

    assert service_api.calls[-1] == ("read", "web-svc", "default", {"_request_timeout": (3, 10)})

    assert service_api.hosts == ["https://test.invalid"]

def test_unrelated_resources_and_other_namespaces_are_not_observed(

    kubernetes_settings, helm, service_api

):

    helm.documents += [

        document("other", "other-namespace"),

        document("secret", kind="Secret"),

        document("private", service_type="ClusterIP"),

    ]

    service_api.response = service()

    assert activate(kubernetes_settings).endpoints

    assert [c[1] for c in service_api.calls if c[0] == "read"] == ["web-svc"]

def test_omitted_target_port_uses_the_service_port(kubernetes_settings, helm, service_api):

    service_api.response = service(port=8080, target=None)

    assert activate(kubernetes_settings).endpoints[0].port == 31234

def test_distinct_interfaces_can_match_distinct_release_services(

    kubernetes_settings, helm, service_api, monkeypatch

):

    helm.documents = [document("web"), document("metrics")]

    observed = {"web": service(), "metrics": service(port=9000, target=9000, node_port=31235)}

    monkeypatch.setattr(

        "openop_tfsdk.contract.adapters.kubernetes._read_service",

        lambda name, namespace, connection: observed[name],

    )

    metrics = _interface(9000, visibility="external")

    metrics.interface_id = "metrics"

    result = activate(kubernetes_settings, [_interface(8080, visibility="external"), metrics])

    assert [(e.interface_id, e.port) for e in result.endpoints] == [

        ("api", 31234),

        ("metrics", 31235),

    ]

@pytest.mark.parametrize(

    "fault",

    [

        "ambiguous",

        "missing",

        "yaml",

        "manifest",

        "forbidden",

        "named-target",

        "wrong-port",

        "wrong-protocol",

        "unallocated",

    ],

)

def test_missing_ambiguous_or_unobservable_metadata_keeps_the_installed_handle(

    kubernetes_settings, helm, service_api, fault

):

    service_api.response = service()

    if fault == "ambiguous":

        helm.documents.append(document("second-public-service"))

    elif fault == "missing":

        helm.documents = []

    elif fault == "yaml":

        helm.raw = "items: [invalid"

    elif fault == "manifest":

        helm.status = 1

    elif fault == "forbidden":

        service_api.response = ApiException(status=403)

    elif fault == "named-target":

        service_api.response = service(target="http")

    elif fault == "wrong-port":

        service_api.response = service(target=9000)

    elif fault == "wrong-protocol":

        service_api.response = service(protocol="UDP")

    elif fault == "unallocated":

        service_api.response = service(node_port=None)

    result = activate(kubernetes_settings)

    assert result.state == "active" and result.external_id == "video-app"

    assert result.endpoints is None

def test_service_type_must_still_be_nodeport_when_observed(kubernetes_settings, helm, service_api):

    service_api.response = service(service_type="LoadBalancer")

    assert activate(kubernetes_settings).endpoints is None

def test_no_host_or_external_interfaces_do_not_run_observation(

    kubernetes_settings, helm, service_api

):

    adapter = KubernetesCapabilityAdapter(kubernetes_settings)

    result = adapter.activate(

        _request("helm", interfaces=[_interface(8080, visibility="external")])

    )

    assert result.endpoints is None

    assert len(helm.calls) == 1 and service_api.calls == []

def test_helm_observation_keeps_two_explicit_cluster_connections_separate(

    kubernetes_settings, helm, service_api, tmp_path, monkeypatch

):

    first = dict(kubernetes_settings)

    data = yaml.safe_load(Path(first["kubeconfig_path"]).read_text())

    data["clusters"][0]["cluster"]["server"] = "https://second.invalid"

    source = tmp_path / "second-config"

    source.write_text(yaml.safe_dump(data))

    second = {**first, "kubeconfig_path": str(source), "control_path_ref": "second"}

    service_api.response = service()

    read = client.CoreV1Api.read_namespaced_service

    barrier = Barrier(2)

    def concurrent_read(self, *args, **kwargs):

        barrier.wait(timeout=5)

        return read(self, *args, **kwargs)

    monkeypatch.setattr(client.CoreV1Api, "read_namespaced_service", concurrent_read)

    monkeypatch.setenv("KUBECONFIG", "/ambient/config")

    monkeypatch.setenv("HELM_KUBECONTEXT", "ambient")

    def run(settings, host):

        bound = KubernetesCapabilityAdapter({**settings, "endpoint_host": host})

        request = _request("helm", interfaces=[_interface(8080, visibility="external")])

        request.target.control_path_ref = settings["control_path_ref"]

        return bound.activate(request)

    with ThreadPoolExecutor(max_workers=2) as pool:

        calls = [pool.submit(run, first, "192.0.2.10"), pool.submit(run, second, "192.0.2.11")]

        endpoints = [call.result(timeout=10).endpoints[0] for call in calls]

    assert [e.fqdn for e in endpoints] == ["192.0.2.10", "192.0.2.11"]

    assert set(service_api.hosts) == {"https://test.invalid", "https://second.invalid"}

reason. Raw response bodies, URLs and exception text are not copied. Error

classes, retryable flags and wrapped HTTP404 classification are unchanged.

SRM's existing bridge can preserve these details without a new contract field.

## Helm workload endpoints

The Kubernetes adapter can report external NodePort endpoints after activating

an application Helm chart. It reads the installed release manifest, then observes

only its NodePort Services in the configured namespace through the same explicit

control-path connection. Service names need not equal the Helm release name.

Declare external interfaces in the deployment unit's resource requirements and

configure the control path's reachable endpoint_host. An interface matches the

observed Service's numeric targetPort and protocol; omitted targetPort defaults

to the Service port. Each interface must have exactly one matching allocated

NodePort across the release. The returned port is the allocated nodePort, not

the application's listening port. Service names are read from the release; ambiguous matches are not selected.

Missing, ambiguous or unavailable observation preserves ACTIVE and the installed

release handle, with endpoints absent and a sanitized warning. Named targetPort,

Ingress, LoadBalancer and automatic endpoint refresh on update/runtime reads are

outside this bounded observation path. A DNS endpoint_host also needs the

consumer's address-resolution support before it can serve an IPv4-only TI flow.