raven_integration_kubernetes

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