feat: collect payload size metric (#1494)

Also skip flaky test for now

Author: edgurgel

lib/extensions/postgres_cdc_rls/replication_poller.ex

@@ -183,7 +183,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
         change <- columns |> Enum.zip(row) |> generate_record() |> List.wrap() do
       topic = "realtime:postgres:" <> tenant_id
 
-      RealtimeWeb.TenantBroadcaster.pubsub_broadcast(topic, change, MessageDispatcher)
+      RealtimeWeb.TenantBroadcaster.pubsub_broadcast(tenant_id, topic, change, MessageDispatcher)
     end
 
     {:ok, rows_count}

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

@@ -22,10 +22,30 @@ defmodule Realtime.PromEx.Plugins.Tenant do
     [
       channel_events(),
       replication_metrics(),
-      subscription_metrics()
+      subscription_metrics(),
+      payload_size_metrics()
     ]
   end
 
+  defp payload_size_metrics do
+    Event.build(
+      :realtime_tenant_payload_size_metrics,
+      [
+        distribution(
+          [:realtime, :tenants, :payload, :size],
+          event_name: [:realtime, :tenants, :payload, :size],
+          measurement: :size,
+          description: "Payload size",
+          tags: [:tenant],
+          unit: :byte,
+          reporter_options: [
+            buckets: [100, 250, 500, 1000, 2000, 3000, 5000]
+          ]
+        )
+      ]
+    )
+  end
+
   defp concurrent_connections(poll_rate) do
     Polling.build(
       :realtime_concurrent_connections,

lib/realtime/tenants/batch_broadcast.ex

@@ -119,7 +119,7 @@ defmodule Realtime.Tenants.BatchBroadcast do
     broadcast = %Phoenix.Socket.Broadcast{topic: topic, event: @event_type, payload: payload}
 
     GenCounter.add(events_per_second_rate.id)
-    TenantBroadcaster.pubsub_broadcast(tenant_topic, broadcast, RealtimeChannel.MessageDispatcher)
+    TenantBroadcaster.pubsub_broadcast(tenant.external_id, tenant_topic, broadcast, RealtimeChannel.MessageDispatcher)
   end
 
   defp permissions_for_message(_, nil, _), do: nil

lib/realtime_web/channels/realtime_channel/broadcast_handler.ex

@@ -25,7 +25,8 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandler do
         self_broadcast: self_broadcast,
         tenant_topic: tenant_topic,
         authorization_context: authorization_context,
-        policies: policies
+        policies: policies,
+        tenant: tenant_id
       }
     } = socket
 
@@ -37,7 +38,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandler do
           |> increment_rate_counter()
 
         %{ack_broadcast: ack_broadcast} = socket.assigns
-        send_message(self_broadcast, tenant_topic, payload)
+        send_message(tenant_id, self_broadcast, tenant_topic, payload)
         if ack_broadcast, do: {:reply, :ok, socket}, else: {:noreply, socket}
 
       {:ok, policies} ->
@@ -54,23 +55,31 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandler do
   end
 
   def handle(payload, _db_conn, %{assigns: %{private?: false}} = socket) do
-    %{assigns: %{tenant_topic: tenant_topic, self_broadcast: self_broadcast, ack_broadcast: ack_broadcast}} = socket
+    %{
+      assigns: %{
+        tenant_topic: tenant_topic,
+        self_broadcast: self_broadcast,
+        ack_broadcast: ack_broadcast,
+        tenant: tenant_id
+      }
+    } = socket
 
     socket = increment_rate_counter(socket)
-    send_message(self_broadcast, tenant_topic, payload)
+    send_message(tenant_id, self_broadcast, tenant_topic, payload)
 
     if ack_broadcast,
       do: {:reply, :ok, socket},
       else: {:noreply, socket}
   end
 
-  defp send_message(self_broadcast, tenant_topic, payload) do
+  defp send_message(tenant_id, self_broadcast, tenant_topic, payload) do
     broadcast = %Phoenix.Socket.Broadcast{topic: tenant_topic, event: @event_type, payload: payload}
 
     if self_broadcast do
-      TenantBroadcaster.pubsub_broadcast(tenant_topic, broadcast, RealtimeChannel.MessageDispatcher)
+      TenantBroadcaster.pubsub_broadcast(tenant_id, tenant_topic, broadcast, RealtimeChannel.MessageDispatcher)
     else
       TenantBroadcaster.pubsub_broadcast_from(
+        tenant_id,
         self(),
         tenant_topic,
         broadcast,

lib/realtime_web/tenant_broadcaster.ex

@@ -3,30 +3,28 @@ defmodule RealtimeWeb.TenantBroadcaster do
   gen_rpc broadcaster
   """
 
-  alias Phoenix.Endpoint
   alias Phoenix.PubSub
 
-  @spec broadcast(Endpoint.topic(), Endpoint.event(), Endpoint.msg()) :: :ok
-  def broadcast(topic, event, msg) do
-    Realtime.GenRpc.multicast(RealtimeWeb.Endpoint, :local_broadcast, [topic, event, msg], key: topic)
-    :ok
-  end
-
-  @spec broadcast_from(from :: pid, Endpoint.topic(), Endpoint.event(), Endpoint.msg()) :: :ok
-  def broadcast_from(from, topic, event, msg) do
-    Realtime.GenRpc.multicast(RealtimeWeb.Endpoint, :local_broadcast_from, [from, topic, event, msg], key: topic)
-    :ok
-  end
+  @spec pubsub_broadcast(tenant_id :: String.t(), PubSub.topic(), PubSub.message(), PubSub.dispatcher()) :: :ok
+  def pubsub_broadcast(tenant_id, topic, message, dispatcher) do
+    collect_payload_size(tenant_id, message)
 
-  @spec pubsub_broadcast(PubSub.topic(), PubSub.message(), PubSub.dispatcher()) :: :ok
-  def pubsub_broadcast(topic, message, dispatcher) do
     Realtime.GenRpc.multicast(PubSub, :local_broadcast, [Realtime.PubSub, topic, message, dispatcher], key: topic)
 
     :ok
   end
 
-  @spec pubsub_broadcast_from(from :: pid, PubSub.topic(), PubSub.message(), PubSub.dispatcher()) :: :ok
-  def pubsub_broadcast_from(from, topic, message, dispatcher) do
+  @spec pubsub_broadcast_from(
+          tenant_id :: String.t(),
+          from :: pid,
+          PubSub.topic(),
+          PubSub.message(),
+          PubSub.dispatcher()
+        ) ::
+          :ok
+  def pubsub_broadcast_from(tenant_id, from, topic, message, dispatcher) do
+    collect_payload_size(tenant_id, message)
+
     Realtime.GenRpc.multicast(
       PubSub,
       :local_broadcast_from,
@@ -36,4 +34,15 @@ defmodule RealtimeWeb.TenantBroadcaster do
 
     :ok
   end
+
+  @payload_size_event [:realtime, :tenants, :payload, :size]
+
+  defp collect_payload_size(tenant_id, payload) when is_struct(payload) do
+    # Extracting from struct so the __struct__ bit is not calculated as part of the payload
+    collect_payload_size(tenant_id, Map.from_struct(payload))
+  end
+
+  defp collect_payload_size(tenant_id, payload) do
+    :telemetry.execute(@payload_size_event, %{size: :erlang.external_size(payload)}, %{tenant: tenant_id})
+  end
 end

mix.exs

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

test/realtime/extensions/cdc_rls/cdc_rls_test.exs

@@ -89,6 +89,7 @@ defmodule Realtime.Extensions.CdcRlsTest do
       %{tenant: tenant}
     end
 
+    @tag skip: "Flaky test. When logger handle_sasl_reports is enabled this test doesn't break"
     test "Check supervisor crash and respawn", %{tenant: tenant} do
       sup =
         Enum.reduce_while(1..30, nil, fn _, acc ->

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

@@ -226,6 +226,26 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
 
       assert metric_value(bucket_pattern) > 0
     end
+
+    test "metric payload size", context do
+      external_id = context.tenant.external_id
+
+      pattern =
+        ~r/realtime_tenants_payload_size_count{tenant="#{external_id}"}\s(?<number>\d+)/
+
+      metric_value = metric_value(pattern)
+
+      message = %{topic: "a topic", event: "an event", payload: ["a", %{"b" => "c"}, 1, 23]}
+      RealtimeWeb.TenantBroadcaster.pubsub_broadcast(external_id, "a topic", message, Phoenix.PubSub)
+
+      Process.sleep(200)
+      assert metric_value(pattern) == metric_value + 1
+
+      bucket_pattern =
+        ~r/realtime_tenants_payload_size_bucket{tenant="#{external_id}",le="100"}\s(?<number>\d+)/
+
+      assert metric_value(bucket_pattern) > 0
+    end
   end
 
   defp metric_value(pattern) do

test/realtime_web/controllers/broadcast_controller_test.exs

@@ -272,7 +272,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
     } do
       request_events_key = Tenants.requests_per_second_key(tenant)
       broadcast_events_key = Tenants.events_per_second_key(tenant)
-      expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _ -> :ok end)
+      expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _, _ -> :ok end)
 
       messages_to_send =
         Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
@@ -294,7 +294,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
 
-      broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/3)
+      broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/4)
 
       Enum.each(messages_to_send, fn %{topic: topic} ->
         broadcast_topic = Tenants.tenant_topic(tenant, topic, false)
@@ -310,7 +310,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
         }
 
         assert Enum.any?(broadcast_calls, fn
-                 [^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+                 [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
                  _ -> false
                end)
       end)
@@ -326,7 +326,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
     } do
       request_events_key = Tenants.requests_per_second_key(tenant)
       broadcast_events_key = Tenants.events_per_second_key(tenant)
-      expect(TenantBroadcaster, :pubsub_broadcast, 6, fn _, _, _ -> :ok end)
+      expect(TenantBroadcaster, :pubsub_broadcast, 6, fn _, _, _, _ -> :ok end)
 
       channels =
         Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
@@ -358,7 +358,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
 
-      broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/3)
+      broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/4)
 
       Enum.each(channels, fn %{topic: topic} ->
         broadcast_topic = Tenants.tenant_topic(tenant, topic, false)
@@ -374,7 +374,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
         }
 
         assert Enum.count(broadcast_calls, fn
-                 [^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+                 [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
                  _ -> false
                end) == 1
       end)
@@ -393,7 +393,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
       open_channel_topic = Tenants.tenant_topic(tenant, "open_channel", true)
 
       assert Enum.count(broadcast_calls, fn
-               [^open_channel_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+               [_, ^open_channel_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
                _ -> false
              end) == 1
 
@@ -408,7 +408,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
     } do
       request_events_key = Tenants.requests_per_second_key(tenant)
       broadcast_events_key = Tenants.events_per_second_key(tenant)
-      expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _ -> :ok end)
+      expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _, _ -> :ok end)
 
       messages_to_send =
         Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
@@ -432,7 +432,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
 
-      broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/3)
+      broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/4)
 
       Enum.each(messages_to_send, fn %{topic: topic} ->
         broadcast_topic = Tenants.tenant_topic(tenant, topic, false)
@@ -448,7 +448,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
         }
 
         assert Enum.count(broadcast_calls, fn
-                 [^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+                 [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
                  _ -> false
                end) == 1
       end)
@@ -461,7 +461,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
     @tag role: "anon"
     test "user without permission won't broadcast", %{conn: conn, db_conn: db_conn, tenant: tenant} do
       request_events_key = Tenants.requests_per_second_key(tenant)
-      reject(&TenantBroadcaster.broadcast/3)
+      reject(&TenantBroadcaster.pubsub_broadcast/4)
 
       messages =
         Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)