Set segment_entry_count per vhost and use a better default

Author: Loïc Hoguin

deps/rabbit/Makefile

@@ -119,12 +119,7 @@ define PROJECT_ENV
 	    %% interval at which connection/channel tracking executes post operations
 	    {tracking_execution_timeout, 15000},
 	    {stream_messages_soft_limit, 256},
-        {track_auth_attempt_source, false},
-        %% Number of entries per index segment.
-        %% This value can only be changed safely
-        %% on an empty node. Default calculated
-        %% as trunc(math:pow(2,?REL_SEQ_BITS))).
-        {queue_index_segment_entry_count, 16384}
+        {track_auth_attempt_source, false}
 	  ]
 endef
 

deps/rabbit/src/rabbit_queue_index.erl

@@ -24,6 +24,9 @@
          read_global_recovery_terms/1,
          cleanup_global_recovery_terms/0]).
 
+%% Used by rabbit_vhost to set the segment_entry_count.
+-export([all_queue_directory_names/1]).
+
 -define(CLEAN_FILENAME, "clean.dot").
 
 %%----------------------------------------------------------------------------
@@ -280,6 +283,8 @@ reset_state(#qistate{ queue_name     = Name,
                  on_sync_fun(), on_sync_fun()) -> qistate().
 
 init(#resource{ virtual_host = VHost } = Name, OnSyncFun, OnSyncMsgFun) ->
+    #{segment_entry_count := SegmentEntryCount} = rabbit_vhost:read_config(VHost),
+    put(segment_entry_count, SegmentEntryCount),
     VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
     State = #qistate { dir = Dir } = blank_state(VHostDir, Name),
     false = rabbit_file:is_file(Dir), %% is_file == is file or dir
@@ -294,6 +299,8 @@ init(#resource{ virtual_host = VHost } = Name, OnSyncFun, OnSyncMsgFun) ->
 
 recover(#resource{ virtual_host = VHost } = Name, Terms, MsgStoreRecovered,
         ContainsCheckFun, OnSyncFun, OnSyncMsgFun) ->
+    #{segment_entry_count := SegmentEntryCount} = rabbit_vhost:read_config(VHost),
+    put(segment_entry_count, SegmentEntryCount),
     VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
     State = blank_state(VHostDir, Name),
     State1 = State #qistate{on_sync     = OnSyncFun,
@@ -1167,9 +1174,7 @@ array_new(Default) ->
     array:new([{default, Default}, fixed, {size, segment_entry_count()}]).
 
 segment_entry_count() ->
-    {ok, SegmentEntryCount} =
-        application:get_env(rabbit, queue_index_segment_entry_count),
-    SegmentEntryCount.
+    get(segment_entry_count).
 
 bool_to_int(true ) -> 1;
 bool_to_int(false) -> 0.

deps/rabbit/src/rabbit_vhost.erl

@@ -10,12 +10,12 @@
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include("vhost.hrl").
 
--export([recover/0, recover/1]).
+-export([recover/0, recover/1, read_config/1]).
 -export([add/2, add/4, delete/2, exists/1, with/2, with_user_and_vhost/3, assert/1, update/2,
          set_limits/2, vhost_cluster_state/1, is_running_on_all_nodes/1, await_running_on_all_nodes/2,
         list/0, count/0, list_names/0, all/0, parse_tags/1]).
 -export([info/1, info/2, info_all/0, info_all/1, info_all/2, info_all/3]).
--export([dir/1, msg_store_dir_path/1, msg_store_dir_wildcard/0]).
+-export([dir/1, msg_store_dir_path/1, msg_store_dir_wildcard/0, config_file_path/1]).
 -export([delete_storage/1]).
 -export([vhost_down/1]).
 -export([put_vhost/5]).
@@ -48,6 +48,7 @@ recover(VHost) ->
     VHostStubFile = filename:join(VHostDir, ".vhost"),
     ok = rabbit_file:ensure_dir(VHostStubFile),
     ok = file:write_file(VHostStubFile, VHost),
+    ok = ensure_config_file(VHost),
     {Recovered, Failed} = rabbit_amqqueue:recover(VHost),
     AllQs = Recovered ++ Failed,
     QNames = [amqqueue:get_name(Q) || Q <- AllQs],
@@ -57,6 +58,38 @@ recover(VHost) ->
     ok = rabbit_mirror_queue_misc:on_vhost_up(VHost),
     ok.
 
+ensure_config_file(VHost) ->
+    Path = config_file_path(VHost),
+    case filelib:is_regular(Path) of
+        %% The config file exists. Do nothing.
+        true ->
+            ok;
+        %% The config file does not exist.
+        %% Check if there are queues in this vhost.
+        false ->
+            QueueDirs = rabbit_queue_index:all_queue_directory_names(VHost),
+            SegmentEntryCount = case QueueDirs of
+                %% There are no queues. Write the configured value for
+                %% the segment entry count, or the new RabbitMQ default
+                %% introduced in v3.8.16.
+                [] ->
+                    application:get_env(rabbit, queue_index_segment_entry_count,
+                        1024); %% @todo Figure out what the new default should be.
+                %% There are queues already. Write the historic RabbitMQ
+                %% default of 16384 for forward compatibility. Historic
+                %% default calculated as trunc(math:pow(2,?REL_SEQ_BITS)).
+                _ ->
+                    16384
+            end,
+            rabbit_log:info("Setting segment_entry_count for vhost '~s' with ~b queues to '~b'",
+                            [VHost, length(QueueDirs), SegmentEntryCount]),
+            file:write_file(Path, io_lib:format("{segment_entry_count, ~b}.", [SegmentEntryCount]))
+    end.
+
+read_config(VHost) ->
+    {ok, Config} = file:consult(config_file_path(VHost)),
+    maps:from_list(Config).
+
 -define(INFO_KEYS, vhost:info_keys()).
 
 -spec parse_tags(binary() | string() | atom()) -> [atom()].
@@ -377,6 +410,10 @@ msg_store_dir_base() ->
     Dir = rabbit_mnesia:dir(),
     filename:join([Dir, "msg_stores", "vhosts"]).
 
+config_file_path(VHost) ->
+    VHostDir = msg_store_dir_path(VHost),
+    filename:join(VHostDir, ".config").
+
 -spec trim_tag(list() | binary() | atom()) -> atom().
 trim_tag(Val) ->
     rabbit_data_coercion:to_atom(string:trim(rabbit_data_coercion:to_list(Val))).