raven_integration_kafka

defmodule Integrations.Kafka do
  @moduledoc """
  Apache Kafka broker monitor.

  Speaks the native Kafka binary wire protocol over a plain TCP connection
  (PLAINTEXT — no SASL or TLS). Sends two requests per check:

  1. **ApiVersions v0** — lightest possible request; confirms the socket is a
     Kafka broker. Round-trip time becomes `latency_ms`.
  2. **Metadata v1** — fetches the full cluster topology: broker list,
     controller election result, topic list, and per-partition leader + ISR
     state. Used to compute `under_replicated_partitions` and
     `offline_partitions`.

  No external libraries required — all encoding and decoding is done with
  Elixir binary pattern matching and bitstring construction.

  Collection only — see `Integrations.Kafka.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.

  ## Params

    * `:host`                — Kafka broker hostname or IP. Required.
    * `:port`                — Kafka broker port. Defaults to `9092`.
    * `:timeout_ms`          — TCP connect and receive timeout. Defaults to `10000`.
    * `:latency_degraded_ms` — ApiVersions round-trip threshold for `:degraded`.
                               Defaults to `500`.

  ## Health signal

    * `:up`       — Connected; broker speaking Kafka protocol; no offline or
                    under-replicated partitions; latency within threshold.
    * `:degraded` — Any partition offline (no leader), any partition
                    under-replicated (ISR < replicas), or latency exceeds
                    threshold.
    * `:down`     — Connection refused, timeout, hostname not found, or
                    invalid protocol response.

  ## Metrics

    * `latency_ms`                — ApiVersions round-trip time
    * `broker_count`              — Active brokers in the cluster
    * `topic_count`               — User topics (internal topics excluded)
    * `partition_count`           — Total partitions across all topics
    * `under_replicated_partitions` — Partitions where ISR < replica set
    * `offline_partitions`        — Partitions with no elected leader
  """

  use CodeNameRaven.Monitor

  @default_port            9092
  @default_timeout_ms      10_000
  @default_lat_degraded_ms 500
  @client_id               "raven"

  # Kafka API keys
  @api_metadata     3
  @api_api_versions 18

  @impl true
  def params_template do
    %{
      host:       "kafka.example.com",
      port:       "9092",
      timeout_ms: "10000"
    }
  end

  @impl true
  def params_schema do
    [
      host:                [type: :string,          required: true,  doc: "Kafka broker hostname or IP"],
      port:                [type: :non_neg_integer, default: 9092,   doc: "Kafka broker port"],
      timeout_ms:          [type: :non_neg_integer, default: 10_000, doc: "TCP connect and receive timeout in milliseconds"],
      latency_degraded_ms: [type: :non_neg_integer, default: 500,    doc: "ApiVersions round-trip threshold for degraded"]
    ]
  end

  @impl true
  def target_uri(params) do
    host = get_param(params, :host)
    port = parse_int(get_param(params, :port), @default_port)
    if host, do: {:ok, "kafka://#{host}:#{port}"}, else: :none
  end

  @impl true
  def identity_params(params) do
    %{
      host: get_param(params, :host),
      port: parse_int(get_param(params, :port), @default_port)
    }
  end

  # ---------------------------------------------------------------------------
  # Collect
  # ---------------------------------------------------------------------------

  @impl true
  def collect(params, state) do
    host    = get_param(params, :host)
    port    = parse_int(get_param(params, :port), @default_port)
    timeout = parse_int(get_param(params, :timeout_ms), @default_timeout_ms)
    lat_deg = parse_int(get_param(params, :latency_degraded_ms), @default_lat_degraded_ms)

    if is_nil(host) do
      {:error, "missing required param :host", state}
    else
      tcp_opts = [:binary, active: false, packet: :raw]

      case :gen_tcp.connect(to_charlist(host), port, tcp_opts, timeout) do
        {:ok, socket} ->
          result = run_checks(socket, lat_deg, timeout)
          :gen_tcp.close(socket)
          case result do
            {:ok, data}      -> {:ok, data, state}
            {:error, reason} -> {:error, reason, state}
          end

        {:error, :econnrefused} -> {:error, "connection refused on port #{port}", state}
        {:error, :timeout}      -> {:error, "connection timed out", state}
        {:error, :nxdomain}     -> {:error, "hostname not found: #{host}", state}
        {:error, reason}        -> {:error, inspect(reason), state}
      end
    end
  end

  # ---------------------------------------------------------------------------
  # Healthy?
  # ---------------------------------------------------------------------------

  @impl true
  def healthy?(%{offline_partitions: n}) when is_integer(n) and n > 0, do: :degraded

  def healthy?(%{under_replicated_partitions: n}) when is_integer(n) and n > 0, do: :degraded

  def healthy?(%{latency_ms: lat, latency_degraded_ms: deg})
      when is_integer(lat) and is_integer(deg) and lat >= deg,
      do: :degraded

  def healthy?(_), do: :up

  # ---------------------------------------------------------------------------
  # Metrics
  # ---------------------------------------------------------------------------

  @impl true
  def metrics(result) do
    %{latency_ms: result.latency_ms}
    |> maybe_put(:broker_count,                result[:broker_count])
    |> maybe_put(:topic_count,                 result[:topic_count])
    |> maybe_put(:partition_count,             result[:partition_count])
    |> maybe_put(:under_replicated_partitions, result[:under_replicated_partitions])
    |> maybe_put(:offline_partitions,          result[:offline_partitions])
  end

  # ---------------------------------------------------------------------------
  # Protocol flow
  # ---------------------------------------------------------------------------

  defp run_checks(socket, lat_deg, timeout) do
    t0 = System.monotonic_time(:millisecond)

    with {:ok, av_payload} <- exchange(socket, build_api_versions_v0(corr: 1), timeout),
         :ok <- validate_api_versions(av_payload) do

      latency_ms = System.monotonic_time(:millisecond) - t0

      meta =
        case exchange(socket, build_metadata_v1(corr: 2), timeout) do
          {:ok, payload} ->
            case decode_metadata_v1(payload) do
              {:ok, m}  -> m
              {:error, _} -> %{}
            end
          {:error, _} ->
            %{}
        end

      {:ok, Map.merge(meta, %{
        latency_ms:          latency_ms,
        latency_degraded_ms: lat_deg
      })}
    end
  end

  # ---------------------------------------------------------------------------
  # Request builders
  # ---------------------------------------------------------------------------

  # Every Kafka request: [4-byte length][2-byte api_key][2-byte version]
  #                      [4-byte correlation_id][2-byte client_id_len][client_id]
  #                      [... body ...]
  defp kafka_request(api_key, version, corr_id, body) do
    client = @client_id
    header = <<
      api_key::big-signed-16,
      version::big-signed-16,
      corr_id::big-signed-32,
      byte_size(client)::big-signed-16,
      client::binary
    >>
    msg = header <> body
    <<byte_size(msg)::big-signed-32, msg::binary>>
  end

  # ApiVersions v0: empty body
  defp build_api_versions_v0(corr: n), do: kafka_request(@api_api_versions, 0, n, <<>>)

  # Metadata v1: topics = null array (-1) means "all topics"
  defp build_metadata_v1(corr: n), do: kafka_request(@api_metadata, 1, n, <<-1::big-signed-32>>)

  # ---------------------------------------------------------------------------
  # TCP send + recv
  # ---------------------------------------------------------------------------

  defp exchange(socket, request, timeout_ms) do
    :gen_tcp.send(socket, request)
    read_frame(socket, timeout_ms)
  end

  # Kafka response framing: [4-byte length][payload]
  defp read_frame(socket, timeout_ms) do
    with {:ok, <<len::big-signed-32>>} <- :gen_tcp.recv(socket, 4, timeout_ms),
         {:ok, payload}               <- :gen_tcp.recv(socket, len, timeout_ms) do
      {:ok, payload}
    else
      {:error, :timeout} -> {:error, "timed out waiting for broker response"}
      {:error, reason}   -> {:error, inspect(reason)}
    end
  end

  # ---------------------------------------------------------------------------
  # ApiVersions v0 response validation
  # ---------------------------------------------------------------------------

  # Response: [4-byte correlation_id][2-byte error_code][...api version entries...]
  # Any well-formed response (4+ bytes with matching structure) proves this is Kafka.
  defp validate_api_versions(<<_corr::big-signed-32, 0::big-signed-16, _::binary>>), do: :ok
  defp validate_api_versions(<<_corr::big-signed-32, code::big-signed-16, _::binary>>),
    do: {:error, "ApiVersions returned error code #{code}"}
  defp validate_api_versions(_),
    do: {:error, "response does not match Kafka wire protocol — wrong port?"}

  # ---------------------------------------------------------------------------
  # Metadata v1 decoder
  # ---------------------------------------------------------------------------

  # Metadata v1 response layout (after the 4-byte correlation_id):
  #   brokers:    ARRAY{ node_id INT32, host STRING, port INT32, rack NULLABLE_STRING }
  #   controller: INT32
  #   topics:     ARRAY{ error_code INT16, name STRING, is_internal BOOLEAN,
  #                       partitions: ARRAY{ error_code INT16, partition_id INT32,
  #                                          leader INT32,
  #                                          replicas ARRAY(INT32), isr ARRAY(INT32) } }
  defp decode_metadata_v1(<<_corr::big-signed-32, rest::binary>>) do
    {brokers, rest}        = decode_broker_array(rest)
    <<ctrl::big-signed-32, rest::binary>> = rest
    {topics, _rest}        = decode_topic_array(rest)

    user_topics    = Enum.reject(topics, & &1.is_internal)
    all_partitions = Enum.flat_map(topics, & &1.partitions)

    under_replicated =
      Enum.count(all_partitions, fn p -> length(p.isr) < length(p.replicas) end)

    offline =
      Enum.count(all_partitions, fn p ->
        # leader == -1 means no leader elected; error_code 5 = LEADER_NOT_AVAILABLE
        p.leader == -1 or p.error_code == 5
      end)

    {:ok, %{
      broker_count:                 length(brokers),
      brokers:                      brokers,
      controller_id:                ctrl,
      topic_count:                  length(user_topics),
      partition_count:              length(all_partitions),
      under_replicated_partitions:  under_replicated,
      offline_partitions:           offline
    }}
  rescue
    e -> {:error, "Metadata decode failed: #{Exception.message(e)}"}
  end

  defp decode_broker_array(<<count::big-signed-32, rest::binary>>) do
    decode_brokers(rest, count, [])
  end

  defp decode_brokers(data, 0, acc), do: {Enum.reverse(acc), data}
  defp decode_brokers(<<node_id::big-signed-32, rest::binary>>, n, acc) do
    {host, rest} = decode_string(rest)
    <<port::big-signed-32, rest::binary>> = rest
    {_rack, rest} = decode_string(rest)                # NULLABLE_STRING rack (v1+)
    decode_brokers(rest, n - 1, [%{node_id: node_id, host: host, port: port} | acc])
  end

  defp decode_topic_array(<<count::big-signed-32, rest::binary>>) do
    decode_topics(rest, count, [])
  end

  defp decode_topics(data, 0, acc), do: {Enum.reverse(acc), data}
  defp decode_topics(<<error_code::big-signed-16, rest::binary>>, n, acc) do
    {name, rest}        = decode_string(rest)
    <<internal::8, rest::binary>> = rest
    {partitions, rest}  = decode_partition_array(rest)
    topic = %{error_code: error_code, name: name, is_internal: internal == 1, partitions: partitions}
    decode_topics(rest, n - 1, [topic | acc])
  end

  defp decode_partition_array(<<count::big-signed-32, rest::binary>>) do
    decode_partitions(rest, count, [])
  end

  defp decode_partitions(data, 0, acc), do: {Enum.reverse(acc), data}
  defp decode_partitions(
    <<error_code::big-signed-16, partition_id::big-signed-32, leader::big-signed-32, rest::binary>>,
    n, acc
  ) do
    {replicas, rest} = decode_int32_array(rest)
    {isr, rest}      = decode_int32_array(rest)
    part = %{error_code: error_code, partition_id: partition_id, leader: leader,
             replicas: replicas, isr: isr}
    decode_partitions(rest, n - 1, [part | acc])
  end

  # Kafka STRING: INT16 length (-1 = null) + UTF-8 bytes
  defp decode_string(<<-1::big-signed-16, rest::binary>>),                          do: {nil, rest}
  defp decode_string(<<len::big-signed-16, s::binary-size(len), rest::binary>>),    do: {s, rest}

  # Kafka ARRAY(INT32): INT32 count (-1 = null/empty) + elements
  defp decode_int32_array(<<-1::big-signed-32, rest::binary>>), do: {[], rest}
  defp decode_int32_array(<<count::big-signed-32, rest::binary>>) do
    decode_int32s(rest, count, [])
  end
  defp decode_int32s(data, 0, acc), do: {Enum.reverse(acc), data}
  defp decode_int32s(<<n::big-signed-32, rest::binary>>, count, acc) do
    decode_int32s(rest, count - 1, [n | acc])
  end

  # ---------------------------------------------------------------------------
  # Helpers
  # ---------------------------------------------------------------------------

  defp maybe_put(map, _key, nil), do: map
  defp maybe_put(map, key, val),  do: Map.put(map, key, val)

  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 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