fix: Instrument RPC calls (#766)

Instrument RPC calls

---------

Co-authored-by: Wen Bo Xie <[email protected]>

Author: filipecabaco

lib/extensions/postgres_cdc_rls/cdc_rls.ex

@@ -9,6 +9,7 @@ defmodule Extensions.PostgresCdcRls do
   alias RealtimeWeb.Endpoint
   alias Extensions.PostgresCdcRls, as: Rls
   alias Rls.Subscriptions
+  alias Realtime.Rpc
 
   @spec handle_connect(map()) :: {:ok, {pid(), pid()}} | nil
   def handle_connect(args) do
@@ -31,7 +32,7 @@ defmodule Extensions.PostgresCdcRls do
     conn_node = node(conn)
 
     if conn_node !== node() do
-      :rpc.call(
+      Rpc.call(
         conn_node,
         Subscriptions,
         :create,
@@ -67,7 +68,7 @@ defmodule Extensions.PostgresCdcRls do
       "Starting distributed postgres extension #{inspect(lauch_node: launch_node, region: region, platform_region: platform_region)}"
     )
 
-    case :rpc.call(launch_node, __MODULE__, :start, [args], 30_000) do
+    case Rpc.call(launch_node, __MODULE__, :start, [args], 30_000) do
       {:ok, _pid} = ok ->
         ok
 

lib/extensions/postgres_cdc_rls/subscriptions_checker.ex

@@ -7,7 +7,7 @@ defmodule Extensions.PostgresCdcRls.SubscriptionsChecker do
   alias Rls.Subscriptions
 
   alias Realtime.Helpers, as: H
-
+  alias Realtime.Rpc
   @timeout 120_000
   @max_delete_records 1000
 
@@ -178,7 +178,7 @@ defmodule Extensions.PostgresCdcRls.SubscriptionsChecker do
       if node == node() do
         acc ++ not_alive_pids(pids)
       else
-        case :rpc.call(node, __MODULE__, :not_alive_pids, [pids], 15_000) do
+        case Rpc.call(node, __MODULE__, :not_alive_pids, [pids], 15_000) do
           {:badrpc, _} = error ->
             Logger.error("Can't check pids on node #{inspect(node)}: #{inspect(error)}")
             acc
@@ -202,18 +202,10 @@ defmodule Extensions.PostgresCdcRls.SubscriptionsChecker do
   end
 
   defp check_delete_queue() do
-    Process.send_after(
-      self(),
-      :check_delete_queue,
-      1000
-    )
+    Process.send_after(self(), :check_delete_queue, 1000)
   end
 
   defp check_active_pids() do
-    Process.send_after(
-      self(),
-      :check_active_pids,
-      @timeout
-    )
+    Process.send_after(self(), :check_active_pids, @timeout)
   end
 end

lib/extensions/postgres_cdc_stream/cdc_stream.ex

@@ -5,6 +5,7 @@ defmodule Extensions.PostgresCdcStream do
   require Logger
 
   alias Extensions.PostgresCdcStream, as: Stream
+  alias Realtime.Rpc
 
   def handle_connect(opts) do
     Enum.reduce_while(1..5, nil, fn retry, acc ->
@@ -59,7 +60,7 @@ defmodule Extensions.PostgresCdcStream do
       "Starting distributed postgres extension #{inspect(lauch_node: launch_node, region: region, platform_region: platform_region)}"
     )
 
-    case :rpc.call(launch_node, __MODULE__, :start, [args], 30_000) do
+    case Rpc.call(launch_node, __MODULE__, :start, [args], 30_000) do
       {:ok, _pid} = ok ->
         ok
 

lib/realtime/helpers.ex

@@ -5,7 +5,7 @@ defmodule Realtime.Helpers do
 
   alias Realtime.Api.Tenant
   alias Realtime.PostgresCdc
-
+  alias Realtime.Rpc
   require Logger
 
   @spec cancel_timer(reference() | nil) :: non_neg_integer() | false | :ok | nil
@@ -318,7 +318,7 @@ defmodule Realtime.Helpers do
     [node() | Node.list()]
     |> Task.async_stream(
       fn node ->
-        :erpc.call(node, __MODULE__, :kill_connections_to_tenant_id, [tenant_id, reason], 5000)
+        Rpc.ecall(node, __MODULE__, :kill_connections_to_tenant_id, [tenant_id, reason], 5000)
       end,
       timeout: 5000
     )

lib/realtime/monitoring/latency.ex

@@ -8,6 +8,7 @@ defmodule Realtime.Latency do
   require Logger
 
   alias Realtime.Helpers
+  alias Realtime.Rpc
 
   defmodule Payload do
     @moduledoc false
@@ -94,7 +95,7 @@ defmodule Realtime.Latency do
       for n <- [Node.self() | Node.list()] do
         Task.Supervisor.async(Realtime.TaskSupervisor, fn ->
           {latency, response} =
-            :timer.tc(fn -> :rpc.call(n, __MODULE__, :pong, [pong_timeout], timer_timeout) end)
+            :timer.tc(fn -> Rpc.call(n, __MODULE__, :pong, [pong_timeout], timer_timeout) end)
 
           latency_ms = latency / 1_000
           region = Application.get_env(:realtime, :region, "not_set")

lib/realtime/rpc.ex

@@ -0,0 +1,54 @@
+defmodule Realtime.Rpc do
+  @moduledoc """
+  RPC module for Realtime with the intent of standardizing the RPC interface and collect telemetry
+  """
+  alias Realtime.Telemetry
+
+  @doc """
+  Calls external node using :rpc.call/5 and collects telemetry
+  """
+  def call(node, mod, func, opts \\ [], timeout \\ 15000) do
+    {latency, response} = :timer.tc(fn -> :rpc.call(node, mod, func, opts, timeout) end)
+
+    Telemetry.execute([:rpc, :call], latency, %{
+      mod: mod,
+      func: func,
+      target_node: node,
+      origin_node: node()
+    })
+
+    response
+  rescue
+    _ ->
+      Telemetry.execute([:erpc, :call], timeout, %{
+        mod: mod,
+        func: func,
+        target_node: node,
+        origin_node: node()
+      })
+  end
+
+  @doc """
+  Calls external node using :erpc.call/5 and collects telemetry
+  """
+  def ecall(node, mod, func, opts \\ [], timeout \\ 15000) do
+    {latency, response} = :timer.tc(fn -> :erpc.call(node, mod, func, opts, timeout) end)
+
+    Telemetry.execute([:erpc, :call], latency, %{
+      mod: mod,
+      func: func,
+      target_node: node,
+      origin_node: node()
+    })
+
+    response
+  rescue
+    _ ->
+      Telemetry.execute([:erpc, :call], timeout, %{
+        mod: mod,
+        func: func,
+        target_node: node,
+        origin_node: node()
+      })
+  end
+end

lib/realtime/tenants/connect.ex

@@ -14,6 +14,7 @@ defmodule Realtime.Tenants.Connect do
   alias Realtime.Helpers
   alias Realtime.Tenants
   alias Realtime.UsersCounter
+  alias Realtime.Rpc
 
   @erpc_timeout_default 5000
   @check_connected_user_interval_default 50_000
@@ -173,7 +174,7 @@ defmodule Realtime.Tenants.Connect do
     with tenant <- Tenants.Cache.get_tenant_by_external_id(tenant_id),
          :ok <- tenant_suspended?(tenant),
          {:ok, node} <- Realtime.Nodes.get_node_for_tenant(tenant) do
-      :erpc.call(node, __MODULE__, :connect, [tenant_id, opts], erpc_timeout)
+      Rpc.ecall(node, __MODULE__, :connect, [tenant_id, opts], erpc_timeout)
     end
   end
 

lib/realtime_web/controllers/metrics_controller.ex

@@ -2,13 +2,14 @@ defmodule RealtimeWeb.MetricsController do
   use RealtimeWeb, :controller
   require Logger
   alias Realtime.PromEx
+  alias Realtime.Rpc
 
   def index(conn, _) do
     cluster_metrics =
       Node.list()
       |> Task.async_stream(
         fn node ->
-          {node, :rpc.call(node, PromEx, :get_metrics, [], 10_000)}
+          {node, Rpc.call(node, PromEx, :get_metrics, [], 10_000)}
         end,
         timeout: :infinity
       )

mix.exs

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