Add discard_threshold to Logger (#7193)

Author: josevalim

lib/logger/lib/logger.ex

@@ -130,6 +130,13 @@ defmodule Logger do
       in the queue is reduced to `sync_threshold * 0.75` messages.
       Defaults to 20 messages.
 
+    * `:discard_threshold` - if the `Logger` manager has more than
+      `:discard_threshold` messages in its queue, `Logger` will change
+      to *discard mode* and messages will be discarded directly in the
+      clients. `Logger` will return to *sync mode* once the number of
+      messages in the queue is reduced to `discard_threshold * 0.75`
+      messages. Defaults to 500 messages.
+
     * `:translator_inspect_opts` - when translating OTP reports and
       errors, the last message and state must be inspected in the
       error reports. This configuration allow developers to change
@@ -159,21 +166,25 @@ defmodule Logger do
       in Erlang syntax until the Logger application kicks in and
       uninstalls SASL's logger in favor of its own. Defaults to `false`.
 
-    * `:discard_threshold_for_error_logger` - a value that, when
-      reached, triggers the error logger to discard messages. This
-      value must be a positive number that represents the maximum
-      number of messages accepted per second. Once above this
-      threshold, the [`:error_logger`](http://erlang.org/doc/man/error_logger.html)
-      enters discard mode for the remainder of that second. Defaults to 500 messages.
+    * `:discard_threshold_for_error_logger` - if `:error_logger` has more than
+      `discard_threshold` messages in its inbox, messages will be dropped
+      until the message queue goes down to `discard_threshold * 0.75`
+      entries. The threshold will be checked once again after 10% of thereshold
+      messages are processed, to avoid messages from being constantly dropped.
+      For exmaple, if the thereshold is 500 (the default) and the inbox has
+      600 messages, 250 messages will dropped, bringing the inbox down to
+      350 (0.75 * threshold) entries and 50 (0.1 * theshold) messages will
+      be processed before the threshold is checked once again.
 
-  For example, to configure `Logger` to redirect all [`:error_logger`](http://erlang.org/doc/man/error_logger.html)
-  messages using a `config/config.exs` file:
+  For example, to configure `Logger` to redirect all
+  [`:error_logger`](http://erlang.org/doc/man/error_logger.html) messages
+  using a `config/config.exs` file:
 
       config :logger,
         handle_otp_reports: true,
         handle_sasl_reports: true
 
-  Furthermore, `Logger` allows messages sent by OTP's [`:error_logger`](http://erlang.org/doc/man/error_logger.html)
+  Furthermore, `Logger` allows messages sent by OTP's `:error_logger`
   to be translated into an Elixir format via translators. Translators
   can be dynamically added at any time with the `add_translator/1`
   and `remove_translator/1` APIs. Check `Logger.Translator` for more
@@ -600,7 +611,7 @@ defmodule Logger do
         %{mode: mode, truncate: truncate, level: min_level, utc_log: utc_log?} =
           Logger.Config.__data__()
 
-        if compare_levels(level, min_level) != :lt do
+        if compare_levels(level, min_level) != :lt and mode != :discard do
           metadata = [pid: self()] ++ Keyword.merge(pdict, metadata)
           {message, metadata} = normalize_message(chardata_or_fun, metadata)
           truncated = truncate(message, truncate)

lib/logger/lib/logger/config.ex

@@ -105,10 +105,28 @@ defmodule Logger.Config do
         {:ok, state}
 
       new_mode ->
+        if new_mode == :discard do
+          message =
+            "Logger has #{message_queue_length()} messages in its queue, " <>
+              "which is above :discard_threshold. Messages will be discarded " <>
+              "until the message queue goes back to 75% of the threshold size"
+
+          log(:warn, message, state)
+        end
+
+        if mode == :discard do
+          log(:warn, "Logger has stopped discarding messages", state)
+        end
+
         {:ok, persist(%{state | mode: new_mode})}
     end
   end
 
+  defp log(level, message, state) do
+    event = {Logger, message, Logger.Utils.timestamp(state.utc_log), pid: self()}
+    :gen_event.notify(self(), {level, Process.group_leader(), event})
+  end
+
   def handle_call(:backends, state) do
     {:ok, Application.get_env(:logger, :backends), state}
   end
@@ -162,18 +180,27 @@ defmodule Logger.Config do
   ## Helpers
 
   defp compute_mode(state) do
-    {:message_queue_len, len} = Process.info(self(), :message_queue_len)
-
-    cond do
-      len > state.sync_threshold and state.mode == :async ->
-        :sync
+    %{
+      mode: mode,
+      async_threshold: async_threshold,
+      sync_threshold: sync_threshold,
+      keep_threshold: keep_threshold,
+      discard_threshold: discard_threshold
+    } = state
 
-      len < state.async_threshold and state.mode == :sync ->
-        :async
+    Logger.Utils.compute_mode(
+      mode,
+      message_queue_length(),
+      async_threshold,
+      sync_threshold,
+      keep_threshold,
+      discard_threshold
+    )
+  end
 
-      true ->
-        state.mode
-    end
+  defp message_queue_length() do
+    {:message_queue_len, messages} = Process.info(self(), :message_queue_len)
+    messages
   end
 
   defp update_backends(fun) do
@@ -196,22 +223,24 @@ defmodule Logger.Config do
     sync_threshold = Application.get_env(:logger, :sync_threshold)
     async_threshold = trunc(sync_threshold * 0.75)
 
+    discard_threshold = Application.get_env(:logger, :discard_threshold)
+    keep_threshold = trunc(discard_threshold * 0.75)
+
     state = %{
       level: level,
       mode: mode,
       truncate: truncate,
       utc_log: utc_log,
-      sync_threshold: sync_threshold,
       async_threshold: async_threshold,
+      sync_threshold: sync_threshold,
+      keep_threshold: keep_threshold,
+      discard_threshold: discard_threshold,
       translators: translators
     }
 
     case compute_mode(state) do
-      ^mode ->
-        persist(state)
-
-      new_mode ->
-        persist(%{state | mode: new_mode})
+      ^mode -> persist(state)
+      new_mode -> persist(%{state | mode: new_mode})
     end
   end
 

lib/logger/lib/logger/error_handler.ex

@@ -1,10 +1,7 @@
 defmodule Logger.ErrorHandler do
   @moduledoc false
-
   @behaviour :gen_event
 
-  require Logger
-
   def init({otp?, sasl?, threshold}) do
     # We store the Logger PID in the state because when we are shutting
     # down the Logger application, the Logger process may be terminated
@@ -14,11 +11,10 @@ defmodule Logger.ErrorHandler do
     state = %{
       otp: otp?,
       sasl: sasl?,
-      threshold: threshold,
+      discard_threshold: threshold,
+      keep_threshold: trunc(threshold * 0.75),
       logger: Process.whereis(Logger),
-      last_length: 0,
-      last_time: :os.timestamp(),
-      dropped: 0
+      skip: 0
     }
 
     {:ok, state}
@@ -31,7 +27,7 @@ defmodule Logger.ErrorHandler do
   end
 
   def handle_event(event, state) do
-    state = check_threshold(state)
+    state = check_threshold_unless_skipping(state)
     log_event(event, state)
     {:ok, state}
   end
@@ -84,10 +80,15 @@ defmodule Logger.ErrorHandler do
   defp log_event(_, _state), do: :ok
 
   defp log_event(level, kind, gl, pid, {type, _} = data, state) do
-    %{level: min_level, truncate: truncate, utc_log: utc_log?, translators: translators} =
-      Logger.Config.__data__()
-
-    with log when log != :lt <- Logger.compare_levels(level, min_level),
+    %{
+      mode: mode,
+      level: min_level,
+      truncate: truncate,
+      utc_log: utc_log?,
+      translators: translators
+    } = Logger.Config.__data__()
+
+    with true <- Logger.compare_levels(level, min_level) != :lt and mode != :discard,
          {:ok, message} <- translate(translators, min_level, level, kind, data, truncate) do
       message = Logger.Utils.truncate(message, truncate)
 
@@ -106,31 +107,35 @@ defmodule Logger.ErrorHandler do
   defp ensure_pid(pid) when is_pid(pid), do: pid
   defp ensure_pid(_), do: self()
 
-  defp check_threshold(state) do
-    %{last_time: last_time, last_length: last_length, dropped: dropped, threshold: threshold} =
-      state
+  defp check_threshold_unless_skipping(%{skip: 0} = state) do
+    check_threshold(state)
+  end
 
-    {m, s, _} = current_time = :os.timestamp()
-    current_length = message_queue_length()
+  defp check_threshold_unless_skipping(%{skip: skip} = state) do
+    %{state | skip: skip - 1}
+  end
 
-    cond do
-      match?({^m, ^s, _}, last_time) and current_length - last_length > threshold ->
-        count = drop_messages(current_time, 0)
-        %{state | dropped: dropped + count, last_length: message_queue_length()}
+  def check_threshold(state) do
+    %{discard_threshold: discard_threshold, keep_threshold: keep_threshold} = state
+    current_length = message_queue_length()
 
-      match?({^m, ^s, _}, last_time) ->
-        state
+    if current_length >= discard_threshold do
+      to_drop = current_length - keep_threshold
+      drop_messages(to_drop)
 
-      true ->
-        if dropped > 0 do
-          message =
-            "Logger dropped #{dropped} OTP/SASL messages as it " <>
-              "exceeded the amount of #{threshold} messages/second"
+      message =
+        "Logger dropped #{to_drop} OTP/SASL messages as it had #{current_length} messages in " <>
+          "its inbox, exceeding the amount of :discard_threshold #{discard_threshold} messages. " <>
+          "The number of messages was reduced to #{keep_threshold} (75% of the threshold)"
 
-          Logger.warn(message)
-        end
+      %{utc_log: utc_log?} = Logger.Config.__data__()
+      event = {Logger, message, Logger.Utils.timestamp(utc_log?), pid: self()}
+      :gen_event.notify(state.logger, {:warn, Process.group_leader(), event})
 
-        %{state | dropped: 0, last_time: current_time, last_length: current_length}
+      # We won't check the threshold for the next 10% of the threshold messages
+      %{state | skip: trunc(discard_threshold * 0.1)}
+    else
+      state
     end
   end
 
@@ -139,17 +144,15 @@ defmodule Logger.ErrorHandler do
     len
   end
 
-  defp drop_messages({m, s, _} = last_time, count) do
-    case :os.timestamp() do
-      {^m, ^s, _} ->
-        receive do
-          {:notify, _event} -> drop_messages(last_time, count + 1)
-        after
-          0 -> count
-        end
+  defp drop_messages(0) do
+    :ok
+  end
 
-      _ ->
-        count
+  defp drop_messages(count) do
+    receive do
+      {:notify, _event} -> drop_messages(count - 1)
+    after
+      0 -> :ok
     end
   end
 

lib/logger/lib/logger/utils.ex

@@ -1,6 +1,37 @@
 defmodule Logger.Utils do
   @moduledoc false
 
+  @doc """
+  Computes the logging mode.
+
+  The result may be either :sync, :async or :discard.
+  """
+  @spec compute_mode(
+          mode,
+          non_neg_integer(),
+          non_neg_integer(),
+          non_neg_integer(),
+          non_neg_integer(),
+          non_neg_integer()
+        ) :: mode
+        when mode: :sync | :async | :discard
+  def compute_mode(
+        mode,
+        messages,
+        async_threshold,
+        sync_threshold,
+        keep_threshold,
+        discard_threshold
+      ) do
+    case mode do
+      _ when messages >= discard_threshold -> :discard
+      :discard when messages > keep_threshold -> :discard
+      _ when messages >= sync_threshold -> :sync
+      :sync when messages > async_threshold -> :sync
+      _ -> :async
+    end
+  end
+
   @doc """
   Truncates a `chardata` into `n` bytes.
 

lib/logger/mix.exs

@@ -20,11 +20,12 @@ defmodule Logger.MixProject do
         backends: [:console],
         translators: [{Logger.Translator, :translate}],
         sync_threshold: 20,
+        discard_threshold: 500,
         handle_otp_reports: true,
         handle_sasl_reports: false,
+        discard_threshold_for_error_logger: 500,
         compile_time_purge_level: :debug,
         compile_time_application: nil,
-        discard_threshold_for_error_logger: 500,
         translator_inspect_opts: [],
         console: []
       ]

lib/logger/test/logger/config_test.exs

@@ -0,0 +1,41 @@
+defmodule Logger.ConfigTest do
+  use Logger.Case
+  require Logger
+
+  test "log/2 relies on sync_threshold" do
+    Logger.remove_backend(:console)
+    Logger.configure(sync_threshold: 0)
+    for _ <- 1..1000, do: Logger.log(:info, "some message")
+  after
+    Logger.configure(sync_threshold: 20)
+    Logger.add_backend(:console)
+  end
+
+  test "log/2 relies on discard_threshold" do
+    Logger.remove_backend(:console)
+    Logger.configure(discard_threshold: 0)
+    for _ <- 1..1000, do: Logger.log(:info, "some message")
+  after
+    Logger.configure(discard_threshold: 10000)
+    Logger.add_backend(:console)
+  end
+
+  test "restarts Logger.Config on Logger exits" do
+    Process.whereis(Logger) |> Process.exit(:kill)
+    wait_for_logger()
+    wait_for_handler(Logger, Logger.Config)
+    wait_for_handler(:error_logger, Logger.ErrorHandler)
+  end
+
+  test "Logger.Config updates config on config_change/3" do
+    :ok = Logger.configure(level: :debug)
+
+    try do
+      Application.put_env(:logger, :level, :error)
+      assert Logger.App.config_change([level: :error], [], []) === :ok
+      assert Logger.level() === :error
+    after
+      Logger.configure(level: :debug)
+    end
+  end
+end