telemetry_metrics_prometheus_core

defmodule TelemetryMetricsPrometheus.Core.Registry do
  @moduledoc false
  use GenServer

  require Logger

  alias Telemetry.Metrics
  alias TelemetryMetricsPrometheus.Core
  alias TelemetryMetricsPrometheus.Core.{Counter, Distribution, LastValue, Sum}

  @type name :: atom()
  @type metric_exists_error() :: {:error, :already_exists, Core.metric()}
  @type unsupported_metric_type_error() :: {:error, :unsupported_metric_type, :summary}
  @type validation_opts() :: [consistent_units: bool(), require_seconds: bool()]

  # metric_name should be the validated and normalized prometheus
  # name - https://prometheus.io/docs/instrumenting/writing_exporters/#naming

  def start_link(opts) do
    case Keyword.get(opts, :metrics) do
      nil -> raise "no :metrics key defined in options"
      _ -> :ok
    end

    GenServer.start_link(__MODULE__, opts, name: opts[:name])
  end

  @impl true
  def init(opts) do
    name = opts[:name]
    aggregates_table_id = create_table(name, :set)
    dist_table_id = create_table(String.to_atom("#{name}_dist"), :duplicate_bag)

    send(self(), {:setup, opts})

    {:ok,
     %{
       config: %{aggregates_table_id: aggregates_table_id, dist_table_id: dist_table_id},
       metrics: []
     }}
  end

  @spec register(Core.metric(), atom()) ::
          :ok | metric_exists_error() | unsupported_metric_type_error()
  def register(metric, name \\ __MODULE__) do
    # validate metrics units ?

    GenServer.call(name, {:register, metric})
  end

  @spec validate_units(Core.metrics(), validation_opts()) ::
          Core.metrics()
  def validate_units(metrics, opts) do
    time_units =
      metrics
      |> Enum.filter(&match?(%Metrics.Distribution{}, &1))
      |> Enum.reduce(MapSet.new([]), fn
        %{unit: {_from, to}}, acc -> MapSet.put(acc, to)
        %{unit: unit}, acc when is_atom(unit) -> MapSet.put(acc, unit)
      end)
      |> MapSet.to_list()

    validate_consistent_units(time_units, opts[:consistent_units])
    validate_units_seconds(time_units, opts[:require_seconds])

    metrics
  end

  @spec validate_consistent_units([Metrics.time_unit()], bool()) :: :ok
  defp validate_consistent_units(_, false), do: :ok

  defp validate_consistent_units(units, true) when length(units) > 1 do
    Logger.warn(
      "Multiple time units found in your Telemetry.Metrics definitions.\n\nPrometheus recommends using consistent time units to make view creation simpler.\n\nYou can disable this validation check by adding `consistent_units: false` in the validations options on reporter init."
    )

    :ok
  end

  defp validate_consistent_units(_units, _), do: :ok

  @spec validate_units_seconds([Metrics.time_unit()], bool()) :: :ok
  defp validate_units_seconds(_, false), do: :ok
  defp validate_units_seconds([:second], _), do: :ok
  defp validate_units_seconds([], _), do: :ok

  defp validate_units_seconds(_, _) do
    Logger.warn(
      "Prometheus requires that time units MUST only be offered in seconds according to their guidelines, though this is not always practical.\n\nhttps://prometheus.io/docs/instrumenting/writing_clientlibs/#histogram.\n\nYou can disable this validation check by adding `require_seconds: false` in the validations options on reporter init."
    )

    :ok
  end

  @spec config(name()) :: %{aggregates_table_id: atom(), dist_table_id: atom()}
  def config(name) do
    GenServer.call(name, :get_config)
  end

  @spec metrics(name()) :: [{Core.metric(), :telemetry.handler_id()}]
  def metrics(name) do
    GenServer.call(name, :get_metrics)
  end

  @impl true
  def handle_info({:setup, opts}, state) do
    metrics = Keyword.get(opts, :metrics, [])
    registered = register_metrics(metrics, opts[:validations], state.config)

    if opts[:monitor_reporter] do
      monitor_tables([state.config.aggregates_table_id, state.config.dist_table_id], opts[:name])
    end

    {:noreply, %{state | metrics: registered}}
  end

  def handle_info(_, state) do
    {:noreply, state}
  end

  def handle_call(:get_config, _from, state) do
    {:reply, state.config, state}
  end

  def handle_call(:get_metrics, _from, state) do
    metrics = Enum.map(state.metrics, &elem(&1, 0))
    {:reply, metrics, state}
  end

  @impl true
  @spec handle_call({:register, Core.metric()}, GenServer.from(), map()) ::
          {:reply, :ok, map()}
          | {:reply, metric_exists_error() | unsupported_metric_type_error(), map()}
  def handle_call({:register, metric}, _from, state) do
    case register_metric(metric, state.config) do
      {:ok, metric} -> {:reply, :ok, %{state | metrics: [metric | state.metrics]}}
      other -> {:reply, other, state}
    end
  end

  @spec create_table(name :: atom, type :: atom) :: :ets.tid() | atom
  defp create_table(name, type) do
    :ets.new(name, [:named_table, :public, type, {:write_concurrency, true}])
  end

  @spec monitor_tables([atom()], atom()) :: Supervisor.on_start()
  def monitor_tables(tables, name) do
    measurement_specs = Enum.map(tables, &{Core.Telemetry, :dispatch_table_stats, [&1]})

    Core.ReporterSupervisor.start_link(name: name, measurements: measurement_specs)
  end

  @spec register_metrics(
          Core.metrics(),
          validation_opts(),
          %{}
        ) :: :ok
  defp register_metrics(metrics, validations, config) do
    metrics
    |> validate_units(validations)
    |> Enum.reduce([], fn metric, acc ->
      case register_metric(metric, config) do
        {:ok, metric} ->
          [metric | acc]

        {:error, :already_exists, metric_name} ->
          Logger.warn(
            "Metric name already exists. Dropping measure. metric_name:=#{inspect(metric_name)}"
          )

          acc

        {:error, :unsupported_metric_type, metric_type} ->
          Logger.warn(
            "Metric type #{metric_type} is unsupported. Dropping measure. metric_name:=#{
              inspect(metric.name)
            }"
          )

          acc
      end
    end)
  end

  defp register_metric(%Metrics.Counter{} = metric, config) do
    case Counter.register(metric, config.aggregates_table_id, self()) do
      {:ok, handler_id} -> {:ok, {metric, handler_id}}
      {:error, :already_exists} -> {:error, :already_exists, metric.name}
    end
  end

  defp register_metric(%Metrics.LastValue{} = metric, config) do
    case LastValue.register(metric, config.aggregates_table_id, self()) do
      {:ok, handler_id} -> {:ok, {metric, handler_id}}
      {:error, :already_exists} -> {:error, :already_exists, metric.name}
    end
  end

  defp register_metric(%Metrics.Sum{} = metric, config) do
    case Sum.register(metric, config.aggregates_table_id, self()) do
      {:ok, handler_id} -> {:ok, {metric, handler_id}}
      {:error, :already_exists} -> {:error, :already_exists, metric.name}
    end
  end

  defp register_metric(%Metrics.Distribution{} = metric, config) do
    case Distribution.register(metric, config.dist_table_id, self()) do
      {:ok, handler_id} -> {:ok, {%{metric | buckets: metric.buckets ++ ["+Inf"]}, handler_id}}
      {:error, :already_exists} -> {:error, :already_exists, metric.name}
    end
  end

  defp register_metric(%Metrics.Summary{}, _config) do
    {:error, :unsupported_metric_type, :summary}
  end
end