raven_integration_open_search

defmodule Integrations.OpenSearch do
  @moduledoc """
  OpenSearch (and Elasticsearch-compatible) cluster monitor.

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

  Uses two REST endpoints:

  * `GET /` — cluster version (non-fatal; skipped if unavailable).
  * `GET /_cluster/health` — cluster status (green/yellow/red), shard assignment
    counts, node count. Round-trip time is used as `latency_ms`.
  * `GET /_cluster/stats` — aggregate document counts, index store size, JVM
    heap, OS memory, and cumulative indexing/search totals. Rates are computed
    by diffing consecutive totals against state.

  Authentication uses HTTP Basic Auth. TLS is enabled by default; set
  `verify_tls: false` to accept self-signed certificates (typical for
  in-cluster deployments managed by the OpenSearch Operator).

  ## Params

    * `:host`                — OpenSearch hostname or IP. Required.
    * `:port`                — HTTP/HTTPS port. Defaults to `9200`.
    * `:user`                — HTTP Basic Auth username.
    * `:password`            — HTTP Basic Auth password.
    * `:tls`                 — Use HTTPS. Defaults to `true`.
    * `:verify_tls`          — Verify TLS certificates. Defaults to `false`
                               (self-signed certs are common in Kubernetes).
    * `:timeout_ms`          — Request timeout in milliseconds. Defaults to `10000`.
    * `:latency_degraded_ms` — Round-trip threshold for `:degraded`. Defaults to `1000`.

  ## Health signal

    * `:up`       — Cluster status is `green`.
    * `:degraded` — Cluster status is `yellow`, unassigned shards exist, or
                    API latency exceeds threshold.
    * `:down`     — Connection failed, auth rejected, or cluster status is `red`.

  ## Metrics

    * `latency_ms`       — `/_cluster/health` round-trip time
    * `node_count`       — Total cluster nodes
    * `active_shards`    — Total active shards
    * `unassigned_shards`— Shards not yet assigned to any node
    * `docs_count`       — Total documents across all indices
    * `store_size_bytes` — Total index storage in bytes
    * `heap_used_pct`    — Aggregate JVM heap pressure (0–100)
    * `indexing_rate`    — Documents indexed per second (requires two checks)
    * `search_rate`      — Search queries per second (requires two checks)
  """

  use CodeNameRaven.Monitor

  @default_port            9200
  @default_timeout_ms      10_000
  @default_lat_degraded_ms 1_000

  @impl true
  def params_template do
    %{
      host:       "opensearch.example.com",
      port:       "9200",
      user:       "admin",
      password:   "",
      tls:        "true",
      verify_tls: "false"
    }
  end

  @impl true
  def params_schema do
    [
      host:                [type: :string,          required: true,  doc: "OpenSearch hostname or IP"],
      port:                [type: :non_neg_integer, default: 9200,   doc: "HTTP/HTTPS port"],
      user:                [type: :string,          required: false, doc: "HTTP Basic Auth username"],
      password:            [type: :string,          required: false, doc: "HTTP Basic Auth password"],
      tls:                 [type: :boolean,         default: true,   doc: "Use HTTPS (recommended)"],
      verify_tls:          [type: :boolean,         default: false,  doc: "Verify TLS certificate; disable for self-signed"],
      timeout_ms:          [type: :non_neg_integer, default: 10_000, doc: "HTTP request timeout in milliseconds"],
      latency_degraded_ms: [type: :non_neg_integer, default: 1_000,  doc: "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)
    tls    = truthy?(get_param(params, :tls), default: true)
    scheme = if tls, do: "https", else: "http"
    if host, do: {:ok, "#{scheme}://#{host}:#{port}/_cluster/health"}, 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)
    user       = get_param(params, :user)
    password   = get_param(params, :password) || ""
    tls        = truthy?(get_param(params, :tls), default: true)
    verify_tls = truthy?(get_param(params, :verify_tls), default: false)
    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
      scheme = if tls, do: "https", else: "http"
      base   = "#{scheme}://#{host}:#{port}"

      ssl_opts = if verify_tls,
        do:   [verify: :verify_peer, cacerts: :public_key.cacerts_get()],
        else: [verify: :verify_none]

      req_opts =
        [
          connect_options: [transport_opts: ssl_opts],
          receive_timeout: timeout,
          retry:           false
        ]
        |> add_auth(user, password)

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

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

  @impl true
  def healthy?(%{cluster_status: "red"}),    do: :down
  def healthy?(%{cluster_status: "yellow"}), do: :degraded

  def healthy?(%{unassigned_shards: 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(:node_count,        result[:node_count])
    |> maybe_put(:active_shards,     result[:active_shards])
    |> maybe_put(:unassigned_shards, result[:unassigned_shards])
    |> maybe_put(:docs_count,        result[:docs_count])
    |> maybe_put(:store_size_bytes,  result[:store_size_bytes])
    |> maybe_put(:heap_used_pct,     result[:heap_used_pct])
    |> maybe_put(:indexing_rate,     result[:indexing_rate])
    |> maybe_put(:search_rate,       result[:search_rate])
  end

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

  defp run_checks(base, req_opts, state, lat_deg) do
    # Non-fatal: fetch version from root endpoint
    version =
      case Req.get(base <> "/", req_opts) do
        {:ok, %{status: 200, body: b}} when is_map(b) -> get_in(b, ["version", "number"])
        _                                              -> nil
      end

    t0 = System.monotonic_time(:millisecond)

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

      # Non-fatal: cluster stats enrich the result but are not required
      cs =
        case fetch(base, "/_cluster/stats", req_opts) do
          {:ok, s} -> s
          _        -> %{}
        end

      cs_indices = cs["indices"]  || %{}
      cs_nodes   = cs["nodes"]    || %{}
      docs       = cs_indices["docs"]    || %{}
      store      = cs_indices["store"]   || %{}
      indexing   = cs_indices["indexing"] || %{}
      search_s   = cs_indices["search"]   || %{}
      jvm_mem    = (cs_nodes["jvm"] || %{})["mem"] || %{}
      os_mem     = (cs_nodes["os"]  || %{})["mem"] || %{}

      heap_used = jvm_mem["heap_used_in_bytes"]
      heap_max  = jvm_mem["heap_max_in_bytes"]
      heap_pct  =
        if is_number(heap_used) and is_number(heap_max) and heap_max > 0,
          do: round(heap_used / heap_max * 100),
          else: nil

      index_total = indexing["index_total"]
      query_total = search_s["query_total"]
      now         = DateTime.utc_now()

      {indexing_rate, search_rate} = compute_rates(index_total, query_total, now, state)

      result = %{
        latency_ms:           latency_ms,
        latency_degraded_ms:  lat_deg,
        version:              version,
        cluster_name:         health["cluster_name"],
        cluster_status:       health["status"],
        node_count:           health["number_of_nodes"],
        data_node_count:      health["number_of_data_nodes"],
        active_shards:        health["active_shards"],
        active_primary:       health["active_primary_shards"],
        relocating_shards:    health["relocating_shards"],
        initializing_shards:  health["initializing_shards"],
        unassigned_shards:    health["unassigned_shards"],
        pending_tasks:        health["number_of_pending_tasks"],
        index_count:          cs_indices["count"],
        docs_count:           docs["count"],
        store_size_bytes:     store["size_in_bytes"],
        heap_used_bytes:      heap_used,
        heap_max_bytes:       heap_max,
        heap_used_pct:        heap_pct,
        os_mem_used_pct:      os_mem["used_percent"],
        indexing_rate:        indexing_rate,
        search_rate:          search_rate
      }

      new_state = %{
        prev_index_total:  index_total,
        prev_query_total:  query_total,
        prev_collected_at: now
      }

      {:ok, result, new_state}
    end
  end

  defp compute_rates(index_total, query_total, now, state) do
    prev_idx = state[:prev_index_total]
    prev_qry = state[:prev_query_total]
    prev_at  = state[:prev_collected_at]

    if is_nil(prev_at) or is_nil(prev_idx) do
      {nil, nil}
    else
      elapsed = DateTime.diff(now, prev_at, :millisecond) / 1000.0

      if elapsed > 0 do
        idx_rate =
          if is_integer(index_total) and is_integer(prev_idx) and index_total >= prev_idx,
            do: Float.round((index_total - prev_idx) / elapsed, 2)

        qry_rate =
          if is_integer(query_total) and is_integer(prev_qry) and query_total >= prev_qry,
            do: Float.round((query_total - prev_qry) / elapsed, 2)

        {idx_rate, qry_rate}
      else
        {nil, nil}
      end
    end
  end

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

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

      {:ok, %{status: 403}} ->
        {:error, "access denied — user may lack cluster:monitor/* permissions"}

      {:ok, %{status: status, body: body}} ->
        reason =
          cond do
            is_map(body) ->
              err = body["error"]
              (is_map(err) && err["reason"]) || (is_binary(err) && err) || body["message"] || "HTTP #{status}"
            is_binary(body) ->
              body
            true ->
              "HTTP #{status}"
          end
        {:error, "API error on #{path}: #{reason}"}

      {:error, %{reason: :econnrefused}} ->
        {:error, "connection refused — verify host, port, and that the node is running"}

      {:error, %{reason: :timeout}} ->
        {:error, "request timed out after #{Keyword.get(req_opts, :receive_timeout, @default_timeout_ms)} ms"}

      {:error, reason} ->
        {:error, format_error(reason)}
    end
  end

  defp add_auth(opts, nil, _password), do: opts
  defp add_auth(opts, user, password),  do: Keyword.put(opts, :auth, {user, password})

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

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

  defp format_error(%{reason: reason}), do: inspect(reason)
  defp format_error(reason),            do: inspect(reason)

  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

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