raven_integration_rabbit_mq

defmodule Integrations.RabbitMq do
  @moduledoc """
  RabbitMQ broker monitor.

  Uses the RabbitMQ Management HTTP API (port 15672 by default) to collect
  cluster-wide queue depth, message throughput rates, connection counts, and
  node-level resource usage. Requires the `rabbitmq_management` plugin to be
  enabled on the broker.

  Rates (publish/s, deliver/s, ack/s) are reported directly by the Management
  API — no delta state is needed on the Raven side.

  Collection only — see `Integrations.RabbitMq.Display` (same package) for
  the dashboard panel. `Display.BundledDefault` auto-hooks it whenever this
  monitor starts.

  ## Params

    * `:host`                — RabbitMQ hostname or IP. Required.
    * `:amqp_port`           — AMQP port (documented for reference). Defaults to `5672`.
    * `:management_port`     — Management API HTTP port. Defaults to `15672`.
    * `:user`                — Management API username. Required.
    * `:password`            — Management API password.
    * `:vhost`               — Virtual host to scope queue stats. Defaults to `/`.
    * `:timeout_ms`          — HTTP request timeout. Defaults to `5000`.
    * `:latency_degraded_ms` — API round-trip threshold for `:degraded`. Defaults to `500`.

  ## Health signal

    * `:up`       — API reachable; no node memory or disk alarms.
    * `:degraded` — API latency above threshold.
    * `:down`     — Connection refused, auth failure, or a node memory/disk alarm is firing.

  ## Metrics

    * `latency_ms`       — `/api/overview` round-trip time
    * `total_messages`   — Total messages across all queues
    * `unacked_messages` — Messages delivered but not yet acknowledged
    * `consumers`        — Total consumer count
    * `connections`      — Total connection count
    * `queues`           — Total queue count
    * `publish_rate`     — Messages published per second
    * `deliver_rate`     — Messages delivered per second
    * `ack_rate`         — Messages acknowledged per second
  """

  use CodeNameRaven.Monitor

  @default_mgmt_port       15672
  @default_timeout_ms      5_000
  @default_lat_degraded_ms 500

  @impl true
  def params_template do
    %{
      host:            "rabbitmq.example.com",
      amqp_port:       "5672",
      management_port: "15672",
      user:            "user",
      password:        "",
      vhost:           "/"
    }
  end

  @impl true
  def params_schema do
    [
      host:                [type: :string,          required: true,  doc: "RabbitMQ hostname or IP"],
      amqp_port:           [type: :non_neg_integer, default: 5672,   doc: "AMQP port (for reference)"],
      management_port:     [type: :non_neg_integer, default: 15672,  doc: "Management plugin HTTP port"],
      user:                [type: :string,          required: true,  doc: "Management API username"],
      password:            [type: :string,          required: false, doc: "Management API password"],
      vhost:               [type: :string,          default: "/",    doc: "Virtual host"],
      timeout_ms:          [type: :non_neg_integer, default: 5_000,  doc: "HTTP request timeout in milliseconds"],
      latency_degraded_ms: [type: :non_neg_integer, default: 500,    doc: "API round-trip threshold for degraded"]
    ]
  end

  @impl true
  def target_uri(params) do
    host = get_param(params, :host)
    port = parse_int(get_param(params, :management_port), @default_mgmt_port)
    if host, do: {:ok, "http://#{host}:#{port}/api/overview"}, else: :none
  end

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

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

  @impl true
  def collect(params, state) do
    host     = get_param(params, :host)
    mgmt_port = parse_int(get_param(params, :management_port), @default_mgmt_port)
    user     = get_param(params, :user)
    password = get_param(params, :password) || ""
    timeout  = parse_int(get_param(params, :timeout_ms), @default_timeout_ms)

    cond do
      is_nil(host) -> {:error, "missing required param :host", state}
      is_nil(user) -> {:error, "missing required param :user", state}
      true ->
        base     = "http://#{host}:#{mgmt_port}"
        req_opts = [
          auth:            {user, password},
          receive_timeout: timeout,
          connect_options: [timeout: timeout],
          retry:           false
        ]

        case run_checks(base, req_opts) do
          {:ok, result}    -> {:ok, result, state}
          {:error, reason} -> {:error, reason, state}
        end
    end
  end

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

  @impl true
  def healthy?(%{assertions: [_ | _], assertion_status: status}), do: status

  def healthy?(%{alarms: %{mem_alarm: true}}),       do: :down
  def healthy?(%{alarms: %{disk_free_alarm: true}}), do: :down

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

  def healthy?(_), do: :up

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

  @impl true
  def metrics(result) do
    qt = result[:queue_totals]  || %{}
    ot = result[:object_totals] || %{}
    ms = result[:message_stats] || %{}

    %{latency_ms: result.latency_ms}
    |> maybe_put(:total_messages,  qt[:messages])
    |> maybe_put(:unacked_messages, qt[:messages_unacknowledged])
    |> maybe_put(:consumers,       ot[:consumers])
    |> maybe_put(:connections,     ot[:connections])
    |> maybe_put(:queues,          ot[:queues])
    |> maybe_put(:publish_rate,    ms[:publish_rate])
    |> maybe_put(:deliver_rate,    ms[:deliver_rate])
    |> maybe_put(:ack_rate,        ms[:ack_rate])
  end

  # ---------------------------------------------------------------------------
  # HTTP helpers
  # ---------------------------------------------------------------------------

  defp run_checks(base, req_opts) do
    t0 = System.monotonic_time(:millisecond)

    with {:ok, overview} <- fetch(base, "/api/overview", req_opts) do
      latency_ms = System.monotonic_time(:millisecond) - t0

      nodes = case fetch(base, "/api/nodes", req_opts) do
        {:ok, list} when is_list(list) -> list
        _                              -> []
      end

      qt = overview["queue_totals"]  || %{}
      ot = overview["object_totals"] || %{}
      ms = overview["message_stats"] || %{}

      mem_alarm  = Enum.any?(nodes, & &1["mem_alarm"])
      disk_alarm = Enum.any?(nodes, & &1["disk_free_alarm"])
      primary    = List.first(nodes) || %{}

      {:ok, %{
        latency_ms:   latency_ms,
        version:      overview["rabbitmq_version"],
        cluster_name: overview["cluster_name"],
        node_count:   length(nodes),
        queue_totals: %{
          messages:                qt["messages"],
          messages_ready:          qt["messages_ready"],
          messages_unacknowledged: qt["messages_unacknowledged"]
        },
        object_totals: %{
          consumers:   ot["consumers"],
          queues:      ot["queues"],
          exchanges:   ot["exchanges"],
          connections: ot["connections"],
          channels:    ot["channels"]
        },
        message_stats: %{
          publish_rate: rate_of(ms, "publish"),
          deliver_rate: rate_of(ms, "deliver"),
          ack_rate:     rate_of(ms, "ack")
        },
        node: %{
          mem_used_mb:  to_mb(primary["mem_used"]),
          mem_limit_mb: to_mb(primary["mem_limit"]),
          disk_free_mb: to_mb(primary["disk_free"]),
          fd_used:      primary["fd_used"],
          fd_total:     primary["fd_total"],
          proc_used:    primary["proc_used"],
          proc_total:   primary["proc_total"]
        },
        alarms: %{
          mem_alarm:       mem_alarm,
          disk_free_alarm: disk_alarm
        }
      }}
    end
  end

  defp fetch(base, path, req_opts) do
    case Req.get(base <> path, req_opts) do
      {:ok, %{status: 200, body: body}} ->
        {:ok, body}

      {:ok, %{status: 401}} ->
        {:error, "authentication failed — check user and password"}

      {:ok, %{status: status, body: body}} ->
        reason =
          (is_map(body) && (body["reason"] || body["error"])) ||
          "HTTP #{status}"
        {:error, "Management API error: #{reason}"}

      {:error, %{reason: :econnrefused}} ->
        {:error, "connection refused — is the rabbitmq_management plugin enabled?"}

      {:error, %{reason: :timeout}} ->
        {:error, "request timed out"}

      {:error, reason} ->
        {:error, "request failed: #{inspect(reason)}"}
    end
  end

  # ---------------------------------------------------------------------------
  # Data helpers
  # ---------------------------------------------------------------------------

  # Management API rate format: %{"publish_details" => %{"rate" => 1.5}, ...}
  defp rate_of(stats, key), do: get_in(stats, ["#{key}_details", "rate"])

  defp to_mb(nil), do: nil
  defp to_mb(b) when is_integer(b), do: div(b, 1_048_576)
  defp to_mb(b) when is_float(b),   do: round(b / 1_048_576)
  defp to_mb(_), do: nil

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

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