Set per-queue-type disk alarms for configured mounts

This introduces a new variant of `rabbit_alarm:resource_alarm_source()`:
`{disk, QueueType}` which triggers when the configured mount 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

@@ -308,10 +308,14 @@ internal_update(#state{limit = DataDirLimit,
     ets:insert(?MOUNT_ETS_NAME, [M || _Path := M <- NewMounts]),
 
     AlarmedMs = alarmed_mounts(Mounts),
+    AlarmedQTs = alarmed_queue_types(Mounts),
     NewAlarmedMs = alarmed_mounts(NewMounts),
+    NewAlarmedQTs = alarmed_queue_types(NewMounts),
 
     NewlyClearedMs = sets:subtract(AlarmedMs, NewAlarmedMs),
+    NewlyClearedQTs = sets:subtract(AlarmedQTs, NewAlarmedQTs),
     NewlyAlarmedMs = sets:subtract(NewAlarmedMs, AlarmedMs),
+    NewlyAlarmedQTs = sets:subtract(NewAlarmedQTs, AlarmedQTs),
 
     lists:foreach(
       fun(Path) ->
@@ -320,15 +324,23 @@ internal_update(#state{limit = DataDirLimit,
                      available = Available} = maps:get(Path, NewMounts),
               emit_update_info(Name, "insufficient", Available, Limit)
       end, lists:sort(sets:to_list(NewlyAlarmedMs))),
-    %% 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{name = Name,
                      limit = Limit,
                      available = Available} = maps:get(Path, NewMounts),
               emit_update_info(Name, "sufficient", Available, Limit)
       end, lists:sort(sets:to_list(NewlyClearedMs))),
-    %% 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,
@@ -353,6 +365,18 @@ alarmed_mounts(Mounts) ->
               Acc
       end, sets:new([{version, 2}]), Mounts).
 
+-spec alarmed_queue_types(mounts()) ->
+    sets:set(module()).
+alarmed_queue_types(MountPoints) ->
+    maps:fold(
+      fun (_Path, #mount{available = Available,
+                         limit = Limit,
+                         queue_types = QTs}, Acc) when Available < Limit ->
+              sets:union(QTs, Acc);
+          (_Path, _Mount, Acc) ->
+              Acc
+      end, sets:new([{version, 2}]), MountPoints).
+
 -spec get_disk_free(#state{}) ->
     #{file:filename() => AvailableB :: non_neg_integer()}.
 get_disk_free(#state{dir = DataDir, mounts = Mounts}) ->

deps/rabbitmq_prometheus/src/collectors/prometheus_rabbitmq_alarm_metrics_collector.erl

@@ -35,11 +35,13 @@ collect_mf(_Registry, Callback) ->
         Alarms = rabbit_alarm:get_local_alarms(500), %% TODO: figure out timeout
         ActiveAlarms =
             lists:foldl(fun ({{resource_limit, disk, _}, _}, Acc) ->
-                                maps:put(disk_limit, 1, Acc);
+                                Acc#{disk_limit => 1};
+                            ({{resource_limit, {disk, QT}, _}, _}, Acc) ->
+                                Acc#{{disk, QT} => 1};
                             ({{resource_limit, memory, _}, _}, Acc) ->
-                                maps:put(memory_limit, 1, Acc);
+                                Acc#{memory_limit => 1};
                             ({file_descriptor_limit, _}, Acc) ->
-                                maps:put(file_descriptor_limit, 1, Acc)
+                                Acc#{file_descriptor_limit => 1}
                         end,
                         #{},
                         Alarms),
@@ -58,6 +60,15 @@ collect_mf(_Registry, Callback) ->
                            <<"is 1 if VM memory watermark alarm is in effect">>,
                            untyped,
                            [untyped_metric(maps:get(memory_limit, ActiveAlarms, 0))])),
+
+        Callback(create_mf(?METRIC_NAME(<<"queue_type_free_disk_space_watermark">>),
+                           <<"is 1 if the queue type disk-space alarm is in effect">>,
+                           untyped,
+                           [prometheus_model_helpers:untyped_metric(
+                              #{queue_type => QT},
+                              maps:get({disk, QT}, ActiveAlarms, 0)) ||
+                            {_, QT} <- rabbit_registry:lookup_all(queue)])),
+
         ok
     catch
         exit:{timeout, _} ->