raven_integration_dragonfly

defmodule Integrations.Dragonfly do
  @moduledoc """
  Dragonfly in-memory store monitor.

  Dragonfly is a Redis-compatible in-memory data store. This monitor connects
  via the RESP protocol over raw TCP, verifies the server is alive with a PING,
  and optionally fetches `INFO` statistics for memory, keyspace hit rate,
  connected clients, ops throughput, and eviction activity.

  Collection only — see `Integrations.Dragonfly.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`                — Dragonfly hostname or IP. Required.
    * `:port`                — Port number. Defaults to `6379`.
    * `:password`            — AUTH password (optional). Leave blank for no auth.
    * `:timeout_ms`          — TCP connect and receive timeout. Defaults to `5000`.
    * `:collect_info`        — Fetch INFO statistics. Defaults to `true`.
    * `:latency_degraded_ms` — Latency threshold for `:degraded`. Defaults to `200`.

  ## Health signal

    * `:up`       — PING received +PONG; latency within threshold.
    * `:degraded` — Server responded but latency exceeded threshold.
    * `:down`     — Connection refused, timeout, AUTH failure, or unexpected response.

  ## Metrics

    * `latency_ms`         — Round-trip time for PING
    * `used_memory_bytes`  — RSS memory in use
    * `maxmemory_bytes`    — Configured memory limit (0 = unlimited)
    * `connected_clients`  — Current client count
    * `hit_rate_pct`       — Keyspace hit rate (hits / (hits + misses) * 100)
    * `ops_per_sec`        — Instantaneous operations per second
    * `evicted_keys`       — Total keys evicted due to memory pressure
    * `expired_keys`       — Total keys expired by TTL
    * `replication_lag`    — `master_repl_offset - slave_repl_offset` for replicas
  """

  use CodeNameRaven.Monitor

  @default_port            6379
  @default_timeout_ms      5_000
  @default_lat_degraded_ms 200

  @impl true
  def params_template do
    %{host: "dragonfly.example.com", port: "6379", password: "", timeout_ms: "5000"}
  end

  @impl true
  def params_schema do
    [
      host:                [type: :string,          required: true,  doc: "Dragonfly hostname or IP"],
      port:                [type: :non_neg_integer, default: 6379,   doc: "Port number"],
      password:            [type: :string,          required: false, doc: "AUTH password (blank = no auth)"],
      timeout_ms:          [type: :non_neg_integer, default: 5_000,  doc: "TCP connect and receive timeout in milliseconds"],
      collect_info:        [type: :boolean,         default: true,   doc: "Fetch INFO statistics"],
      latency_degraded_ms: [type: :non_neg_integer, default: 200,    doc: "Latency 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, "redis://#{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)

    if is_nil(host) do
      {:error, "missing required param :host", state}
    else
      port         = parse_int(get_param(params, :port), @default_port)
      timeout_ms   = parse_int(get_param(params, :timeout_ms), @default_timeout_ms)
      password     = get_param(params, :password)
      collect_info = truthy?(get_param(params, :collect_info), default: true)

      tcp_opts = [:binary, active: false, packet: :raw]

      case :gen_tcp.connect(to_charlist(host), port, tcp_opts, timeout_ms) do
        {:ok, socket} ->
          result = do_check(socket, password, collect_info, timeout_ms)
          :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 after #{timeout_ms}ms", state}

        {:error, :nxdomain} ->
          {:error, "hostname not found: #{host}", state}

        {:error, reason} ->
          {:error, inspect(reason), state}
      end
    end
  end

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

  @impl true
  def healthy?(result) do
    lat_deg = result[:latency_degraded_ms] || @default_lat_degraded_ms
    if result.latency_ms >= lat_deg, do: :degraded, else: :up
  end

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

  @impl true
  def metrics(result) do
    %{latency_ms: result.latency_ms}
    |> maybe_put(:used_memory_bytes, result[:used_memory_bytes])
    |> maybe_put(:maxmemory_bytes,   result[:maxmemory_bytes])
    |> maybe_put(:connected_clients, result[:connected_clients])
    |> maybe_put(:hit_rate_pct,      result[:hit_rate_pct])
    |> maybe_put(:ops_per_sec,       result[:ops_per_sec])
    |> maybe_put(:evicted_keys,      result[:evicted_keys])
    |> maybe_put(:expired_keys,      result[:expired_keys])
    |> maybe_put(:replication_lag,   result[:replication_lag])
  end

  # ---------------------------------------------------------------------------
  # RESP protocol
  # ---------------------------------------------------------------------------

  defp do_check(socket, password, collect_info, timeout_ms) do
    with :ok <- maybe_auth(socket, password, timeout_ms) do
      t0 = System.monotonic_time(:millisecond)
      :gen_tcp.send(socket, "*1\r\n$4\r\nPING\r\n")

      case recv_line(socket, timeout_ms) do
        {:ok, "+PONG"} ->
          latency_ms = System.monotonic_time(:millisecond) - t0
          base = %{latency_ms: latency_ms}

          if collect_info do
            case fetch_info(socket, timeout_ms) do
              {:ok, info} -> {:ok, Map.merge(base, info)}
              _           -> {:ok, base}
            end
          else
            {:ok, base}
          end

        {:ok, other} ->
          {:error, "unexpected PING response: #{inspect(other)}"}

        {:error, reason} ->
          {:error, reason}
      end
    end
  end

  defp maybe_auth(_socket, nil, _timeout_ms), do: :ok
  defp maybe_auth(_socket, "", _timeout_ms),  do: :ok

  defp maybe_auth(socket, password, timeout_ms) do
    cmd = "*2\r\n$4\r\nAUTH\r\n$#{byte_size(password)}\r\n#{password}\r\n"
    :gen_tcp.send(socket, cmd)

    case recv_line(socket, timeout_ms) do
      {:ok, "+OK"}       -> :ok
      {:ok, "-" <> err} -> {:error, "AUTH failed: #{String.trim(err)}"}
      {:ok, other}       -> {:error, "AUTH unexpected: #{inspect(other)}"}
      {:error, reason}   -> {:error, reason}
    end
  end

  defp fetch_info(socket, timeout_ms) do
    :gen_tcp.send(socket, "*1\r\n$4\r\nINFO\r\n")

    case recv_bulk_string(socket, timeout_ms) do
      {:ok, body} -> {:ok, parse_info(body)}
      _           -> :error
    end
  end

  defp recv_bulk_string(socket, timeout_ms) do
    case recv_line(socket, timeout_ms) do
      {:ok, "$" <> size_str} ->
        size = String.to_integer(String.trim(size_str))
        case :gen_tcp.recv(socket, size + 2, timeout_ms) do
          {:ok, data} -> {:ok, binary_part(data, 0, size)}
          {:error, r} -> {:error, inspect(r)}
        end

      {:ok, other} ->
        {:error, "expected bulk string, got: #{inspect(other)}"}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp recv_line(socket, timeout_ms), do: recv_line_acc(socket, timeout_ms, "")

  defp recv_line_acc(socket, timeout_ms, acc) do
    case :gen_tcp.recv(socket, 1, timeout_ms) do
      {:ok, "\n"} -> {:ok, String.trim_trailing(acc, "\r")}
      {:ok, byte} -> recv_line_acc(socket, timeout_ms, acc <> byte)
      {:error, r} -> {:error, "recv failed: #{inspect(r)}"}
    end
  end

  defp parse_info(body) do
    fields =
      body
      |> String.split("\r\n")
      |> Enum.reject(&(String.starts_with?(&1, "#") or &1 == ""))
      |> Map.new(fn line ->
        case String.split(line, ":", parts: 2) do
          [k, v] -> {k, String.trim(v)}
          _      -> {"", ""}
        end
      end)

    %{}
    |> put_string(fields, "dragonfly_version", :version)
    |> put_integer(fields, "used_memory_rss",           :used_memory_bytes)
    |> put_integer(fields, "maxmemory",                 :maxmemory_bytes)
    |> put_integer(fields, "connected_clients",         :connected_clients)
    |> put_integer(fields, "instantaneous_ops_per_sec", :ops_per_sec)
    |> put_integer(fields, "evicted_keys",              :evicted_keys)
    |> put_integer(fields, "expired_keys",              :expired_keys)
    |> put_hit_rate(fields)
    |> put_replication_lag(fields)
  end

  defp put_string(result, fields, key, atom) do
    case Map.get(fields, key) do
      nil -> result
      v   -> Map.put(result, atom, v)
    end
  end

  defp put_integer(result, fields, key, atom) do
    case Integer.parse(fields[key] || "") do
      {n, _} -> Map.put(result, atom, n)
      _      -> result
    end
  end

  defp put_hit_rate(result, fields) do
    with {hits, _}   <- Integer.parse(fields["keyspace_hits"] || ""),
         {misses, _} <- Integer.parse(fields["keyspace_misses"] || ""),
         total = hits + misses,
         true <- total > 0 do
      Map.put(result, :hit_rate_pct, round(hits / total * 100))
    else
      _ -> result
    end
  end

  defp put_replication_lag(result, fields) do
    with {master, _} <- Integer.parse(fields["master_repl_offset"] || ""),
         {slave, _}  <- Integer.parse(fields["slave_repl_offset"] || "") do
      Map.put(result, :replication_lag, master - slave)
    else
      _ -> result
    end
  end

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

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

  defp truthy?(nil, opts),   do: Keyword.get(opts, :default, false)
  defp truthy?("true", _),   do: true
  defp truthy?(true, _),     do: true
  defp truthy?("false", _),  do: false
  defp truthy?(false, _),    do: false
  defp truthy?(_, opts),     do: Keyword.get(opts, :default, false)

  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