fix: improve metrics handling (#1654)

* Allow 3x more max heap size than other processes
* Avoid compressing given that gen_rpc does this for us already
* fix: remove limit_concurrent metric
* fix: revert the way we fetched tenants to report on connected metric

Author: edgurgel

lib/realtime/monitoring/prom_ex.ex

@@ -134,11 +134,4 @@ defmodule Realtime.PromEx do
 
     metrics
   end
-
-  @doc "Compressed metrics using :zlib.compress/1"
-  @spec get_compressed_metrics() :: binary()
-  def get_compressed_metrics do
-    get_metrics()
-    |> :zlib.compress()
-  end
 end

lib/realtime/monitoring/prom_ex/plugins/tenant.ex

@@ -22,7 +22,6 @@ defmodule Realtime.PromEx.Plugins.Tenant do
     [
       channel_events(),
       replication_metrics(),
-      subscription_metrics(),
       payload_size_metrics()
     ]
   end
@@ -78,21 +77,14 @@ defmodule Realtime.PromEx.Plugins.Tenant do
           description: "The cluster total count of connected clients for a tenant.",
           measurement: :connected_cluster,
           tags: [:tenant]
-        ),
-        last_value(
-          [:realtime, :connections, :limit_concurrent],
-          event_name: [:realtime, :connections],
-          description: "The total count of connected clients for a tenant.",
-          measurement: :limit,
-          tags: [:tenant]
         )
       ],
       detach_on_error: false
     )
   end
 
   def execute_tenant_metrics do
-    tenants = Tenants.Connect.list_tenants()
+    tenants = Tenants.list_connected_tenants(Node.self())
     cluster_counts = UsersCounter.tenant_counts()
     node_counts = UsersCounter.tenant_counts(Node.self())
 
@@ -136,28 +128,6 @@ defmodule Realtime.PromEx.Plugins.Tenant do
     )
   end
 
-  defp subscription_metrics do
-    Event.build(
-      :realtime_tenant_channel_event_metrics,
-      [
-        sum(
-          [:realtime, :subscriptions_checker, :pid_not_found],
-          event_name: [:realtime, :subscriptions_checker, :pid_not_found],
-          measurement: :sum,
-          description: "Sum of pids not found in Subscription tables.",
-          tags: [:tenant]
-        ),
-        sum(
-          [:realtime, :subscriptions_checker, :phantom_pid_detected],
-          event_name: [:realtime, :subscriptions_checker, :phantom_pid_detected],
-          measurement: :sum,
-          description: "Sum of phantom pids detected in Subscription tables.",
-          tags: [:tenant]
-        )
-      ]
-    )
-  end
-
   defmodule PolicyAuthorization.Buckets do
     @moduledoc false
     use Peep.Buckets.Custom, buckets: [10, 250, 5000, 15_000]
@@ -237,20 +207,6 @@ defmodule Realtime.PromEx.Plugins.Tenant do
           measurement: :size,
           tags: [:tenant]
         ),
-        last_value(
-          [:realtime, :channel, :events, :limit_per_second],
-          event_name: [:realtime, :rate_counter, :channel, :events],
-          measurement: :limit,
-          description: "Rate limit of messages per second sent on a Realtime Channel.",
-          tags: [:tenant]
-        ),
-        last_value(
-          [:realtime, :channel, :joins, :limit_per_second],
-          event_name: [:realtime, :rate_counter, :channel, :joins],
-          measurement: :limit,
-          description: "Rate limit of joins per second on a Realtime Channel.",
-          tags: [:tenant]
-        ),
         distribution(
           [:realtime, :tenants, :read_authorization_check],
           event_name: [:realtime, :tenants, :read_authorization_check],

lib/realtime/tenants.ex

@@ -15,6 +15,15 @@ defmodule Realtime.Tenants do
   alias Realtime.Tenants.Migrations
   alias Realtime.UsersCounter
 
+  @doc """
+  Gets a list of connected tenant `external_id` strings in the cluster or a node.
+  """
+  @spec list_connected_tenants(atom()) :: [String.t()]
+  def list_connected_tenants(node) do
+    UsersCounter.scopes()
+    |> Enum.flat_map(fn scope -> :syn.group_names(scope, node) end)
+  end
+
   @doc """
   Gets the database connection pid managed by the Tenants.Connect process.
 

lib/realtime_web/controllers/metrics_controller.ex

@@ -4,40 +4,50 @@ defmodule RealtimeWeb.MetricsController do
   alias Realtime.PromEx
   alias Realtime.GenRpc
 
+  # We give more memory and time to collect metrics from all nodes as this is a lot of work
   def index(conn, _) do
-    timeout = Application.fetch_env!(:realtime, :metrics_rpc_timeout)
-
-    cluster_metrics =
-      Node.list()
-      |> Task.async_stream(
-        fn node ->
-          {node, GenRpc.call(node, PromEx, :get_compressed_metrics, [], timeout: timeout)}
-        end,
-        timeout: :infinity
-      )
-      |> Enum.reduce([PromEx.get_metrics()], fn {_, {node, response}}, acc ->
-        case response do
-          {:error, :rpc_error, reason} ->
-            Logger.error("Cannot fetch metrics from the node #{inspect(node)} because #{inspect(reason)}")
-            acc
-
-          metrics ->
-            [uncompress(metrics) | acc]
-        end
-      end)
-      |> Enum.reverse()
+    {time, metrics} = :timer.tc(&cluster_metrics/0, :millisecond)
+    Logger.info("Collected cluster metrics in #{time} milliseconds")
 
     conn
     |> put_resp_content_type("text/plain")
-    |> send_resp(200, cluster_metrics)
+    |> send_resp(200, metrics)
+  end
+
+  defp cluster_metrics() do
+    bump_max_heap_size()
+    timeout = Application.fetch_env!(:realtime, :metrics_rpc_timeout)
+
+    Node.list()
+    |> Task.async_stream(
+      fn node ->
+        {node, GenRpc.call(node, __MODULE__, :get_metrics, [], timeout: timeout)}
+      end,
+      timeout: :infinity
+    )
+    |> Enum.reduce([PromEx.get_metrics()], fn {_, {node, response}}, acc ->
+      case response do
+        {:error, :rpc_error, reason} ->
+          Logger.error("Cannot fetch metrics from the node #{inspect(node)} because #{inspect(reason)}")
+          acc
+
+        metrics ->
+          [metrics | acc]
+      end
+    end)
   end
 
-  defp uncompress(compressed_data) do
-    :zlib.uncompress(compressed_data)
-  rescue
-    error ->
-      Logger.error("Failed to decompress metrics data: #{inspect(error)}")
-      # Return empty string to not impact the aggregated metrics
-      ""
+  def get_metrics() do
+    bump_max_heap_size()
+    PromEx.get_metrics()
+  end
+
+  defp bump_max_heap_size() do
+    system_max_heap_size = :erlang.system_info(:max_heap_size)[:size]
+
+    # it's 0 when there is no limit
+    if is_integer(system_max_heap_size) and system_max_heap_size > 0 do
+      Process.flag(:max_heap_size, system_max_heap_size * 3)
+    end
   end
 end

mix.exs

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

test/integration/rt_channel_test.exs

@@ -2163,7 +2163,7 @@ defmodule Realtime.Integration.RtChannelTest do
       # Postgres Change events
       for _ <- 1..5, do: Postgrex.query!(conn, "insert into test (details) values ('test')", [])
 
-      for _ <- 1..5 do
+      for _ <- 1..10 do
         assert_receive %Message{
                          topic: ^topic,
                          event: "postgres_changes",

test/realtime/monitoring/prom_ex/plugins/tenant_test.exs

@@ -105,20 +105,12 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
 
       _ = Rpc.call(node, FakeUserCounter, :fake_add, [external_id])
 
-      # fake empty tenant_id
-      empty_tenant = tenant_fixture()
-      empty_tenant_id = empty_tenant.external_id
-      :syn.register(Realtime.Tenants.Connect, empty_tenant_id, self(), %{conn: nil})
-
       Process.sleep(500)
       Tenant.execute_tenant_metrics()
 
       assert_receive {[:realtime, :connections], %{connected: 1, limit: 200, connected_cluster: 2},
                       %{tenant: ^external_id}}
 
-      assert_receive {[:realtime, :connections], %{connected: 0, limit: 200, connected_cluster: 0},
-                      %{tenant: ^empty_tenant_id}}
-
       refute_receive {[:realtime, :connections], %{connected: 1, limit: 200, connected_cluster: 2},
                       %{tenant: ^bad_tenant_id}}
     end

test/realtime/monitoring/prom_ex_test.exs

@@ -20,24 +20,4 @@ defmodule Realtime.PromExTest do
              )
     end
   end
-
-  describe "get_compressed_metrics/0" do
-    test "builds metrics compressed using zlib" do
-      compressed_metrics = PromEx.get_compressed_metrics()
-
-      metrics = :zlib.uncompress(compressed_metrics)
-
-      assert String.contains?(
-               metrics,
-               "# HELP beam_system_schedulers_online_info The number of scheduler threads that are online."
-             )
-
-      assert String.contains?(metrics, "# TYPE beam_system_schedulers_online_info gauge")
-
-      assert String.contains?(
-               metrics,
-               "beam_system_schedulers_online_info{host=\"nohost\",id=\"nohost\",region=\"us-east-1\"}"
-             )
-    end
-  end
 end