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 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) {: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 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, %Metrics.Counter{} = metric}, _from, state) do with {:ok, handler_id} <- Counter.register(metric, state.config.aggregates_table_id, self()) do {:reply, :ok, %{state | metrics: [{metric, handler_id} | state.metrics]}} else {:error, :already_exists} -> {:reply, {:error, :already_exists, metric.name}, state} end end def handle_call({:register, %Metrics.LastValue{} = metric}, _from, state) do with {:ok, handler_id} <- LastValue.register(metric, state.config.aggregates_table_id, self()) do {:reply, :ok, %{state | metrics: [{metric, handler_id} | state.metrics]}} else {:error, :already_exists} -> {:reply, {:error, :already_exists, metric.name}, state} end end def handle_call({:register, %Metrics.Sum{} = metric}, _from, state) do with {:ok, handler_id} <- Sum.register(metric, state.config.aggregates_table_id, self()) do {:reply, :ok, %{state | metrics: [{metric, handler_id} | state.metrics]}} else {:error, :already_exists} -> {:reply, {:error, :already_exists, metric.name}, state} end end def handle_call({:register, %Metrics.Distribution{} = metric}, _from, state) do with {:ok, handler_id} <- Distribution.register(metric, state.config.dist_table_id, self()) do {:reply, :ok, %{ state | metrics: [ {%{metric | buckets: metric.buckets ++ ["+Inf"]}, handler_id} | state.metrics ] }} else {:error, :already_exists} -> {:reply, {:error, :already_exists, metric.name}, state} end end def handle_call({:register, %Metrics.Summary{} = _metric}, _from, state) do {:reply, {:error, :unsupported_metric_type, :summary}, state} 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()) :: DynamicSupervisor.on_start_child() def monitor_tables(tables, name) do measurement_specs = Enum.map(tables, &{Core.Telemetry, :dispatch_table_stats, [&1]}) DynamicSupervisor.start_child( TelemetryMetricsPrometheus.Core.DynamicSupervisor, {:telemetry_poller, [measurements: measurement_specs, name: String.to_atom("#{name}_poller")]} ) end end