defmodule Integrations.Kubernetes do @moduledoc """ Kubernetes cluster health monitor. Checks node readiness, deployment replica availability, and pod phase distribution. Supports both in-cluster (ServiceAccount) and external (bearer token) authentication. Collection only — see `Integrations.Kubernetes.Display` (same package) for the dashboard panel. `Display.BundledDefault` auto-hooks it whenever this monitor starts, same as a single-module package would; a release without `raven_web` simply never compiles the display half and runs this monitor headless. ## Authentication **In-cluster (default):** When `:api_url`, `:token`, and `:ca_cert_path` are all unset, the monitor reads its credentials from the standard ServiceAccount mount at `/var/run/secrets/kubernetes.io/serviceaccount/`. Deploy with a ServiceAccount bound to a ClusterRole with `get` and `list` on `nodes`, `pods`, and `deployments`. **External:** Set `:api_url` to your API server URL, `:token` to a long-lived ServiceAccount token (or any bearer token with sufficient RBAC), and `:ca_cert_path` to the cluster CA bundle path on the Raven host. Use `ca_cert_path = "insecure"` to skip TLS verification (development only). ## Params * `:api_url` — Kubernetes API server URL. Defaults to `https://kubernetes.default.svc` in-cluster. * `:token` — Bearer token. Defaults to the ServiceAccount token when running in-cluster. * `:ca_cert_path` — CA certificate file path for TLS verification. Defaults to the ServiceAccount CA in-cluster. Set to `"insecure"` to skip verification. * `:namespace` — Limit pod and deployment checks to one namespace. Omit or leave blank to check all namespaces. * `:timeout_ms` — Per-request timeout. Defaults to `10000`. ## Health signal * `:up` — All nodes Ready; all deployments at full replica count. * `:degraded` — A deployment has fewer available replicas than desired, or a pod is CrashLoopBackOff / OOMKilled / Error. * `:down` — A node is NotReady, a deployment has 0 available replicas when desired > 0, or the API server is unreachable. """ use CodeNameRaven.Monitor @in_cluster_token_path "/var/run/secrets/kubernetes.io/serviceaccount/token" @in_cluster_ca_path "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" @in_cluster_api_url "https://kubernetes.default.svc" @default_timeout_ms 10_000 @impl true def params_template do %{ api_url: "", token: "", ca_cert_path: "", namespace: "", timeout_ms: "" } end @impl true def params_schema do [ api_url: [type: :string, required: false, doc: "Kubernetes API server URL (defaults to in-cluster https://kubernetes.default.svc)"], token: [type: :string, required: false, doc: "Bearer token (defaults to ServiceAccount token when in-cluster)"], ca_cert_path: [type: :string, required: false, doc: "CA cert file path, or 'insecure' to skip TLS verification"], namespace: [type: :string, required: false, doc: "Namespace to check (blank = all namespaces)"], timeout_ms: [type: :non_neg_integer, default: 10_000, doc: "Per-request timeout in milliseconds"] ] end @impl true def target_uri(params) do case get_param(params, :api_url) do url when is_binary(url) and url != "" -> {:ok, url} _ -> {:ok, @in_cluster_api_url} end end @impl true def identity_params(params) do %{ api_url: get_param(params, :api_url), namespace: get_param(params, :namespace) } end # --------------------------------------------------------------------------- # Collect # --------------------------------------------------------------------------- @impl true def collect(params, state) do case build_conn(params) do {:error, reason} -> {:error, reason, state} {:ok, conn} -> timeout_ms = parse_int(get_param(params, :timeout_ms), @default_timeout_ms) namespace = get_param(params, :namespace) || "" with {:ok, nodes} <- fetch_nodes(conn, timeout_ms), {:ok, deps} <- fetch_deployments(conn, namespace, timeout_ms), {:ok, pods} <- fetch_pods(conn, namespace, timeout_ms) do {:ok, %{nodes: nodes, deployments: deps, pods: pods}, state} else {:error, reason} -> {:error, reason, state} end end end # --------------------------------------------------------------------------- # Healthy? # --------------------------------------------------------------------------- @impl true def healthy?(%{nodes: nodes, deployments: deps, pods: pods}) do not_ready = Enum.count(nodes, &(&1.ready == false)) down_deps = Enum.count(deps, fn d -> d.desired > 0 and d.available == 0 end) deg_deps = Enum.count(deps, fn d -> d.available < d.desired end) crashing = Enum.count(pods, & &1.crash_looping) failed = Enum.count(pods, &(&1.phase == "Failed")) cond do not_ready > 0 -> :down down_deps > 0 -> :down deg_deps > 0 -> :degraded crashing > 0 -> :degraded failed > 0 -> :degraded true -> :up end end # --------------------------------------------------------------------------- # Metrics # --------------------------------------------------------------------------- @impl true def metrics(%{nodes: nodes, deployments: deps, pods: pods}) do %{ nodes_total: length(nodes), nodes_ready: Enum.count(nodes, &(&1.ready == true)), nodes_not_ready: Enum.count(nodes, &(&1.ready == false)), deployments_total: length(deps), deployments_ok: Enum.count(deps, fn d -> d.available == d.desired end), deployments_degraded: Enum.count(deps, fn d -> d.available < d.desired end), replicas_desired: Enum.sum(Enum.map(deps, & &1.desired)), replicas_available: Enum.sum(Enum.map(deps, & &1.available)), pods_total: length(pods), pods_running: Enum.count(pods, &(&1.phase == "Running")), pods_pending: Enum.count(pods, &(&1.phase == "Pending")), pods_failed: Enum.count(pods, &(&1.phase == "Failed")), pods_crash_looping: Enum.count(pods, & &1.crash_looping) } end # --------------------------------------------------------------------------- # Connection builder # --------------------------------------------------------------------------- defp build_conn(params) do api_url = get_param(params, :api_url) token = get_param(params, :token) ca_cert_path = get_param(params, :ca_cert_path) in_cluster = is_nil_or_empty(api_url) and is_nil_or_empty(token) {api_url, token, ca_cert_path} = if in_cluster do { @in_cluster_api_url, read_file(@in_cluster_token_path) || token, if(is_nil_or_empty(ca_cert_path), do: @in_cluster_ca_path, else: ca_cert_path) } else {api_url || @in_cluster_api_url, token, ca_cert_path} end if is_nil_or_empty(token) do {:error, "no bearer token: set :token param or run in-cluster with a ServiceAccount"} else ssl_opts = cond do ca_cert_path == "insecure" -> [verify: :verify_none] is_binary(ca_cert_path) and ca_cert_path != "" -> [verify: :verify_peer, cacertfile: String.to_charlist(ca_cert_path)] true -> [verify: :verify_peer, cacerts: :public_key.cacerts_get()] end {:ok, %{api_url: api_url, token: token, ssl_opts: ssl_opts}} end end defp api_get(%{api_url: base, token: token, ssl_opts: ssl_opts}, path, timeout_ms) do url = base <> path case Req.get(url, auth: {:bearer, token}, connect_options: [transport_opts: ssl_opts], receive_timeout: timeout_ms, retry: false, decode_body: true ) do {:ok, %{status: 200, body: body}} -> {:ok, body} {:ok, %{status: 401}} -> {:error, "unauthorized — check token and RBAC permissions"} {:ok, %{status: 403}} -> {:error, "forbidden — ServiceAccount lacks required RBAC permissions"} {:ok, %{status: status}} -> {:error, "API returned HTTP #{status} for #{path}"} {:error, %{reason: reason}} -> {:error, "request failed: #{inspect(reason)}"} {:error, reason} -> {:error, "request failed: #{inspect(reason)}"} end end # --------------------------------------------------------------------------- # Resource fetchers # --------------------------------------------------------------------------- defp fetch_nodes(conn, timeout_ms) do with {:ok, body} <- api_get(conn, "/api/v1/nodes", timeout_ms) do nodes = (body["items"] || []) |> Enum.map(fn item -> name = get_in(item, ["metadata", "name"]) ready = item |> get_in(["status", "conditions"]) |> List.wrap() |> Enum.find(&(&1["type"] == "Ready")) |> case do %{"status" => "True"} -> true %{"status" => "False"} -> false _ -> :unknown end roles = item |> get_in(["metadata", "labels"]) |> Kernel.||(%{}) |> Map.keys() |> Enum.filter(&String.starts_with?(&1, "node-role.kubernetes.io/")) |> Enum.map(&String.replace_prefix(&1, "node-role.kubernetes.io/", "")) |> Enum.join(",") %{name: name, ready: ready, roles: roles} end) {:ok, nodes} end end defp fetch_deployments(conn, namespace, timeout_ms) do path = if namespace != "" and not is_nil(namespace), do: "/apis/apps/v1/namespaces/#{namespace}/deployments", else: "/apis/apps/v1/deployments" with {:ok, body} <- api_get(conn, path, timeout_ms) do deps = (body["items"] || []) |> Enum.map(fn item -> %{ name: get_in(item, ["metadata", "name"]), namespace: get_in(item, ["metadata", "namespace"]), desired: get_in(item, ["spec", "replicas"]) || 0, ready: get_in(item, ["status", "readyReplicas"]) || 0, available: get_in(item, ["status", "availableReplicas"]) || 0 } end) {:ok, deps} end end defp fetch_pods(conn, namespace, timeout_ms) do path = if namespace != "" and not is_nil(namespace), do: "/api/v1/namespaces/#{namespace}/pods", else: "/api/v1/pods" with {:ok, body} <- api_get(conn, path, timeout_ms) do pods = (body["items"] || []) |> Enum.map(fn item -> phase = get_in(item, ["status", "phase"]) || "Unknown" crash_looping = item |> get_in(["status", "containerStatuses"]) |> List.wrap() |> Enum.any?(fn cs -> get_in(cs, ["state", "waiting", "reason"]) in [ "CrashLoopBackOff", "OOMKilled", "Error", "CreateContainerError", "CreateContainerConfigError" ] end) %{ name: get_in(item, ["metadata", "name"]), namespace: get_in(item, ["metadata", "namespace"]), phase: phase, crash_looping: crash_looping } end) {:ok, pods} end end # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- defp get_param(params, key) when is_atom(key) do v = params[key] || params[to_string(key)] if is_binary(v) and String.trim(v) == "", do: nil, else: v end defp is_nil_or_empty(nil), do: true defp is_nil_or_empty(""), do: true defp is_nil_or_empty(_), do: false defp read_file(path) do case File.read(path) do {:ok, content} -> String.trim(content) {:error, _} -> nil end end defp parse_int(nil, default), do: default defp parse_int(v, _) when is_integer(v), do: v defp parse_int(v, default) when is_binary(v) do case Integer.parse(v) do {n, _} -> n :error -> default end end defp parse_int(_, default), do: default end