Organize ra_counters by Ra system

src/ra.erl

@@ -73,8 +73,10 @@
          cast_aux_command/2,
          member_overview/1,
          member_overview/2,
+         % deprecated
          key_metrics/1,
          key_metrics/2,
+         key_metrics/3,
          trigger_compaction/1
         ]).
 
@@ -716,8 +718,7 @@ overview(System) ->
                          wal := Wal}} = Config,
             #{node => node(),
               servers => ra_directory:overview(System),
-              %% TODO:filter counter keys by system
-              counters => ra_counters:overview(),
+              counters => ra_counters:overview(System),
               wal => #{status => lists:nth(5, element(4, sys:get_status(Wal))),
                        open_mem_tables => ets:info(OpenTbls, size)
                       },
@@ -1181,17 +1182,31 @@ member_overview(ServerId) ->
 member_overview(ServerId, Timeout) ->
     ra_server_proc:local_state_query(ServerId, overview, Timeout).
 
+%% @doc Returns a map of key metrics about a Ra member
+%%
+%% For backwards compatibility, since key_metrics/2 can
+%% call key_metrics on a remote node during a rolling upgrade.
+%% In the past, ra_counters were all under the ra namespace
+%% and were not scoped by a Ra system name.
+%%
+%% @param ServerId the Ra server to obtain key metrics for
+%% DEPRECATED: use key_metrics/2
+%% @end
+key_metrics(ServerId) ->
+    key_metrics(ra, ServerId, ?DEFAULT_TIMEOUT).
+
 %% @doc Returns a map of key metrics about a Ra member
 %%
 %% The keys and values may vary depending on what state
 %% the member is in. This function will never call into the
 %% Ra process itself so is likely to return swiftly even
 %% when the Ra process is busy (such as when it is recovering)
 %%
+%% @param System the system name
 %% @param ServerId the Ra server to obtain key metrics for
 %% @end
-key_metrics(ServerId) ->
-    key_metrics(ServerId, ?DEFAULT_TIMEOUT).
+key_metrics(System, ServerId) ->
+    key_metrics(System, ServerId, ?DEFAULT_TIMEOUT).
 
 %% @doc Returns a map of key metrics about a Ra member
 %%
@@ -1200,18 +1215,19 @@ key_metrics(ServerId) ->
 %% Ra process itself so is likely to return swiftly even
 %% when the Ra process is busy (such as when it is recovering)
 %%
+%% @param System the system name
 %% @param ServerId the Ra server to obtain key metrics for
 %% @param Timeout The time to wait for the server to reply
 %% @end
-key_metrics({Name, N} = ServerId, _Timeout) when N == node() ->
+key_metrics(System, {Name, N} = ServerId, _Timeout) when N == node() ->
     Fields = [last_applied,
               commit_index,
               snapshot_index,
               last_written_index,
               last_index,
               commit_latency,
               term],
-    Counters = case ra_counters:counters(ServerId, Fields) of
+    Counters = case ra_counters:counters(System, ServerId, Fields) of
                    undefined ->
                        #{};
                    C -> C
@@ -1230,8 +1246,8 @@ key_metrics({Name, N} = ServerId, _Timeout) when N == node() ->
                               membership => Membership}
             end
     end;
-key_metrics({_, N} = ServerId, Timeout) ->
-    erpc:call(N, ?MODULE, ?FUNCTION_NAME, [ServerId], Timeout).
+key_metrics(System, {_, N} = ServerId, Timeout) ->
+    erpc:call(N, ?MODULE, ?FUNCTION_NAME, [System, ServerId], Timeout).
 
 %% @doc Potentially triggers a major compaction for the provided member
 %% @param ServerId the Ra server to send the request to

src/ra_bench.erl

@@ -15,6 +15,7 @@
 
 -include_lib("eunit/include/eunit.hrl").
 
+-define(BENCH_SYSTEM_NAME, default).
 -define(PIPE_SIZE, 500).
 -define(DATA_SIZE, 256).
 
@@ -149,7 +150,7 @@ start(Name, Nodes) when is_atom(Name) ->
                      initial_members => ServerIds,
                      machine => {module, ?MODULE, #{}}}
                end || N <- Nodes],
-    ra:start_cluster(default, Configs).
+    ra:start_cluster(?BENCH_SYSTEM_NAME, Configs).
 
 prepare() ->
     _ = application:ensure_all_started(ra),
@@ -202,9 +203,10 @@ spawn_client(Parent, Leader, Num, DataSize, Pipe, Counter) ->
               end
       end).
 
-print_metrics(_Name) ->
+print_metrics(Name) ->
     io:format("Node: ~w~n", [node()]),
-    io:format("counters ~p~n", [ra_counters:overview()]).
+    io:format("metrics ~p~n", [ets:lookup(ra_metrics, Name)]),
+    io:format("counters ~p~n", [ra_counters:overview(?BENCH_SYSTEM_NAME)]).
 
 
 

src/ra_counters.erl

@@ -8,51 +8,56 @@
 -include("ra.hrl").
 
 -export([
-         init/0,
-         new/2,
+         init/1,
+         new/4,
          new/3,
-         fetch/1,
+         fetch/2,
          overview/0,
          overview/1,
-         counters/2,
-         delete/1
+         overview/2,
+         counters/3,
+         delete/2
          ]).
 
 -type name() :: term().
 
 
--spec init() -> ok.
-init() ->
+-spec init(atom()) -> ok.
+init(System) ->
     _ = application:ensure_all_started(seshat),
-    _ = seshat:new_group(ra),
+    _ = seshat:new_group(System),
     persistent_term:put(?FIELDSPEC_KEY, ?RA_COUNTER_FIELDS),
     ok.
 
--spec new(name(), seshat:fields_spec()) ->
+-spec new(atom(), name(), seshat:fields_spec()) ->
     counters:counters_ref().
-new(Name, FieldsSpec) ->
-    seshat:new(ra, Name, FieldsSpec).
+new(System, Name, FieldsSpec) ->
+    seshat:new(System, Name, FieldsSpec).
 
-new(Name, FieldsSpec, MetricLabels) ->
-    seshat:new(ra, Name, FieldsSpec, MetricLabels).
+new(System, Name, FieldsSpec, MetricLabels) ->
+    seshat:new(System, Name, FieldsSpec, MetricLabels).
 
--spec fetch(name()) -> undefined | counters:counters_ref().
-fetch(Name) ->
-    seshat:fetch(ra, Name).
+-spec fetch(atom(), name()) -> undefined | counters:counters_ref().
+fetch(System, Name) ->
+    seshat:fetch(System, Name).
 
--spec delete(term()) -> ok.
-delete(Name) ->
-    seshat:delete(ra, Name).
+-spec delete(atom(), term()) -> ok.
+delete(System, Name) ->
+    seshat:delete(System, Name).
 
 -spec overview() -> #{name() => #{atom() => non_neg_integer()}}.
 overview() ->
-    seshat:counters(ra).
+    seshat:counters(quorum_queues).
 
--spec overview(name()) -> #{atom() => non_neg_integer()} | undefined.
-overview(Name) ->
-    seshat:counters(ra, Name).
+-spec overview(atom()) -> #{atom() => non_neg_integer()} | undefined.
+overview(System) ->
+    seshat:counters(System).
 
--spec counters(name(), [atom()]) ->
+-spec overview(atom(), name()) -> #{atom() => non_neg_integer()} | undefined.
+overview(System, Name) ->
+    seshat:counters(System, Name).
+
+-spec counters(atom(), name(), [atom()]) ->
     #{atom() => non_neg_integer()} | undefined.
-counters(Name, Fields) ->
-    seshat:counters(ra, Name, Fields).
+counters(System, Name, Fields) ->
+    seshat:counters(System, Name, Fields).

src/ra_log_segment_writer.erl

@@ -114,7 +114,7 @@ init([#{data_dir := DataDir,
         name := SegWriterName,
         system := System} = Conf]) ->
     process_flag(trap_exit, true),
-    CRef = ra_counters:new(SegWriterName, ?COUNTER_FIELDS,
+    CRef = ra_counters:new(System, SegWriterName, ?COUNTER_FIELDS,
                            #{ra_system => System,
                              module => ?MODULE}),
     SegmentConf = maps:get(segment_conf, Conf, #{}),

src/ra_log_sup.erl

@@ -57,7 +57,8 @@ init([#{data_dir := DataDir,
 
 make_wal_conf(#{data_dir := DataDir,
                 name := System,
-                names := #{} = Names} = Cfg) ->
+                names := #{} = Names0} = Cfg) ->
+    Names = Names0#{system => System},
     WalDir = case Cfg of
                  #{wal_data_dir := D} -> D;
                  _ -> DataDir

src/ra_log_wal.erl

@@ -266,19 +266,22 @@ init(#{system := System,
       garbage_collect := Gc,
       min_heap_size := MinHeapSize,
       min_bin_vheap_size := MinBinVheapSize,
+      system := System,
       names := #{wal := WalName,
                  segment_writer := SegWriter,
-                 open_mem_tbls := MemTablesName} = Names} =
+                 open_mem_tbls := MemTablesName} = Names0} =
         merge_conf_defaults(Conf0),
     ?NOTICE("WAL in ~ts initialising with name ~ts", [System, WalName]),
     process_flag(trap_exit, true),
     % given ra_log_wal is effectively a fan-in sink it is likely that it will
     % at times receive large number of messages from a large number of
     % writers
+    Names = Names0#{system => System},
     process_flag(message_queue_data, off_heap),
     process_flag(min_bin_vheap_size, MinBinVheapSize),
     process_flag(min_heap_size, MinHeapSize),
-    CRef = ra_counters:new(WalName,
+    CRef = ra_counters:new(System,
+                           WalName,
                            ?COUNTER_FIELDS,
                            #{ra_system => System, module => ?MODULE}),
     Conf = #conf{dir = Dir,
@@ -329,7 +332,7 @@ terminate(Reason, #state{conf = #conf{system = System}} = State) ->
 
 format_status(#state{conf = #conf{sync_method = SyncMeth,
                                   compute_checksums = Cs,
-                                  names = #{wal := WalName},
+                                  names = #{system := System, wal := WalName},
                                   max_size_bytes = MaxSize},
                      writers = Writers,
                      wal = #wal{file_size = FSize,
@@ -340,7 +343,7 @@ format_status(#state{conf = #conf{sync_method = SyncMeth,
       filename => filename:basename(Fn),
       current_size => FSize,
       max_size_bytes => MaxSize,
-      counters => ra_counters:overview(WalName)
+      counters => ra_counters:overview(System, WalName)
      }.
 
 %% Internal
@@ -1094,4 +1097,3 @@ named_cast(Wal, Msg) ->
         Pid ->
             named_cast(Pid, Msg)
     end.
-

src/ra_metrics_ets.erl

@@ -38,7 +38,6 @@ init([]) ->
                    {write_concurrency, true},
                    public],
     _ = ets:new(ra_log_metrics, [set | TableFlags]),
-    ok = ra_counters:init(),
     ok = ra_leaderboard:init(),
 
     %% Table for ra processes to record their current snapshot index so that

src/ra_server_proc.erl

@@ -330,13 +330,14 @@ init(#{reply_to := ReplyTo} = Config) ->
 do_init(#{id := Id,
           uid := UId,
           parent := ParentPid,
-          cluster_name := ClusterName} = Config0)
+          cluster_name := ClusterName,
+          system_config := #{name := System}} = Config0)
   when is_pid(ParentPid) ->
     Key = ra_lib:ra_server_id_to_local_name(Id),
     true = ets:insert(ra_state, {Key, init, unknown}),
     process_flag(trap_exit, true),
     MetricLabels = maps:get(metrics_labels, Config0, #{}),
-    Config = maps:merge(config_defaults(Id, MetricLabels), Config0),
+    Config = maps:merge(config_defaults(System, Id, MetricLabels), Config0),
     #{counter := Counter,
       system_config := #{names := Names} = SysConf} = Config,
     MsgQData = maps:get(message_queue_data, SysConf, off_heap),
@@ -1183,10 +1184,11 @@ handle_event(_EventType, EventContent, StateName, State) ->
     {next_state, StateName, State}.
 
 terminate(Reason, StateName,
-          #state{conf = #conf{name = Key,
-                              cluster_name = ClusterName,
-                              worker_pid = WorkerPid},
-                 server_state = ServerState} = State) ->
+              #state{conf = #conf{name = Key,
+                                  cluster_name = ClusterName,
+                                  worker_pid = WorkerPid},
+                     server_state = #{cfg := #cfg{
+                         system_config = #{name := System}}} = ServerState} = State) ->
     ?DEBUG("~ts: terminating with ~w in state ~w",
            [log_id(State), Reason, StateName]),
     #{names := #{server_sup := SrvSup,
@@ -1212,7 +1214,7 @@ terminate(Reason, StateName,
             catch ra_directory:unregister_name(Names, UId),
             _ = ra_server:terminate(ServerState, Reason),
             catch ra_log_meta:delete_sync(MetaName, UId),
-            catch ra_counters:delete(Id),
+            catch ra_counters:delete(System, Id),
             Self = self(),
             %% we have to terminate the child spec from the supervisor as it
             %% won't do this automatically, even for transient children
@@ -1630,7 +1632,7 @@ handle_effect(_RaftState, {reply, Reply}, {call, From}, State, Actions) ->
 handle_effect(_RaftState, {reply, _From, _Reply}, _EvtType, State, Actions) ->
     {State, Actions};
 handle_effect(leader, {send_snapshot, {_, ToNode} = To, {SnapState, _Id, Term}}, _,
-              #state{server_state = SS0,
+              #state{server_state = #{cfg := #cfg{system_config = #{name := System}}} = SS0,
                      monitors = Monitors,
                      conf = #conf{snapshot_chunk_size = ChunkSize,
                                   log_id = LogId,
@@ -1653,7 +1655,7 @@ handle_effect(leader, {send_snapshot, {_, ToNode} = To, {SnapState, _Id, Term}},
                            end,
             Id = ra_server:id(SS0),
             Pid = spawn(fun () ->
-                                send_snapshots(Id, Term, To,
+                                send_snapshots(System, Id, Term, To,
                                                ChunkSize, InstallSnapTimeout,
                                                SnapState, Machine, LogId)
                         end),
@@ -1972,15 +1974,15 @@ reset_commit_rate({LI, _LastCI, _LastRate}, ServerState) ->
     #{commit_index := CI} = ServerState,
     {ra_li:reset(LI), CI, 0.0}.
 
-config_defaults(ServerId, MetricLabels) ->
-    Counter = case ra_counters:fetch(ServerId) of
-                  undefined ->
-                      ra_counters:new(ServerId,
-                                      {persistent_term, ?FIELDSPEC_KEY},
-                                      MetricLabels);
-                  C ->
-                      C
-              end,
+config_defaults(System, ServerId, MetricLabels) ->
+    Counter = case ra_counters:fetch(System, ServerId) of
+                    undefined ->
+                        ra_counters:new(System, ServerId,
+                                        {persistent_term, ?FIELDSPEC_KEY},
+                                        MetricLabels);
+                    C ->
+                        C
+                end,
     #{broadcast_time => ?DEFAULT_BROADCAST_TIME,
       tick_timeout => ?TICK_INTERVAL_MS,
       install_snap_rpc_timeout => ?INSTALL_SNAP_RPC_TIMEOUT,
@@ -2044,7 +2046,7 @@ read_entries0(From, Idxs, #state{server_state = #{log := Log}} = State) ->
                                                end),
     {keep_state, State, [{reply, From, {ok, ReadState}}]}.
 
-send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
+send_snapshots(System, Id, Term, {_, ToNode} = To, ChunkSize,
                InstallTimeout, SnapState, Machine, LogId) ->
     Context = ra_snapshot:context(SnapState, ToNode),
     {ok, #{index := SnapIdx,
@@ -2093,7 +2095,7 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
                     %% there are live indexes to send before the snapshot
                     #{last_applied := LastApplied} =
                         erpc:call(ToNode, ra_counters, counters,
-                                  [To, [last_applied]]),
+                                  [System, To, [last_applied]]),
                     %% remove all indexes lower than the target's last applied
                     Indexes = ra_seq:floor(LastApplied + 1, Indexes0),
                     ?DEBUG("~ts: sending ~b live indexes in the range ~w to ~w ",

src/ra_server_sup_sup.erl

@@ -187,7 +187,7 @@ delete_server_rpc(System, RaName) ->
             catch ets:delete(ra_log_snapshot_state, UId),
             catch ets:delete(ra_state, RaName),
             catch ets:delete(ra_open_file_metrics, Pid),
-            catch ra_counters:delete({RaName, node()}),
+            catch ra_counters:delete(System, {RaName, node()}),
             catch ra_leaderboard:clear(RaName),
             ok
     end.