rabbit_alarm: Add a helper to format resource alarm sources

Author: the-mikedavis

deps/rabbit/src/rabbit_alarm.erl

@@ -24,6 +24,7 @@
 
 -export([start_link/0, start/0, stop/0, register/2, set_alarm/1,
          clear_alarm/1, get_alarms/0, get_alarms/1, get_local_alarms/0, get_local_alarms/1, on_node_up/1, on_node_down/1,
+         format_resource_alarm_source/1,
          format_as_map/1, format_as_maps/1, is_local/1]).
 
 -export([init/1, handle_call/2, handle_event/2, handle_info/2,
@@ -124,25 +125,23 @@ is_local({file_descriptor_limit, _}) -> true;
 is_local({{resource_limit, _Resource, Node}, _}) when Node =:= node() -> true;
 is_local({{resource_limit, _Resource, Node}, _}) when Node =/= node() -> false.
 
+-spec format_resource_alarm_source(resource_alarm_source()) -> iodata().
+format_resource_alarm_source(disk) ->
+    ?DISK_SPACE_RESOURCE;
+format_resource_alarm_source(memory) ->
+    ?MEMORY_RESOURCE;
+format_resource_alarm_source(Unknown) ->
+    io_lib:format("~w", [Unknown]).
+
 -spec format_as_map(alarm()) -> #{binary() => term()}.
 format_as_map(file_descriptor_limit) ->
     #{
         <<"resource">> => ?FILE_DESCRIPTOR_RESOURCE,
         <<"node">> => node()
     };
-format_as_map({resource_limit, disk, Node}) ->
-    #{
-        <<"resource">> => ?DISK_SPACE_RESOURCE,
-        <<"node">> => Node
-    };
-format_as_map({resource_limit, memory, Node}) ->
-    #{
-        <<"resource">> => ?MEMORY_RESOURCE,
-        <<"node">> => Node
-    };
 format_as_map({resource_limit, Limit, Node}) ->
     #{
-        <<"resource">> => rabbit_data_coercion:to_binary(Limit),
+        <<"resource">> => iolist_to_binary(format_resource_alarm_source(Limit)),
         <<"node">> => Node
     }.
 
@@ -237,7 +236,7 @@ handle_event({node_down, Node}, #alarms{alarmed_nodes = AN} = State) ->
     AlarmsForDeadNode = maps:get(Node, AN, []),
     {ok, lists:foldr(fun(Source, AccState) ->
                              ?LOG_WARNING("~ts resource limit alarm cleared for dead node ~tp",
-                                          [Source, Node]),
+                                          [format_resource_alarm_source(Source), Node]),
                              maybe_alert(fun map_unappend/3, Node, Source, false, AccState)
                      end, State, AlarmsForDeadNode)};
 
@@ -283,7 +282,8 @@ maybe_alert(UpdateFun, Node, Source, WasAlertAdded,
                        end, AN1),
     case StillHasAlerts of
         true -> ok;
-        false -> ?LOG_WARNING("~ts resource limit alarm cleared across the cluster", [Source])
+        false -> ?LOG_WARNING("~ts resource limit alarm cleared across the cluster",
+                              [format_resource_alarm_source(Source)])
     end,
     Alert = {WasAlertAdded, StillHasAlerts, Node},
     case node() of
@@ -326,7 +326,7 @@ handle_set_resource_alarm(Source, Node, State) ->
       "**********************************************************~n"
       "*** Publishers will be blocked until this alarm clears ***~n"
       "**********************************************************~n",
-      [Source, Node]),
+      [format_resource_alarm_source(Source), Node]),
     {ok, maybe_alert(fun map_append/3, Node, Source, true, State)}.
 
 handle_set_alarm({file_descriptor_limit, []}, State) ->
@@ -342,7 +342,7 @@ handle_set_alarm(Alarm, State) ->
 
 handle_clear_resource_alarm(Source, Node, State) ->
     ?LOG_WARNING("~ts resource limit alarm cleared on node ~tp",
-                 [Source, Node]),
+                 [format_resource_alarm_source(Source), Node]),
     {ok, maybe_alert(fun map_unappend/3, Node, Source, false, State)}.
 
 handle_clear_alarm(file_descriptor_limit, State) ->

deps/rabbit/src/rabbit_reader.erl

@@ -1727,14 +1727,11 @@ send_error_on_channel0_and_close(Channel, Protocol, Reason, State) ->
 blocked_by_message(#throttle{blocked_by = Reasons}) ->
   %% we don't want to report internal flow as a reason here since
   %% it is entirely transient
-  Reasons1 = sets:del_element(flow, Reasons),
-  RStr = string:join([format_blocked_by(R) || R <- sets:to_list(Reasons1)], " & "),
+  RStr = lists:join([rabbit_alarm:format_resource_alarm_source(R) ||
+                     {resource, R} <- sets:to_list(Reasons)],
+                    " & "),
   list_to_binary(rabbit_misc:format("low on ~ts", [RStr])).
 
-format_blocked_by({resource, memory}) -> "memory";
-format_blocked_by({resource, disk})   -> "disk";
-format_blocked_by({resource, disc})   -> "disk".
-
 update_last_blocked_at(Throttle) ->
     Throttle#throttle{last_blocked_at = erlang:monotonic_time()}.