Set per-queue-type disk alarms for configured mount points

This introduces a new variant of `rabbit_alarm:resource_alarm_source()`:
`{disk, QueueType}` which triggers when the configured mount point for
queue type(s) fall under their limit of available space.

Author: the-mikedavis

deps/rabbit/src/rabbit_alarm.erl

@@ -49,7 +49,10 @@
 
 -export_type([alarm/0]).
 -type local_alarm() :: 'file_descriptor_limit'.
--type resource_alarm_source() :: 'disk' | 'memory'.
+-type resource_alarm_source() ::
+    memory
+    | disk
+    | {disk, rabbit_queue_type:queue_type()}.
 -type resource_alarm() :: {resource_limit, resource_alarm_source(), node()}.
 -type alarm() :: local_alarm() | resource_alarm().
 -type resource_alert() :: {WasAlarmSetForNode :: boolean(),
@@ -128,6 +131,8 @@ 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({disk, QueueType}) ->
+    io_lib:format("disk for queue type '~ts'", [QueueType]);
 format_resource_alarm_source(memory) ->
     ?MEMORY_RESOURCE;
 format_resource_alarm_source(Unknown) ->

deps/rabbit/src/rabbit_disk_monitor.erl

@@ -288,10 +288,14 @@ internal_update(#state{limit = DataDirLimit,
                                 limit = Limit} <- NewMountPoints]),
 
     AlarmedMPs = alarmed_mount_points(MountPoints),
+    AlarmedQTs = alarmed_queue_types(MountPoints),
     NewAlarmedMPs = alarmed_mount_points(NewMountPoints),
+    NewAlarmedQTs = alarmed_queue_types(NewMountPoints),
 
     NewlyClearedMPs = sets:subtract(AlarmedMPs, NewAlarmedMPs),
+    NewlyClearedQTs = sets:subtract(AlarmedQTs, NewAlarmedQTs),
     NewlyAlarmedMPs = sets:subtract(NewAlarmedMPs, AlarmedMPs),
+    NewlyAlarmedQTs = sets:subtract(NewAlarmedQTs, AlarmedQTs),
 
     lists:foreach(
       fun(Path) ->
@@ -301,7 +305,11 @@ internal_update(#state{limit = DataDirLimit,
                                                              NewMountPoints),
               emit_update_info(Id, "insufficient", Available, Limit)
       end, lists:sort(sets:to_list(NewlyAlarmedMPs))),
-    %% TODO: rabbit_alarm:set_alarm/1 for affected queue types
+    lists:foreach(
+      fun(QT) ->
+              Alarm = {resource_limit, {disk, QT}, node()},
+              rabbit_alarm:set_alarm({Alarm, []})
+      end, lists:sort(sets:to_list(NewlyAlarmedQTs))),
     lists:foreach(
       fun(Path) ->
               #mount_point{id = Id,
@@ -310,7 +318,11 @@ internal_update(#state{limit = DataDirLimit,
                                                              NewMountPoints),
               emit_update_info(Id, "sufficient", Available, Limit)
       end, lists:sort(sets:to_list(NewlyClearedMPs))),
-    %% TODO: rabbit_alarm:clear_alarm/1 for affected queue types
+    lists:foreach(
+      fun(QT) ->
+              Alarm = {resource_limit, {disk, QT}, node()},
+              rabbit_alarm:clear_alarm(Alarm)
+      end, lists:sort(sets:to_list(NewlyClearedQTs))),
 
     State#state{alarmed = NewAlarmed,
                 actual = DataDirFree,
@@ -336,6 +348,19 @@ alarmed_mount_points(MountPoints) ->
               Acc
       end, sets:new([{version, 2}]), MountPoints).
 
+-spec alarmed_queue_types(mount_points()) ->
+    sets:set(module()).
+alarmed_queue_types(MountPoints) ->
+    maps:fold(
+      fun (_Path, #mount_point{available = Available,
+                               limit = Limit,
+                               queue_types = QTs}, Acc)
+            when Available < Limit ->
+              sets:union(QTs, Acc);
+          (_Path, _MP, Acc) ->
+              Acc
+      end, sets:new([{version, 2}]), MountPoints).
+
 -spec get_disk_free(#state{}) ->
     #{file:filename_all() => AvailableB :: non_neg_integer()}.
 get_disk_free(#state{dir = DataDir, mount_points = MountPoints}) ->