fix: log rate limit (#1793)

Author: filipecabaco

config/runtime.exs

@@ -91,6 +91,7 @@ metrics_pusher_auth = System.get_env("METRICS_PUSHER_AUTH")
 metrics_pusher_interval_ms = Env.get_integer("METRICS_PUSHER_INTERVAL_MS", :timer.seconds(30))
 metrics_pusher_timeout_ms = Env.get_integer("METRICS_PUSHER_TIMEOUT_MS", :timer.seconds(15))
 metrics_pusher_compress = Env.get_boolean("METRICS_PUSHER_COMPRESS", true)
+log_throttle_janitor_interval_ms = Env.get_integer("LOG_THROTTLE_JANITOR_INTERVAL_IN_MS", :timer.minutes(10))
 
 metrics_pusher_extra_labels =
   case System.get_env("METRICS_PUSHER_EXTRA_LABELS", "") do
@@ -177,6 +178,7 @@ config :realtime,
     max_calls: client_presence_max_calls,
     window_ms: client_presence_window_ms
   ],
+  log_throttle_janitor_interval_ms: log_throttle_janitor_interval_ms,
   disable_healthcheck_logging: disable_healthcheck_logging,
   metrics_pusher_enabled: metrics_pusher_enabled,
   metrics_pusher_url: metrics_pusher_url,

lib/realtime/application.ex

@@ -4,6 +4,7 @@ defmodule Realtime.Application do
   @moduledoc false
 
   use Application
+  require Cachex.Spec
   require Logger
 
   alias Realtime.Repo.Replica
@@ -106,7 +107,16 @@ defmodule Realtime.Application do
              message_module: Realtime.BeaconPubSubAdapter
            ]
          ]},
-        {Cachex, name: Realtime.RateCounter},
+        Supervisor.child_spec({Cachex, name: Realtime.RateCounter}, id: Realtime.RateCounter),
+        Supervisor.child_spec(
+          {Cachex,
+           name: Realtime.LogThrottle,
+           expiration:
+             Cachex.Spec.expiration(
+               interval: Application.get_env(:realtime, :log_throttle_janitor_interval_ms, :timer.minutes(10))
+             )},
+          id: Realtime.LogThrottle
+        ),
         Realtime.Tenants.Cache,
         Realtime.RateCounter.DynamicSupervisor,
         Realtime.Latency,

lib/realtime_web/channels/realtime_channel/logging.ex

@@ -36,59 +36,86 @@ defmodule RealtimeWeb.RealtimeChannel.Logging do
   end
 
   @doc """
-  Logs an error if the log level is set to error
+  Logs an error if the log level is set to error.
+
+  Accepts an optional `throttle: {max_count, window_ms}` option to limit
+  how many times the log is emitted per tenant+code within the given time window.
   """
-  @spec maybe_log_error(socket :: Phoenix.Socket.t(), code :: binary(), msg :: any()) :: {:error, %{reason: binary}}
-  def maybe_log_error(socket, code, msg), do: maybe_log(socket, :error, code, msg)
+  @spec maybe_log_error(socket :: Phoenix.Socket.t(), code :: binary(), msg :: any(), opts :: keyword()) ::
+          {:error, %{reason: binary}}
+  def maybe_log_error(socket, code, msg, opts \\ []), do: maybe_log(socket, :error, code, msg, opts)
 
   @doc """
-  Logs a warning if the log level is set to warning
+  Logs a warning if the log level is set to warning.
+
+  Accepts an optional `throttle: {max_count, window_ms}` option to limit
+  how many times the log is emitted per tenant+code within the given time window.
   """
-  @spec maybe_log_warning(socket :: Phoenix.Socket.t(), code :: binary(), msg :: any()) :: {:error, %{reason: binary}}
-  def maybe_log_warning(socket, code, msg), do: maybe_log(socket, :warning, code, msg)
+  @spec maybe_log_warning(socket :: Phoenix.Socket.t(), code :: binary(), msg :: any(), opts :: keyword()) ::
+          {:error, %{reason: binary}}
+  def maybe_log_warning(socket, code, msg, opts \\ []), do: maybe_log(socket, :warning, code, msg, opts)
 
   @doc """
-  Logs an info if the log level is set to info
+  Logs an info if the log level is set to info.
   """
   @spec maybe_log_info(socket :: Phoenix.Socket.t(), msg :: any()) :: :ok
-  def maybe_log_info(socket, msg), do: maybe_log(socket, :info, nil, msg)
+  def maybe_log_info(socket, msg), do: maybe_log(socket, :info, nil, msg, [])
 
-  defp build_msg(code, msg) do
-    msg = stringify!(msg)
-    if code, do: "#{code}: #{msg}", else: msg
-  end
+  defp build_msg(nil, msg), do: stringify!(msg)
+  defp build_msg(code, msg), do: "#{code}: #{stringify!(msg)}"
 
-  defp log(%{assigns: %{tenant: tenant, access_token: access_token}}, level, code, msg) do
+  defp log(%{assigns: assigns}, level, code, msg) do
+    tenant = assigns.tenant
     Logger.metadata(external_id: tenant, project: tenant)
-    if level in [:error, :warning], do: update_metadata_with_token_claims(access_token)
+    enrich_metadata(level, Map.get(assigns, :access_token))
     Logger.log(level, msg, error_code: code)
-    if level in [:error], do: emit_system_error(level, code, tenant)
+    emit_telemetry(level, code, tenant)
   end
 
-  defp maybe_log(%{assigns: %{log_level: log_level}} = socket, level, code, msg) do
-    msg = build_msg(code, msg)
-    if Logger.compare_levels(log_level, level) != :gt, do: log(socket, level, code, msg)
-    if level in [:error, :warning], do: {:error, %{reason: msg}}, else: :ok
+  defp enrich_metadata(level, token) when level in [:error, :warning],
+    do: update_metadata_with_token_claims(token)
+
+  defp enrich_metadata(_level, _token), do: :ok
+
+  defp emit_telemetry(:error, code, tenant),
+    do: Telemetry.execute([:realtime, :channel, :error], %{count: 1}, %{code: code, tenant: tenant})
+
+  defp emit_telemetry(_level, _code, _tenant), do: :ok
+
+  defp maybe_log(%{assigns: %{log_level: log_level}} = socket, level, code, msg, opts) do
+    built_msg = build_msg(code, msg)
+    if Logger.compare_levels(log_level, level) != :gt, do: do_log(socket, level, code, built_msg, opts)
+    if level in [:error, :warning], do: {:error, %{reason: built_msg}}, else: :ok
   end
 
-  defp emit_system_error(level, code, tenant_id),
-    do: Telemetry.execute([:realtime, :channel, level], %{count: 1}, %{code: code, tenant: tenant_id})
+  defp do_log(socket, level, code, msg, []), do: log(socket, level, code, msg)
+
+  defp do_log(%{assigns: %{tenant: tenant}} = socket, level, code, msg, throttle: {max_count, window_ms}) do
+    key = {tenant, level, code}
+
+    case Cachex.get(Realtime.LogThrottle, key) do
+      {:ok, nil} ->
+        Cachex.put(Realtime.LogThrottle, key, 1, expire: window_ms)
+        log(socket, level, code, msg)
+
+      {:ok, count} when count < max_count ->
+        Cachex.incr(Realtime.LogThrottle, key)
+        log(socket, level, code, msg)
+
+      _ ->
+        emit_telemetry(level, code, tenant)
+    end
+  end
 
   defp stringify!(msg) when is_binary(msg), do: msg
   defp stringify!(msg), do: inspect(msg, pretty: true)
 
-  defp update_metadata_with_token_claims(nil), do: nil
+  defp update_metadata_with_token_claims(nil), do: :ok
 
   defp update_metadata_with_token_claims(token) do
     case Joken.peek_claims(token) do
-      {:ok, claims} ->
-        sub = Map.get(claims, "sub")
-        exp = Map.get(claims, "exp")
-        iss = Map.get(claims, "iss")
-        Logger.metadata(sub: sub, exp: exp, iss: iss)
-
-      _ ->
-        nil
+      {:ok, claims} -> Logger.metadata(sub: claims["sub"], exp: claims["exp"], iss: claims["iss"])
+      _ -> :ok
     end
   end
 end

test/realtime_web/channels/realtime_channel/logging_test.exs

@@ -1,5 +1,5 @@
 defmodule RealtimeWeb.RealtimeChannel.LoggingTest do
-  # async: false due to changes in Logger levels
+  # async: false due to changes in Logger levels and shared Cachex state
   use Realtime.DataCase, async: false
   import ExUnit.CaptureLog
   alias RealtimeWeb.RealtimeChannel.Logging
@@ -11,6 +11,7 @@ defmodule RealtimeWeb.RealtimeChannel.LoggingTest do
     level = Logger.level()
     Logger.configure(level: :info)
     tenant = tenant_fixture()
+    Cachex.clear(Realtime.LogThrottle)
 
     on_exit(fn ->
       :telemetry.detach(__MODULE__)
@@ -62,7 +63,7 @@ defmodule RealtimeWeb.RealtimeChannel.LoggingTest do
     end
   end
 
-  describe "maybe_log_error/3" do
+  describe "maybe_log_error/4" do
     test "logs error message when log_level is less or equal to error" do
       log_levels = [:debug, :info, :warning, :error]
 
@@ -102,8 +103,8 @@ defmodule RealtimeWeb.RealtimeChannel.LoggingTest do
     end
   end
 
-  describe "maybe_log_warning/3" do
-    test "logs error message when log_level is less or equal to warning" do
+  describe "maybe_log_warning/4" do
+    test "logs warning message when log_level is less or equal to warning" do
       log_levels = [:debug, :info, :warning]
 
       for log_level <- log_levels do
@@ -194,4 +195,87 @@ defmodule RealtimeWeb.RealtimeChannel.LoggingTest do
     log = capture_log(fn -> Logging.maybe_log_error(socket, "TestError", "test error") end)
     assert log =~ tenant_id
   end
+
+  describe "throttle option" do
+    test "logs exactly max_count times within the window but always emits telemetry" do
+      tenant_id = random_string()
+      socket = %{assigns: %{log_level: :error, tenant: tenant_id, access_token: "test_token"}}
+
+      logs =
+        capture_log(fn ->
+          for _ <- 1..5 do
+            Logging.maybe_log_error(socket, "ThrottleCode", "msg", throttle: {3, :timer.seconds(60)})
+          end
+        end)
+
+      assert logs |> String.split("ThrottleCode: msg") |> length() == 4
+
+      for _ <- 1..5 do
+        assert_receive {[:realtime, :channel, :error], %{count: 1}, %{code: "ThrottleCode", tenant: ^tenant_id}}
+      end
+    end
+
+    test "still returns {:error, reason} even when throttled" do
+      tenant_id = random_string()
+      socket = %{assigns: %{log_level: :error, tenant: tenant_id, access_token: "test_token"}}
+
+      for _ <- 1..5 do
+        assert Logging.maybe_log_error(socket, "ThrottleCode", "msg", throttle: {2, :timer.seconds(60)}) ==
+                 {:error, %{reason: "ThrottleCode: msg"}}
+      end
+    end
+
+    test "resets after the window expires" do
+      tenant_id = random_string()
+      socket = %{assigns: %{log_level: :error, tenant: tenant_id, access_token: "test_token"}}
+
+      logs_before =
+        capture_log(fn ->
+          for _ <- 1..3, do: Logging.maybe_log_error(socket, "WindowCode", "msg", throttle: {2, 200})
+        end)
+
+      assert logs_before |> String.split("WindowCode: msg") |> length() == 3
+
+      Process.sleep(400)
+
+      logs_after =
+        capture_log(fn ->
+          for _ <- 1..3, do: Logging.maybe_log_error(socket, "WindowCode", "msg", throttle: {2, 200})
+        end)
+
+      assert logs_after |> String.split("WindowCode: msg") |> length() == 3
+    end
+
+    test "different tenant+code pairs have independent counters" do
+      socket_a = %{assigns: %{log_level: :error, tenant: random_string(), access_token: "t"}}
+      socket_b = %{assigns: %{log_level: :error, tenant: random_string(), access_token: "t"}}
+
+      logs =
+        capture_log(fn ->
+          for _ <- 1..3 do
+            Logging.maybe_log_error(socket_a, "CodeA", "msg", throttle: {2, :timer.seconds(60)})
+            Logging.maybe_log_error(socket_b, "CodeB", "msg", throttle: {2, :timer.seconds(60)})
+          end
+        end)
+
+      assert logs |> String.split("CodeA: msg") |> length() == 3
+      assert logs |> String.split("CodeB: msg") |> length() == 3
+    end
+
+    test "concurrent callers do not exceed max_count" do
+      tenant_id = random_string()
+      socket = %{assigns: %{log_level: :error, tenant: tenant_id, access_token: "test_token"}}
+
+      logs =
+        capture_log(fn ->
+          1..20
+          |> Task.async_stream(fn _ ->
+            Logging.maybe_log_error(socket, "ConcurrentCode", "msg", throttle: {5, :timer.seconds(60)})
+          end)
+          |> Stream.run()
+        end)
+
+      assert logs |> String.split("ConcurrentCode: msg") |> length() <= 6
+    end
+  end
 end