DO NOT MERGE - Testing integration of #415, #417 and #418 on staging

cloud_pipelines_backend/launchers/kubernetes_launchers.py

@@ -9,8 +9,9 @@
 import os
 import pathlib
 import threading
+import time
 import typing
-from typing import Any, Optional
+from typing import Any, Final, Optional
 
 import kubernetes.client.exceptions
 from kubernetes import client as k8s_client_lib
@@ -796,6 +797,43 @@ class KubernetesWithGcsFuseContainerLauncher(GoogleKubernetesEngineLauncher):
     pass
 
 
+# A failed log read means the node is gone, so retrying only blocks the caller.
+_LOG_READ_MAX_RETRIES: Final[int] = 1
+_LOG_READ_WAIT_SECONDS: Final[float] = 1.0
+_LOG_READ_MAX_TOTAL_SECONDS: Final[float | None] = 10.0
+
+
+def _read_log_within_budget(
+    read_log: typing.Callable[[], _T],
+    *,
+    max_retries: int = _LOG_READ_MAX_RETRIES,
+    wait_seconds: float = _LOG_READ_WAIT_SECONDS,
+    max_total_seconds: float | None = _LOG_READ_MAX_TOTAL_SECONDS,
+) -> _T:
+    """Calls `read_log`, retrying within a budget. Raises if the budget runs out.
+
+    A retry never starts once the deadline has passed. The final attempt can
+    still overrun it by one request timeout, which `request_timeout` bounds.
+    """
+    start_seconds = time.monotonic()
+    for attempt in range(max_retries - 1):
+        try:
+            return read_log()
+        except Exception:
+            _logger.exception(
+                f"Exception reading a container log on attempt {attempt + 1} of "
+                f"{max_retries}. Retrying."
+            )
+            if (
+                max_total_seconds is not None
+                and time.monotonic() - start_seconds + wait_seconds > max_total_seconds
+            ):
+                raise
+            time.sleep(wait_seconds)
+    # The last attempt is not retried, so its exception is the caller's.
+    return read_log()
+
+
 class LaunchedKubernetesContainer(interfaces.LaunchedContainer):
     def __init__(
         self,
@@ -931,7 +969,9 @@ def launcher_error_message(self) -> str | None:
         return launcher_error_message
 
     def to_dict(self) -> dict[str, Any]:
-        pod_dict = _serialize_kubernetes_object_to_compact_dict(self._debug_pod)
+        pod_dict = _serialize_kubernetes_object_to_compact_dict_and_censor_it(
+            obj=self._debug_pod
+        )
         result = dict(
             kubernetes=dict(
                 # launched_container_class_name=self.__class__.__name__,
@@ -1009,8 +1049,14 @@ def get_log(self) -> str:
             response.release_conn()
 
     def upload_log(self):
+        try:
+            log = _read_log_within_budget(self.get_log)
+        except Exception:
+            # Not raising: the caller retries `upload_log`, so raising costs it
+            # this budget five times over. Writing nothing keeps an earlier log.
+            _logger.exception(f"Giving up on the log of Pod {self._pod_name}.")
+            return
         launcher = self._get_launcher()
-        log = self.get_log()
         uri_writer = launcher._storage_provider.make_uri(self._log_uri).get_writer()
         uri_writer.upload_from_text(log)
 
@@ -1577,12 +1623,16 @@ def launcher_error_message(self) -> str | None:
     SERIALIZATION_ROOT_KEY = "kubernetes_job"
 
     def to_dict(self) -> dict[str, Any]:
-        job_dict = _serialize_kubernetes_object_to_compact_dict(self._debug_job)
+        job_dict = _serialize_kubernetes_object_to_compact_dict_and_censor_it(
+            obj=self._debug_job
+        )
         pod_dicts = None
         if self._debug_pods is not None:
             pod_dicts = {
                 pod_name: (
-                    _serialize_kubernetes_object_to_compact_dict(pod) if pod else None
+                    _serialize_kubernetes_object_to_compact_dict_and_censor_it(obj=pod)
+                    if pod
+                    else None
                 )
                 for pod_name, pod in self._debug_pods.items()
             }
@@ -1749,6 +1799,16 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None:
                     f"Pod {pod_name} no longer exists; its logs are unrecoverable."
                 )
                 return None
+            if _is_kubelet_unreachable_error(ex):
+                # The API server could not reach the kubelet of the Pod's node. The
+                # cluster-autoscaler removes a node soon after the last Pod on it
+                # completes, and the logs lived on that node, so they are gone. Retrying
+                # cannot recover them and only blocks the caller.
+                _logger.warning(
+                    f"Kubelet of Pod {pod_name} is unreachable; its logs are "
+                    f"unrecoverable. {_extract_api_exception_body(ex)}"
+                )
+                return None
             raise
 
     def _get_all_attempt_pods(self) -> dict[str, list[k8s_client_lib.V1Pod]]:
@@ -1859,7 +1919,12 @@ def get_log(self) -> str:
         return merged_log
 
     def upload_log(self):
-        all_logs = self._get_all_logs()
+        try:
+            all_logs = _read_log_within_budget(self._get_all_logs)
+        except Exception:
+            # See the note on `LaunchedKubernetesContainer.upload_log`.
+            _logger.exception(f"Giving up on the logs of Job {self._job_name}.")
+            return
         merged_log = self._merge_logs(all_logs)
 
         # Uploading the merged log
@@ -2145,6 +2210,15 @@ def _serialize_kubernetes_object_to_compact_dict(obj) -> dict[str, Any]:
     return obj_dict
 
 
+def _serialize_kubernetes_object_to_compact_dict_and_censor_it(
+    *, obj: k8s_client_lib.V1Job | k8s_client_lib.V1Pod
+) -> dict[str, Any]:
+    """Serializes a Kubernetes object without the command lines, which can hold a secret that the container consumes as a value."""
+    obj_dict = _serialize_kubernetes_object_to_compact_dict(obj)
+    _remove_command_lines(obj_dict)
+    return obj_dict
+
+
 def windows_path_to_docker_path(path: str) -> str:
     if os.name != "nt":
         return path
@@ -2191,6 +2265,57 @@ def _remove_keys_with_none_values(d: dict):
             _remove_keys_with_none_values(v)
 
 
+def _remove_command_lines(d: dict):
+    for k, v in list(d.items()):
+        if k in ("command", "args"):
+            del d[k]
+        elif isinstance(v, dict):
+            _remove_command_lines(v)
+        elif isinstance(v, list):
+            for item in v:
+                if isinstance(item, dict):
+                    _remove_command_lines(item)
+
+
+def _extract_api_exception_body(
+    exception: kubernetes.client.exceptions.ApiException,
+) -> str:
+    """The response body as text, whichever way the installed client stored it.
+
+    Older Kubernetes clients assign the raw `bytes`; newer ones decode to `str` in
+    `ApiException.__init__` and leave the body `None` if decoding fails.
+    """
+    body = exception.body
+    if isinstance(body, bytes):
+        return body.decode("utf-8", errors="replace")
+    return body or ""
+
+
+def _is_kubelet_unreachable_error(
+    exception: kubernetes.client.exceptions.ApiException,
+) -> bool:
+    """Whether the API server failed to dial the kubelet while proxying a log read.
+
+    Raised as `ServiceException`, a subclass of `ApiException`. An observed response
+    body, with cluster identifiers replaced:
+
+        {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure",
+         "message":"Get \\"https://<node-ip>:10250/containerLogs/<namespace>/<pod>/main?timestamps=true\\":
+                     dial tcp <node-ip>:10250: i/o timeout","code":500}
+
+    The wording comes from the API server's log proxy and is not a documented
+    contract, so treat a non-match as "some other error" rather than as proof the
+    kubelet was reachable. Callers that must bound their wall time regardless set
+    the log-upload retry budget on the launched container instead.
+    """
+    body = _extract_api_exception_body(exception)
+    return (
+        exception.status == http.HTTPStatus.INTERNAL_SERVER_ERROR
+        and "/containerLogs/" in body
+        and "dial tcp" in body
+    )
+
+
 def _launcher_error_from_api_exception(
     exception: kubernetes.client.exceptions.ApiException,
     *,

tests/launchers/kubernetes/test_kubernetes_command_line_removal.py

@@ -0,0 +1,170 @@
+"""Tests that serialized (stored) Kubernetes objects carry no container command lines.
+
+A container that consumes a secret as a value receives that value on its command line, so a
+stored command line can hold a secret in plaintext.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Final
+
+from kubernetes import client as k8s_client_lib
+
+from cloud_pipelines_backend.launchers import kubernetes_launchers
+
+_SECRET_VALUE: Final[str] = "plaintext-secret-value"
+
+
+def _make_container() -> k8s_client_lib.V1Container:
+    return k8s_client_lib.V1Container(
+        name="main",
+        image="alpine",
+        command=["python3", "-c", "print(1)"],
+        args=["--token", _SECRET_VALUE],
+    )
+
+
+def _make_pod() -> k8s_client_lib.V1Pod:
+    return k8s_client_lib.V1Pod(
+        metadata=k8s_client_lib.V1ObjectMeta(name="test-pod", namespace="test"),
+        spec=k8s_client_lib.V1PodSpec(containers=[_make_container()]),
+    )
+
+
+def _make_job() -> k8s_client_lib.V1Job:
+    return k8s_client_lib.V1Job(
+        metadata=k8s_client_lib.V1ObjectMeta(name="test-job", namespace="test"),
+        spec=k8s_client_lib.V1JobSpec(
+            template=k8s_client_lib.V1PodTemplateSpec(
+                spec=k8s_client_lib.V1PodSpec(containers=[_make_container()]),
+            ),
+        ),
+    )
+
+
+def test_serialized_pod_has_no_command_line() -> None:
+    pod_dict = (
+        kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it(
+            obj=_make_pod()
+        )
+    )
+
+    container = pod_dict["spec"]["containers"][0]
+    assert "command" not in container
+    assert "args" not in container
+    assert container["image"] == "alpine"
+    assert _SECRET_VALUE not in json.dumps(pod_dict)
+
+
+def test_serialized_job_has_no_command_line_on_its_pod_template() -> None:
+    job_dict = (
+        kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it(
+            obj=_make_job()
+        )
+    )
+
+    container = job_dict["spec"]["template"]["spec"]["containers"][0]
+    assert "command" not in container
+    assert "args" not in container
+    assert _SECRET_VALUE not in json.dumps(job_dict)
+
+
+def test_serialized_pod_has_no_command_line_anywhere() -> None:
+    """The removal is not selective: every command line goes, whatever holds it.
+
+    An init container, a sidecar and a probe are all stored, and any of them can be the one
+    carrying a secret, so none is left behind.
+    """
+    pod = k8s_client_lib.V1Pod(
+        metadata=k8s_client_lib.V1ObjectMeta(name="test-pod", namespace="test"),
+        spec=k8s_client_lib.V1PodSpec(
+            init_containers=[
+                k8s_client_lib.V1Container(
+                    name="init",
+                    image="alpine",
+                    command=["sh", "-c", _SECRET_VALUE],
+                ),
+            ],
+            containers=[
+                _make_container(),
+                k8s_client_lib.V1Container(
+                    name="sidecar",
+                    image="alpine",
+                    args=["--token", _SECRET_VALUE],
+                    readiness_probe=k8s_client_lib.V1Probe(
+                        _exec=k8s_client_lib.V1ExecAction(
+                            command=["check", _SECRET_VALUE],
+                        ),
+                    ),
+                ),
+            ],
+        ),
+    )
+
+    pod_dict = (
+        kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it(
+            obj=pod
+        )
+    )
+
+    assert "command" not in pod_dict["spec"]["initContainers"][0]
+    sidecar = pod_dict["spec"]["containers"][1]
+    assert "args" not in sidecar
+    assert "command" not in sidecar["readinessProbe"]["exec"]
+    assert _SECRET_VALUE not in json.dumps(pod_dict)
+
+
+def test_launched_container_to_dict_has_no_command_line() -> None:
+    """The Pod launcher's stored payload: what the orchestrator writes to `launcher_data`."""
+    launched = kubernetes_launchers.LaunchedKubernetesContainer(
+        pod_name="test-pod",
+        namespace="test",
+        output_uris={},
+        log_uri="gs://bucket/log",
+        debug_pod=_make_pod(),
+    )
+
+    launcher_data = launched.to_dict()
+
+    stored_pod = launcher_data["kubernetes"]["debug_pod"]
+    assert "command" not in stored_pod["spec"]["containers"][0]
+    assert "args" not in stored_pod["spec"]["containers"][0]
+    assert _SECRET_VALUE not in json.dumps(launcher_data)
+
+
+def test_launched_job_to_dict_has_no_command_line_in_any_copy() -> None:
+    """The Job launcher stores the Pod template and one copy per Pod; all of them are cleaned."""
+    launched = kubernetes_launchers.LaunchedKubernetesJob(
+        job_name="test-job",
+        namespace="test",
+        output_uris={},
+        log_uri="gs://bucket/log",
+        debug_job=_make_job(),
+        debug_pods={"test-job-0-aaa": _make_pod(), "test-job-0-bbb": _make_pod()},
+    )
+
+    launcher_data = launched.to_dict()
+
+    stored = launcher_data["kubernetes_job"]
+    template_container = stored["debug_job"]["spec"]["template"]["spec"]["containers"][
+        0
+    ]
+    assert "command" not in template_container
+    assert "args" not in template_container
+    assert len(stored["debug_pods"]) == 2
+    for stored_pod in stored["debug_pods"].values():
+        assert "command" not in stored_pod["spec"]["containers"][0]
+        assert "args" not in stored_pod["spec"]["containers"][0]
+    assert _SECRET_VALUE not in json.dumps(launcher_data)
+
+
+def test_launched_pod_keeps_its_command_line() -> None:
+    """The object sent to the cluster must be untouched: it is how the container is started."""
+    pod = _make_pod()
+
+    kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it(
+        obj=pod
+    )
+
+    assert pod.spec.containers[0].args == ["--token", _SECRET_VALUE]