cluster wide queue limit

Author: SimonUnge

deps/rabbit/priv/schema/rabbit.schema

@@ -1437,6 +1437,21 @@ fun(Conf) ->
 end}.
 
 
+{mapping, "cluster_queue_limit", "rabbit.cluster_queue_limit",
+  [{datatype, [{atom, infinity}, integer]}]}.
+
+{translation, "rabbit.cluster_queue_limit",
+ fun(Conf) ->
+  case cuttlefish:conf_get("cluster_queue_limit", Conf, undefined) of
+      undefined -> cuttlefish:unset();
+      infinity  -> infinity;
+      Val when is_integer(Val) andalso Val > 0 -> Val;
+      _         -> cuttlefish:invalid("should be positive integer or 'infinity'")
+  end
+ end
+}.
+
+
 %% Interval (in milliseconds) at which we send keepalive messages
 %% to other cluster members. Note that this is not the same thing
 %% as net_ticktime; missed keepalive messages will not cause nodes

deps/rabbit/src/rabbit_channel.erl

@@ -1065,6 +1065,21 @@ check_vhost_queue_limit(#resource{name = QueueName}, VHost) ->
 
   end.
 
+check_cluster_queue_limit(#resource{name = QueueName}) ->
+    case rabbit_misc:get_env(rabbit, cluster_queue_limit, infinity) of
+        infinity ->
+            false;
+        Limit ->
+            case rabbit_db_queue:count() >= Limit of
+                true ->
+                    rabbit_misc:precondition_failed("cannot declare queue '~ts': "
+                                                    "queue limit in cluster (~tp) is reached",
+                                                    [QueueName, Limit]);
+                false ->
+                    ok
+            end
+    end.
+
 qbin_to_resource(QueueNameBin, VHostPath) ->
     name_to_resource(queue, QueueNameBin, VHostPath).
 
@@ -2526,6 +2541,7 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
         {error, not_found} ->
             %% enforce the limit for newly declared queues only
             check_vhost_queue_limit(QueueName, VHostPath),
+            check_cluster_queue_limit(QueueName),
             DlxKey = <<"x-dead-letter-exchange">>,
             case rabbit_misc:r_arg(VHostPath, exchange, Args, DlxKey) of
                undefined ->