fix: speed up tenant connected metrics

Author: edgurgel

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

@@ -93,16 +93,20 @@ defmodule Realtime.PromEx.Plugins.Tenant do
 
   def execute_tenant_metrics do
     tenants = Tenants.Connect.list_tenants()
+    cluster_counts = UsersCounter.tenant_counts()
+    node_counts = UsersCounter.tenant_counts(Node.self())
 
     for t <- tenants do
-      count = UsersCounter.tenant_users(Node.self(), t)
-      cluster_count = UsersCounter.tenant_users(t)
       tenant = Tenants.Cache.get_tenant_by_external_id(t)
 
       if tenant != nil do
         Telemetry.execute(
           [:realtime, :connections],
-          %{connected: count, connected_cluster: cluster_count, limit: tenant.max_concurrent_users},
+          %{
+            connected: Map.get(node_counts, t, 0),
+            connected_cluster: Map.get(cluster_counts, t, 0),
+            limit: tenant.max_concurrent_users
+          },
           %{tenant: t}
         )
       end

lib/realtime/user_counter.ex

@@ -22,6 +22,54 @@ defmodule Realtime.UsersCounter do
   @spec tenant_users(atom, String.t()) :: non_neg_integer()
   def tenant_users(node_name, tenant_id), do: tenant_id |> scope() |> :syn.member_count(tenant_id, node_name)
 
+  @count_all_nodes_spec [
+    {
+      # Match the tuple structure, capture group_name
+      {{:"$1", :_}, :_, :_, :_, :_},
+      # No guards
+      [],
+      # Return only the group_name
+      [:"$1"]
+    }
+  ]
+
+  @doc """
+  Returns the counts of all connected clients for all tenants for the cluster.
+  """
+  @spec tenant_counts() :: %{String.t() => non_neg_integer()}
+  def tenant_counts() do
+    scopes()
+    |> Stream.flat_map(fn scope ->
+      :syn_backbone.get_table_name(:syn_pg_by_name, scope)
+      |> :ets.select(@count_all_nodes_spec)
+    end)
+    |> Enum.frequencies()
+  end
+
+  @doc """
+  Returns the counts of all connected clients for all tenants for a single node.
+  """
+  @spec tenant_counts(node) :: %{String.t() => non_neg_integer()}
+  def tenant_counts(node) do
+    count_single_node_spec = [
+      {
+        # Match the tuple structure with specific node, capture group_name
+        {{:"$1", :_}, :_, :_, :_, node},
+        # No guards
+        [],
+        # Return only the group_name
+        [:"$1"]
+      }
+    ]
+
+    scopes()
+    |> Stream.flat_map(fn scope ->
+      :syn_backbone.get_table_name(:syn_pg_by_name, scope)
+      |> :ets.select(count_single_node_spec)
+    end)
+    |> Enum.frequencies()
+  end
+
   @doc """
   Returns the scope for a given tenant id.
   """

mix.exs

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

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

@@ -104,12 +104,21 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
       UsersCounter.add(self(), bad_tenant_id)
 
       _ = 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/user_counter_test.exs

@@ -3,6 +3,12 @@ defmodule Realtime.UsersCounterTest do
   alias Realtime.UsersCounter
   alias Realtime.Rpc
 
+  setup_all do
+    tenant_id = random_string()
+    {nodes, count} = generate_load(tenant_id)
+    %{tenant_id: tenant_id, count: count, nodes: nodes}
+  end
+
   describe "add/1" do
     test "starts counter for tenant" do
       assert UsersCounter.add(self(), random_string()) == :ok
@@ -14,61 +20,88 @@ defmodule Realtime.UsersCounterTest do
                 def ping(),
                   do:
                     spawn(fn ->
-                      Process.sleep(3000)
+                      Process.sleep(15000)
                       :pong
                     end)
               end
             end)
 
   Code.eval_quoted(@aux_mod)
 
+  describe "tenant_counts/0" do
+    test "map of tenant and number of users", %{tenant_id: tenant_id, count: expected} do
+      assert UsersCounter.add(self(), tenant_id) == :ok
+      Process.sleep(1000)
+      counts = UsersCounter.tenant_counts()
+      assert counts[tenant_id] == expected + 1
+
+      assert map_size(counts) >= 41
+    end
+  end
+
+  describe "tenant_counts/1" do
+    test "map of tenant and number of users for a node only", %{tenant_id: tenant_id, nodes: nodes} do
+      assert UsersCounter.add(self(), tenant_id) == :ok
+      Process.sleep(1000)
+      my_counts = UsersCounter.tenant_counts(Node.self())
+      # Only one connection from this test process on this node
+      assert my_counts == %{tenant_id => 1}
+
+      another_node_counts = UsersCounter.tenant_counts(hd(nodes))
+      assert another_node_counts[tenant_id] == 2
+
+      assert map_size(another_node_counts) == 21
+    end
+  end
+
   describe "tenant_users/1" do
-    test "returns count of connected clients for tenant on cluster node" do
-      tenant_id = random_string()
-      expected = generate_load(tenant_id)
+    test "returns count of connected clients for tenant on cluster node", %{tenant_id: tenant_id, count: expected} do
       Process.sleep(1000)
       assert UsersCounter.tenant_users(tenant_id) == expected
     end
   end
 
   describe "tenant_users/2" do
-    test "returns count of connected clients for tenant on target cluster" do
-      tenant_id = random_string()
-      generate_load(tenant_id)
+    test "returns count of connected clients for tenant on target cluster", %{tenant_id: tenant_id} do
       {:ok, node} = Clustered.start(@aux_mod)
       pid = Rpc.call(node, Aux, :ping, [])
       UsersCounter.add(pid, tenant_id)
       assert UsersCounter.tenant_users(node, tenant_id) == 1
     end
   end
 
-  defp generate_load(tenant_id, nodes \\ 2, processes \\ 2) do
-    for i <- 1..nodes do
-      # Avoid port collision
-      extra_config = [
-        {:gen_rpc, :tcp_server_port, 15970 + i}
-      ]
-
-      {:ok, node} = Clustered.start(@aux_mod, extra_config: extra_config, phoenix_port: 4012 + i)
-
-      for _ <- 1..processes do
-        pid = Rpc.call(node, Aux, :ping, [])
-
-        for _ <- 1..10 do
-          # replicate same pid added multiple times concurrently
-          Task.start(fn ->
-            UsersCounter.add(pid, tenant_id)
-          end)
-
-          # noisy neighbors to test handling of bigger loads on concurrent calls
-          Task.start(fn ->
-            pid = Rpc.call(node, Aux, :ping, [])
-            UsersCounter.add(pid, random_string())
-          end)
+  defp generate_load(tenant_id, n_nodes \\ 2, processes \\ 2) do
+    nodes =
+      for i <- 1..n_nodes do
+        # Avoid port collision
+        extra_config = [
+          {:gen_rpc, :tcp_server_port, 15970 + i}
+        ]
+
+        {:ok, node} = Clustered.start(@aux_mod, extra_config: extra_config, phoenix_port: 4012 + i)
+
+        for _ <- 1..processes do
+          pid = Rpc.call(node, Aux, :ping, [])
+
+          for _ <- 1..10 do
+            # replicate same pid added multiple times concurrently
+            Task.start(fn ->
+              UsersCounter.add(pid, tenant_id)
+              Process.sleep(10000)
+            end)
+
+            # noisy neighbors to test handling of bigger loads on concurrent calls
+            Task.start(fn ->
+              pid = Rpc.call(node, Aux, :ping, [])
+              UsersCounter.add(pid, random_string())
+              Process.sleep(10000)
+            end)
+          end
         end
+
+        node
       end
-    end
 
-    nodes * processes
+    {nodes, n_nodes * processes}
   end
 end