rabbit.schema: Add config options for per-queue-type disk limits

Author: the-mikedavis

deps/rabbit/priv/schema/rabbit.schema

@@ -1238,6 +1238,73 @@ fun(Conf) ->
     end
 end}.
 
+%% Per-queue-type / per-mount disk alarms
+{mapping, "disk_free_limits.$num.name", "rabbit.disk_free_limits", [
+    {datatype, [binary]}
+]}.
+{mapping, "disk_free_limits.$num.mount", "rabbit.disk_free_limits", [
+    {datatype, [string]}
+]}.
+{mapping, "disk_free_limits.$num.limit", "rabbit.disk_free_limits", [
+    {datatype, [integer, string]},
+    {validators, ["is_supported_information_unit"]}
+]}.
+{mapping, "disk_free_limits.$num.queue_types", "rabbit.disk_free_limits", [
+    {datatype, [binary]}
+]}.
+
+{translation, "rabbit.disk_free_limits",
+fun(Conf) ->
+    case cuttlefish_variable:filter_by_prefix("disk_free_limits", Conf) of
+        [] ->
+            cuttlefish:unset();
+        Settings ->
+            Ls = lists:foldl(
+                   fun ({["disk_free_limits", Num, Key0], Value0}, Acc) ->
+                           Idx = case string:to_integer(Num) of
+                                     {N, []} -> N;
+                                     _ -> cuttlefish:invalid(lists:flatten(io_lib:format("~p could not be parsed as a number", [Num])))
+                                 end,
+                           Key = case Key0 of
+                                     "name" -> name;
+                                     "mount" -> mount;
+                                     "limit" -> limit;
+                                     "queue_types" -> queue_types;
+                                      _ -> cuttlefish:invalid(lists:flatten(io_lib:format("~p is invalid", [Key0])))
+                                 end,
+                           Value = case Key of
+                                       queue_types -> string:split(Value0, ",");
+                                       _ -> Value0
+                                   end,
+                           maps:update_with(
+                             Idx,
+                             fun (#{Key := ExistingValue} = Limit) ->
+                                     cuttlefish:warn(
+                                       io_lib:format("Disk limit ~b has duplicate setting ~ts, "
+                                                     "using ~tp instead of ~tp",
+                                                     [Idx, Key, Value, ExistingValue])),
+                                     Limit#{Key := Value};
+                                 (Limit) ->
+                                     Limit#{Key => Value}
+                             end, #{Key => Value}, Acc);
+                       (Other, _Acc) ->
+                           cuttlefish:invalid(
+                             lists:flatten(io_lib:format("~p is invalid", [Other])))
+                   end, #{}, Settings),
+            maps:fold(
+              fun(_Idx, #{name := Name}, Names) ->
+                      case sets:is_element(Name, Names) of
+                          true ->
+                              cuttlefish:invalid(
+                                lists:flatten(io_lib:format("name ~ts is used by multiple mounts", [Name])));
+                          false ->
+                              sets:add_element(Name, Names)
+                      end
+              end, sets:new([{version, 2}]), Ls),
+            Ls
+    end
+end}.
+
 %%
 %% Clustering
 %% =====================

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -467,6 +467,26 @@ tcp_listen_options.exit_on_close = false",
   "total_memory_available_override_value = 1024MB",
   [{rabbit,[{total_memory_available_override_value, "1024MB"}]}],
   []},
+ {disk_free_limits_per_mount,
+  "disk_free_limits.1.name = messaging
+   disk_free_limits.1.mount = /data/queues
+   disk_free_limits.1.limit = 2GB
+   disk_free_limits.1.queue_types = classic,quorum
+
+   disk_free_limits.2.name = streaming
+   disk_free_limits.2.mount = /data/streams
+   disk_free_limits.2.limit = 2GB
+   disk_free_limits.2.queue_types = stream",
+  [{rabbit,[{disk_free_limits,
+             #{1 => #{name => <<"messaging">>,
+                      mount => "/data/queues",
+                      limit => "2GB",
+                      queue_types => [<<"classic">>, <<"quorum">>]},
+               2 => #{name => <<"streaming">>,
+                      mount => "/data/streams",
+                      limit => "2GB",
+                      queue_types => [<<"stream">>]}}}]}],
+  []},
  {ranch_connection_max,
   "ranch_connection_max = 999",
   [{rabbit,[{ranch_connection_max, 999}]}],