fix: Metrics cleanup based on syn module (#1233)

Adds a new worker that will cleanup metrics based on our syn information to reduce the amount of data pulled by Prometheus.

config/runtime.exs

@@ -22,6 +22,8 @@ config :realtime,
     System.get_env("TENANT_MAX_EVENTS_PER_SECOND", "100") |> String.to_integer(),
   tenant_max_joins_per_second:
     System.get_env("TENANT_MAX_JOINS_PER_SECOND", "100") |> String.to_integer(),
+  metrics_cleaner_schedule_timer_in_ms:
+    System.get_env("METRICS_CLEANER_SCHEDULE_TIMER_IN_MS", "1800000") |> String.to_integer(),
   rpc_timeout: System.get_env("RPC_TIMEOUT", "30000") |> String.to_integer()
 
 run_janitor? = System.get_env("RUN_JANITOR", "false") == "true"

lib/realtime/application.ex

@@ -84,7 +84,8 @@ defmodule Realtime.Application do
          strategy: :one_for_one,
          name: Realtime.BroadcastChanges.Handler.DynamicSupervisor},
         RealtimeWeb.Endpoint,
-        RealtimeWeb.Presence
+        RealtimeWeb.Presence,
+        Realtime.MetricsCleaner
       ] ++ extensions_supervisors() ++ janitor_tasks()
 
     children =

lib/realtime/metrics_cleaner.ex

@@ -0,0 +1,60 @@
+defmodule Realtime.MetricsCleaner do
+  @moduledoc false
+
+  use GenServer
+  require Logger
+
+  defstruct [:check_ref, :interval]
+
+  def start_link(args),
+    do: GenServer.start_link(__MODULE__, args, name: __MODULE__)
+
+  def init(_args) do
+    interval = Application.get_env(:realtime, :metrics_cleaner_schedule_timer_in_ms)
+
+    Logger.info("Starting MetricsCleaner")
+    {:ok, %{check_ref: check(interval), interval: interval}}
+  end
+
+  def handle_info(:check, %{interval: interval} = state) do
+    Process.cancel_timer(state.check_ref)
+
+    {exec_time, _} = :timer.tc(fn -> loop_and_cleanup_metrics_table() end)
+
+    if exec_time > :timer.seconds(5),
+      do: Logger.warning("Metrics check took: #{exec_time} ms")
+
+    {:noreply, %{state | check_ref: check(interval)}}
+  end
+
+  def handle_info(msg, state) do
+    Logger.error("Unexpected message: #{inspect(msg)}")
+    {:noreply, state}
+  end
+
+  defp check(interval) do
+    Process.send_after(self(), :check, interval)
+  end
+
+  @table_name :"syn_registry_by_name_Elixir.Realtime.Tenants.Connect"
+  @metrics_table Realtime.PromEx.Metrics
+  @filter_spec [{{{:_, %{tenant: :"$1"}}, :_}, [], [:"$1"]}]
+  @tenant_id_spec [{{:"$1", :_, :_, :_, :_, :_}, [], [:"$1"]}]
+  defp loop_and_cleanup_metrics_table do
+    tenant_ids = :ets.select(@table_name, @tenant_id_spec)
+
+    :ets.select(@metrics_table, @filter_spec)
+    |> Enum.uniq()
+    |> Enum.reject(fn tenant_id -> tenant_id in tenant_ids end)
+    |> Enum.each(fn tenant_id -> delete_metric(tenant_id) end)
+  end
+
+  @doc """
+  Deletes all metrics that contain the given tenant.
+  """
+  @spec delete_metric(String.t()) :: :ok
+  def delete_metric(tenant) do
+    :ets.select_delete(@metrics_table, [{{{:_, %{tenant: tenant}}, :_}, [], [true]}])
+    :ok
+  end
+end

lib/realtime/tenants/connect.ex

@@ -248,6 +248,12 @@ defmodule Realtime.Tenants.Connect do
     {:stop, :kill, state}
   end
 
+  @impl true
+  def terminate(_, %{tenant_id: tenant_id}) do
+    Realtime.MetricsCleaner.delete_metric(tenant_id)
+    :ok
+  end
+
   ## Private functions
 
   defp call_external_node(tenant_id, opts) do

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.33.55",
+      version: "2.33.56",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,