kube: Migrate Pod parsing to from_json

... and away from the Python kubernetes client.

This diff looks big, but about half of it is tests, and most of the rest
is straightforward: creating TypedDicts for the responses, pulling the
relevant parsing functions out of transform.py and into the respective
from_json modules, and updating accesses to pull data from the dict
instead of from the kubernetes library models. Then updating
api_server.py and controllers.py to make use of the new structures.

The goal is to backport this to 2.4, because it's needed to be able to
pull out the new pod-level resources field eventually, since the version
of the Kubernetes library in 2.4 doesn't provide this field.

The new test cases were written by Claude though I have given them a
once-over.

Change-Id: I1cb1539090e2cbe9f018a247329a9c95a64a9a4d

Author: relrod

cmk/plugins/kube/api_server.py

@@ -21,6 +21,7 @@
 from cmk.plugins.kube.controllers import map_controllers, map_controllers_top_to_down
 from cmk.plugins.kube.from_json.deployment import deployment_from_json, JSONDeploymentList
 from cmk.plugins.kube.from_json.node import JSONNodeList, node_list_from_json
+from cmk.plugins.kube.from_json.pod.pod import JSONPodList, pod_from_client
 from cmk.plugins.kube.from_json.statefulset import JSONStatefulSetList, statefulset_list_from_json
 from cmk.plugins.kube.schemata import api
 from cmk.plugins.kube.transform import (
@@ -30,7 +31,6 @@
     namespace_from_client,
     parse_object_to_owners,
     persistent_volume_claim_from_client,
-    pod_from_client,
     resource_quota_from_client,
 )
 from cmk.plugins.kube.transform_any import parse_open_metric_samples
@@ -82,10 +82,9 @@ def query_raw_jobs(self) -> Sequence[client.V1Job]:
 
 
 class ClientCoreAPI(ClientAPI):
-    def query_raw_pods(self) -> Sequence[client.V1Pod]:
+    def query_raw_pods(self) -> JSONPodList:
         request = requests.Request("GET", self._config.url("/api/v1/pods"))
-        response = send_request(self._config, self._client, request)
-        return self._deserializer.run("V1PodList", response).items
+        return send_request(self._config, self._client, request).json()
 
     def query_raw_resource_quotas(self) -> Sequence[client.V1ResourceQuota]:
         request = requests.Request("GET", self._config.url("/api/v1/resourcequotas"))
@@ -347,7 +346,7 @@ class APIData:
 class UnparsedAPIData:
     raw_jobs: Sequence[client.V1Job]
     raw_cron_jobs: Sequence[client.V1CronJob]
-    raw_pods: Sequence[client.V1Pod]
+    raw_pods: JSONPodList
     raw_nodes: JSONNodeList
     raw_namespaces: Sequence[client.V1Namespace]
     raw_resource_quotas: Sequence[client.V1ResourceQuota]
@@ -395,7 +394,7 @@ def query_raw_api_data_v2(
 
 def parse_api_data(
     raw_cron_jobs: Sequence[client.V1CronJob],
-    raw_pods: Sequence[client.V1Pod],
+    raw_pods: JSONPodList,
     raw_jobs: Sequence[client.V1Job],
     raw_nodes: JSONNodeList,
     raw_namespaces: Sequence[client.V1Namespace],
@@ -415,7 +414,7 @@ def parse_api_data(
 ) -> APIData:
     """Parses the Kubernetes API to the format used"""
     job_uids = {raw_job.metadata.uid for raw_job in raw_jobs}
-    pod_uids = {raw_pod.metadata.uid for raw_pod in raw_pods}
+    pod_uids = {raw_pod["metadata"]["uid"] for raw_pod in raw_pods["items"]}
 
     cron_jobs = [
         cron_job_from_client(
@@ -453,7 +452,10 @@ def parse_api_data(
     statefulsets = statefulset_list_from_json(raw_statefulsets, controller_to_pods)
     namespaces = [namespace_from_client(raw_namespace) for raw_namespace in raw_namespaces]
     nodes = node_list_from_json(raw_nodes, node_to_kubelet_health)
-    pods = [pod_from_client(pod, pod_to_controllers.get(pod.metadata.uid, [])) for pod in raw_pods]
+    pods = [
+        pod_from_client(pod, pod_to_controllers.get(api.PodUID(pod["metadata"]["uid"]), []))
+        for pod in raw_pods["items"]
+    ]
     persistent_volume_claims = [
         persistent_volume_claim_from_client(pvc) for pvc in raw_persistent_volume_claims
     ]
@@ -518,11 +520,11 @@ def create_api_data_v2(
             raw_api_data.raw_replica_sets,
             raw_api_data.raw_cron_jobs,
             raw_api_data.raw_jobs,
-            raw_api_data.raw_pods,
         ),
         workload_resources_json=itertools.chain(
             raw_api_data.raw_statefulsets["items"],
             raw_api_data.raw_deployments["items"],
+            raw_api_data.raw_pods["items"],
         ),
     )
     controller_to_pods, pod_to_controllers = map_controllers(

cmk/plugins/kube/controllers.py

@@ -6,9 +6,8 @@
 from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass
 
-from kubernetes import client  # type: ignore[import-untyped]
-
-from cmk.plugins.kube.schemata import api
+from .from_json.pod.pod import JSONPodList
+from .schemata import api
 
 
 @dataclass(frozen=True)
@@ -104,11 +103,11 @@ def _match_controllers(
 
 
 def map_controllers(
-    raw_pods: Sequence[client.V1Pod],
+    raw_pods: JSONPodList,
     object_to_owners: Mapping[str, api.OwnerReferences],
 ) -> tuple[Mapping[str, Sequence[api.PodUID]], Mapping[api.PodUID, Sequence[api.Controller]]]:
     pod_to_controllers = _find_control_chains(
-        pod_uids=(pod.metadata.uid for pod in raw_pods),
+        pod_uids=(api.PodUID(pod["metadata"]["uid"]) for pod in raw_pods["items"]),
         object_to_owners=object_to_owners,
     )
     return _match_controllers(pod_to_controllers), pod_to_controllers

cmk/plugins/kube/from_json/pod/__init__.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+# Copyright (C) 2026 Checkmk GmbH - License: GNU General Public License v2
+# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
+# conditions defined in the file COPYING, which is part of this source code package.
+
+"""
+JSON parsing for Kubernetes Pods.
+
+Separate package down a level from from_json, simply so we can split out
+pod-specific objects (like container specs and status), to avoid having a single
+large "pod.py".
+"""

cmk/plugins/kube/from_json/pod/container_spec.py

@@ -0,0 +1,49 @@
+# Copyright (C) 2026 Checkmk GmbH - License: GNU General Public License v2
+# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
+# conditions defined in the file COPYING, which is part of this source code package.
+
+from collections.abc import Sequence
+from typing import Literal, NotRequired, TypedDict
+
+from ...schemata import api
+from ..resources import JSONResourceRequirements
+
+
+class JSONContainerSpec(TypedDict):
+    name: str
+    imagePullPolicy: Literal["Always", "IfNotPresent", "Never"]
+    resources: NotRequired[JSONResourceRequirements]
+
+
+def container_resources(container: JSONContainerSpec) -> api.ResourceRequirements:
+    parsed_limits = api.ResourceRequirement()
+    parsed_requests = api.ResourceRequirement()
+    if resources := container.get("resources"):
+        if limits := resources.get("limits"):
+            parsed_limits = api.ResourceRequirement(
+                memory=api.parse_resource_value(limits["memory"]) if "memory" in limits else None,
+                cpu=api.parse_cpu_cores(limits["cpu"]) if "cpu" in limits else None,
+            )
+        if requests := resources.get("requests"):
+            parsed_requests = api.ResourceRequirement(
+                memory=api.parse_resource_value(requests["memory"])
+                if "memory" in requests
+                else None,
+                cpu=api.parse_cpu_cores(requests["cpu"]) if "cpu" in requests else None,
+            )
+
+    return api.ResourceRequirements(
+        limits=parsed_limits,
+        requests=parsed_requests,
+    )
+
+
+def containers_spec(containers: Sequence[JSONContainerSpec]) -> Sequence[api.ContainerSpec]:
+    return [
+        api.ContainerSpec(
+            name=api.ContainerName(container["name"]),
+            resources=container_resources(container),
+            image_pull_policy=container["imagePullPolicy"],
+        )
+        for container in containers
+    ]

cmk/plugins/kube/from_json/pod/container_status.py

@@ -0,0 +1,95 @@
+# Copyright (C) 2026 Checkmk GmbH - License: GNU General Public License v2
+# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
+# conditions defined in the file COPYING, which is part of this source code package.
+
+from collections.abc import Sequence
+from typing import NotRequired, TypedDict
+
+from ...schemata import api
+
+
+class JSONContainerStateTerminated(TypedDict):
+    exitCode: int
+    reason: NotRequired[str]
+    message: NotRequired[str]
+    startedAt: NotRequired[str]
+    finishedAt: NotRequired[str]
+
+
+class JSONContainerStateRunning(TypedDict):
+    startedAt: str
+
+
+class JSONContainerStateWaiting(TypedDict):
+    reason: NotRequired[str]
+    message: NotRequired[str]
+
+
+class JSONContainerState(TypedDict):
+    terminated: NotRequired[JSONContainerStateTerminated]
+    running: NotRequired[JSONContainerStateRunning]
+    waiting: NotRequired[JSONContainerStateWaiting]
+
+
+class JSONContainerStatus(TypedDict):
+    state: JSONContainerState
+    imageID: str
+    image: str
+    name: str
+    ready: bool
+    restartCount: int
+    containerID: NotRequired[str]
+
+
+def pod_containers(
+    container_statuses: Sequence[JSONContainerStatus] | None,
+) -> dict[str, api.ContainerStatus]:
+    result: dict[str, api.ContainerStatus] = {}
+    if container_statuses is None:
+        return {}
+    for status in container_statuses:
+        details: (
+            JSONContainerStateTerminated
+            | JSONContainerStateWaiting
+            | JSONContainerStateRunning
+            | None
+        )
+        state: api.ContainerTerminatedState | api.ContainerRunningState | api.ContainerWaitingState
+        if (details := status["state"].get("terminated")) is not None:
+            state = api.ContainerTerminatedState(
+                exit_code=details["exitCode"],
+                start_time=(
+                    int(api.convert_to_timestamp(started_at))
+                    if (started_at := details.get("startedAt"))
+                    else None
+                ),
+                end_time=(
+                    int(api.convert_to_timestamp(finished_at))
+                    if (finished_at := details.get("finishedAt"))
+                    else None
+                ),
+                reason=details.get("reason"),
+                detail=details.get("message"),
+            )
+        elif (details := status["state"].get("running")) is not None:
+            state = api.ContainerRunningState(
+                start_time=int(api.convert_to_timestamp(details["startedAt"])),
+            )
+        elif (details := status["state"].get("waiting")) is not None:
+            state = api.ContainerWaitingState(
+                reason=details.get("reason"),
+                detail=details.get("message"),
+            )
+        else:
+            raise AssertionError(f"Unknown container state {status['state']}")
+
+        result[status["name"]] = api.ContainerStatus(
+            container_id=status.get("containerID"),
+            image_id=status["imageID"],
+            name=status["name"],
+            image=status["image"],
+            ready=status["ready"],
+            state=state,
+            restart_count=status["restartCount"],
+        )
+    return result

cmk/plugins/kube/from_json/pod/pod.py

@@ -0,0 +1,33 @@
+# Copyright (C) 2026 Checkmk GmbH - License: GNU General Public License v2
+# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
+# conditions defined in the file COPYING, which is part of this source code package.
+
+from collections.abc import Sequence
+from typing import TypedDict
+
+from ...schemata import api
+from ..metadata import _metadata_from_json, JSONObjectWithMetadata
+from .container_status import pod_containers
+from .pod_spec import JSONPodSpec, pod_spec
+from .pod_status import JSONPodStatus, pod_status
+
+
+class JSONPod(JSONObjectWithMetadata):
+    spec: JSONPodSpec
+    status: JSONPodStatus
+
+
+class JSONPodList(TypedDict):
+    items: Sequence[JSONPod]
+
+
+def pod_from_client(pod: JSONPod, controllers: Sequence[api.Controller]) -> api.Pod:
+    return api.Pod(
+        uid=api.PodUID(pod["metadata"]["uid"]),
+        metadata=_metadata_from_json(pod["metadata"]),
+        status=pod_status(pod["status"]),
+        spec=pod_spec(pod["spec"]),
+        containers=pod_containers(pod["status"].get("containerStatuses")),
+        init_containers=pod_containers(pod["status"].get("initContainerStatuses")),
+        controllers=controllers,
+    )

cmk/plugins/kube/from_json/pod/pod_condition.py

@@ -0,0 +1,38 @@
+# Copyright (C) 2026 Checkmk GmbH - License: GNU General Public License v2
+# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
+# conditions defined in the file COPYING, which is part of this source code package.
+
+from collections.abc import Sequence
+from typing import NotRequired, TypedDict
+
+from ...schemata import api
+
+
+class JSONPodCondition(TypedDict):
+    status: str
+    type: str
+    reason: NotRequired[str]
+    message: NotRequired[str]
+    lastTransitionTime: NotRequired[str]
+
+
+def pod_condition(condition: JSONPodCondition) -> api.PodCondition:
+    type_ = api.ConditionType.from_kube_api(condition["type"])
+    custom_type = None if type_ is not None else condition["type"]
+    return api.PodCondition(
+        # TODO: CMK-33030, the JSON type is right, the api model is wrong
+        status=condition["status"],  # type: ignore[arg-type]
+        reason=condition.get("reason"),
+        detail=condition.get("message"),
+        last_transition_time=(
+            int(api.convert_to_timestamp(last_transition_time))
+            if (last_transition_time := condition.get("lastTransitionTime"))
+            else None
+        ),
+        type=type_,
+        custom_type=custom_type,
+    )
+
+
+def pod_conditions(conditions: Sequence[JSONPodCondition]) -> list[api.PodCondition]:
+    return [pod_condition(condition) for condition in conditions]

cmk/plugins/kube/from_json/pod/pod_spec.py

@@ -0,0 +1,36 @@
+# Copyright (C) 2026 Checkmk GmbH - License: GNU General Public License v2
+# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
+# conditions defined in the file COPYING, which is part of this source code package.
+
+from collections.abc import Sequence
+from typing import Literal, NotRequired, TypedDict
+
+from ...schemata import api
+from .container_spec import containers_spec, JSONContainerSpec
+from .volume import JSONPodVolume, parse_pod_volumes
+
+
+class JSONPodSpec(TypedDict):
+    nodeName: NotRequired[api.NodeName]
+    hostNetwork: NotRequired[bool]
+    dnsPolicy: NotRequired[str]
+    restartPolicy: Literal["Always", "OnFailure", "Never"]
+    containers: Sequence[JSONContainerSpec]
+    initContainers: NotRequired[Sequence[JSONContainerSpec]]
+    priorityClassName: NotRequired[str]
+    activeDeadlineSeconds: NotRequired[int]
+    volumes: NotRequired[Sequence[JSONPodVolume]]
+
+
+def pod_spec(spec: JSONPodSpec) -> api.PodSpec:
+    return api.PodSpec(
+        node=spec.get("nodeName"),
+        host_network=spec.get("hostNetwork"),
+        dns_policy=spec.get("dnsPolicy"),
+        restart_policy=spec["restartPolicy"],
+        containers=containers_spec(spec["containers"]),
+        init_containers=containers_spec(spec.get("initContainers", [])),
+        priority_class_name=spec.get("priorityClassName"),
+        active_deadline_seconds=spec.get("activeDeadlineSeconds"),
+        volumes=parse_pod_volumes(volumes) if (volumes := spec.get("volumes")) else None,
+    )