fix: correctly cleanup metrics tables

Author: hauleth

lib/supavisor/metrics_cleaner.ex

@@ -5,9 +5,14 @@ defmodule Supavisor.MetricsCleaner do
   require Logger
 
   @interval :timer.minutes(30)
+  @name __MODULE__
 
   def start_link(args),
-    do: GenServer.start_link(__MODULE__, args, name: __MODULE__)
+    do: GenServer.start_link(__MODULE__, args, name: @name)
+
+  def clean do
+    GenServer.cast(@name, :clean)
+  end
 
   def init(_args) do
     Logger.info("Starting MetricsCleaner")
@@ -30,7 +35,7 @@ defmodule Supavisor.MetricsCleaner do
       do: Logger.warning("Metrics check took: #{exec_time} ms")
   end
 
-  def handle_info(:check, state) do
+  def handle_continue(:clean, state) do
     Process.cancel_timer(state.check_ref)
 
     :telemetry.span([:supavisor, :metrics_cleaner], %{}, fn ->
@@ -41,39 +46,54 @@ defmodule Supavisor.MetricsCleaner do
     {:noreply, %{state | check_ref: check()}}
   end
 
+  def handle_cast(:clean, state) do
+    {:noreply, state, {:continue, :clean}}
+  end
+
+  def handle_info(:check, state) do
+    {:noreply, state, {:continue, :clean}}
+  end
+
   def handle_info(msg, state) do
     Logger.error("Unexpected message: #{inspect(msg)}")
     {:noreply, state}
   end
 
-  def check, do: Process.send_after(self(), :check, @interval)
-
-  def loop_and_cleanup_metrics_table do
-    metrics_table = Supavisor.Monitoring.PromEx.Metrics
-    tenant_registry_table = :syn_registry_by_name_tenants
-
-    func = fn elem, acc ->
-      with {{_,
-             %{
-               type: type,
-               mode: mode,
-               user: user,
-               tenant: tenant,
-               db_name: db,
-               search_path: search_path
-             }} = key, _} <- elem,
-           [] <- :ets.lookup(tenant_registry_table, {{type, tenant}, user, mode, db, search_path}) do
-        Logger.warning("Found orphaned metric: #{inspect(key)}")
-        :ets.delete(metrics_table, key)
-
-        acc + 1
-      else
-        _ -> acc
-      end
-    end
+  defp check, do: Process.send_after(self(), :check, @interval)
 
+  defp loop_and_cleanup_metrics_table do
     {_, tids} = Peep.Persistent.storage(Supavisor.Monitoring.PromEx.Metrics)
 
-    Enum.each(List.wrap(tids), &:ets.foldl(func, 0, &1))
+    tids
+    |> List.wrap()
+    |> Enum.sum_by(&clean_table/1)
+  end
+
+  @tenant_registry_table :syn_registry_by_name_tenants
+
+  defp clean_table(tid) do
+    func =
+      fn elem, acc ->
+        with {{_,
+               %{
+                 type: type,
+                 mode: mode,
+                 user: user,
+                 tenant: tenant,
+                 db_name: db,
+                 search_path: search_path
+               }} = key, _} <- elem,
+             [] <-
+               :ets.lookup(@tenant_registry_table, {{type, tenant}, user, mode, db, search_path}) do
+          Logger.warning("Found orphaned metric: #{inspect(key)}")
+          :ets.delete(tid, key)
+
+          acc + 1
+        else
+          _ -> acc
+        end
+      end
+
+    :ets.foldl(func, 0, tid)
   end
 end

lib/supavisor/monitoring/tenant.ex

@@ -22,6 +22,7 @@ defmodule Supavisor.PromEx.Plugins.Tenant do
   @impl true
   def event_metrics(_opts) do
     [
+      system_metrics(),
       client_metrics(),
       db_metrics()
     ]
@@ -33,6 +34,19 @@ defmodule Supavisor.PromEx.Plugins.Tenant do
       buckets: [1, 5, 10, 100, 1_000, 5_000, 10_000]
   end
 
+  defp system_metrics do
+    Event.build(
+      :supavisor_metrics_cleaner_metrics,
+      [
+        counter(
+          [:supavisor, :metrics_cleaner, :orphaned_metrics],
+          event_name: [:supavisor, :metrics, :orphaned],
+          description: "Amount of orphaned metrics that were cleaned up"
+        )
+      ]
+    )
+  end
+
   defp client_metrics do
     Event.build(
       :supavisor_tenant_client_event_metrics,

test/supavisor/metrics_cleaner_test.exs

@@ -0,0 +1,34 @@
+defmodule Supavisor.MetricsCleanerTest do
+  use ExUnit.Case, async: false
+
+  alias Supavisor.PromEx.Plugins.Tenant, as: Metrics
+
+  @subject Supavisor.MetricsCleaner
+
+  doctest @subject
+
+  setup ctx do
+    :telemetry.attach(ctx, [:supavisor, :metrics_cleaner, :stop], &__MODULE__.handler/4, %{parent: self()})
+
+    :ok
+  end
+
+  def handler(_, measurements, _, %{parent: pid}) do
+    send(pid, {:metrics, measurements})
+  end
+
+  test "metrics for unknown tenant are removed" do
+    :ok = Metrics.emit_telemetry_for_tenant({{{:single, "non-existent"}, "foo", :transaction, "bar", nil}, 2137})
+    metrics = Supavisor.Monitoring.PromEx.get_metrics()
+
+    assert IO.iodata_to_binary(metrics) =~ ~r/non-existent/
+
+    @subject.clean()
+
+    assert_receive {:metrics, _}
+
+    metrics = Supavisor.Monitoring.PromEx.get_metrics()
+
+    refute IO.iodata_to_binary(metrics) =~ ~r/non-existent/
+  end
+end