diff --git a/src/ra.erl b/src/ra.erl index a6547da79..0f0f586f9 100644 --- a/src/ra.erl +++ b/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,6 +1182,19 @@ 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 @@ -1188,10 +1202,11 @@ member_overview(ServerId, Timeout) -> %% 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,10 +1215,11 @@ 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, @@ -1211,7 +1227,7 @@ key_metrics({Name, N} = ServerId, _Timeout) when N == node() -> 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 diff --git a/src/ra_bench.erl b/src/ra_bench.erl index eaf467998..f39f02078 100644 --- a/src/ra_bench.erl +++ b/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)]). diff --git a/src/ra_counters.erl b/src/ra_counters.erl index 674f6ba46..13ed4fa18 100644 --- a/src/ra_counters.erl +++ b/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). diff --git a/src/ra_log_segment_writer.erl b/src/ra_log_segment_writer.erl index 45bc42938..83d67bab0 100644 --- a/src/ra_log_segment_writer.erl +++ b/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, #{}), diff --git a/src/ra_log_sup.erl b/src/ra_log_sup.erl index fa5409371..b8620fc90 100644 --- a/src/ra_log_sup.erl +++ b/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 diff --git a/src/ra_log_wal.erl b/src/ra_log_wal.erl index ce9fa7ba3..04b8012ca 100644 --- a/src/ra_log_wal.erl +++ b/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. - diff --git a/src/ra_metrics_ets.erl b/src/ra_metrics_ets.erl index c852998c8..4117b2012 100644 --- a/src/ra_metrics_ets.erl +++ b/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 diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl index deb1f8fdf..e6102b772 100644 --- a/src/ra_server_proc.erl +++ b/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 ", diff --git a/src/ra_server_sup_sup.erl b/src/ra_server_sup_sup.erl index 796d5d9a7..94fb24560 100644 --- a/src/ra_server_sup_sup.erl +++ b/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. diff --git a/src/ra_system.erl b/src/ra_system.erl index 6b2080cab..3a7a16fbe 100644 --- a/src/ra_system.erl +++ b/src/ra_system.erl @@ -15,7 +15,8 @@ stop_default/0 ]). --type names() :: #{wal := atom(), +-type names() :: #{system := atom(), + wal := atom(), wal_sup := atom(), log_sup := atom(), log_ets := atom(), @@ -145,7 +146,8 @@ default_config() -> low_priority_commands_flush_size => LowPriorityCommandsFlushSize, low_priority_commands_in_memory_size => LowPriorityInMemSize, machine_upgrade_strategy => MachineUpgradeStrategy, - names => #{wal => ra_log_wal, + names => #{system => default, + wal => ra_log_wal, wal_sup => ra_log_wal_sup, log_sup => ra_log_sup, log_ets => ra_log_ets, @@ -158,7 +160,8 @@ default_config() -> }}. derive_names(SysName) when is_atom(SysName) -> - #{wal => derive(SysName, <<"log_wal">>), + #{system => SysName, + wal => derive(SysName, <<"log_wal">>), wal_sup => derive(SysName, <<"log_wal_sup">>), log_sup => derive(SysName, <<"log_sup">>), log_ets => derive(SysName, <<"log_ets">>), diff --git a/src/ra_system_sup.erl b/src/ra_system_sup.erl index ba89abe72..099f1a2d1 100644 --- a/src/ra_system_sup.erl +++ b/src/ra_system_sup.erl @@ -40,6 +40,7 @@ init(#{data_dir := DataDir, start => {ra_server_sup_sup, start_link, [Cfg]}}, Recover = #{id => ra_system_recover, start => {ra_system_recover, start_link, [maps:get(name, Cfg)]}}, + ra_counters:init(Name), {ok, {SupFlags, [Ets, RaLogSup, RaServerSupSup, Recover]}}; {error, Code} -> ?ERR("Failed to create Ra data directory at '~ts', file system operation error: ~p", [DataDir, Code]), diff --git a/test/coordination_SUITE.erl b/test/coordination_SUITE.erl index 669450027..5188765f0 100644 --- a/test/coordination_SUITE.erl +++ b/test/coordination_SUITE.erl @@ -615,7 +615,7 @@ nonvoter_catches_up(Config) -> ?assertMatch(#{Group := #{membership := promotable}}, rpc:call(NodeC, ra_directory, overview, [?SYS])), ?assertMatch(#{membership := promotable}, - ra:key_metrics(C)), + ra:key_metrics(?SYS, C)), ?assertMatch({ok, #{membership := promotable}, _}, ra:member_overview(C)), @@ -627,7 +627,7 @@ nonvoter_catches_up(Config) -> ?assertMatch(#{Group := #{membership := voter}}, rpc:call(NodeC, ra_directory, overview, [?SYS])), ?assertMatch(#{membership := voter}, - ra:key_metrics(C)), + ra:key_metrics(?SYS, C)), stop_peers(Peers), ok. @@ -651,7 +651,7 @@ nonvoter_catches_up_after_restart(Config) -> ?assertMatch(#{Group := #{membership := promotable}}, rpc:call(NodeC, ra_directory, overview, [?SYS])), ?assertMatch(#{membership := promotable}, - ra:key_metrics(C)), + ra:key_metrics(?SYS, C)), ?assertMatch({ok, #{membership := promotable}, _}, ra:member_overview(C)), ok = ra:stop_server(?SYS, C), @@ -665,7 +665,7 @@ nonvoter_catches_up_after_restart(Config) -> ?assertMatch(#{Group := #{membership := voter}}, rpc:call(NodeC, ra_directory, overview, [?SYS])), ?assertMatch(#{membership := voter}, - ra:key_metrics(C)), + ra:key_metrics(?SYS, C)), stop_peers(Peers), ok. @@ -689,7 +689,7 @@ nonvoter_catches_up_after_leader_restart(Config) -> ?assertMatch(#{Group := #{membership := promotable}}, rpc:call(NodeC, ra_directory, overview, [?SYS])), ?assertMatch(#{membership := promotable}, - ra:key_metrics(C)), + ra:key_metrics(?SYS, C)), ?assertMatch({ok, #{membership := promotable}, _}, ra:member_overview(C)), ok = ra:stop_server(?SYS, Leader), @@ -703,7 +703,7 @@ nonvoter_catches_up_after_leader_restart(Config) -> ?assertMatch(#{Group := #{membership := voter}}, rpc:call(NodeC, ra_directory, overview, [?SYS])), ?assertMatch(#{membership := voter}, - ra:key_metrics(C)), + ra:key_metrics(?SYS, C)), stop_peers(Peers), ok. @@ -726,7 +726,7 @@ key_metrics(Config) -> timer:sleep(100), TestId = lists:last(Started), ok = ra:stop_server(?SYS, TestId), - StoppedMetrics = ra:key_metrics(TestId), + StoppedMetrics = ra:key_metrics(?SYS, TestId), ct:pal("StoppedMetrics ~p", [StoppedMetrics]), ?assertMatch(#{state := noproc, last_applied := LA, @@ -740,12 +740,12 @@ key_metrics(Config) -> {ok, _, _} = ra:process_command(Leader, {data, Data}), await_condition( fun () -> - Metrics = ra:key_metrics(TestId), + Metrics = ra:key_metrics(?SYS, TestId), ct:pal("FollowerMetrics ~p", [Metrics]), follower == maps:get(state, Metrics) end, 200), [begin - M = ra:key_metrics(S), + M = ra:key_metrics(?SYS, S), ct:pal("Metrics ~p", [M]), ?assertMatch(#{state := _, last_applied := LA, @@ -884,7 +884,7 @@ recover_from_checkpoint(Config) -> checkpoints_promoted := 0 } when B > 0, ct_rpc:call(N, ra_counters, counters, - [ServerId, CounterKeys])) + [?SYS, ServerId, CounterKeys])) end || {_, N} = ServerId <- ServerIds], @@ -924,7 +924,7 @@ recover_from_checkpoint(Config) -> checkpoints_promoted := 1 } when B > 0, ct_rpc:call(N, ra_counters, counters, - [ServerId, CounterKeys])) + [?SYS, ServerId, CounterKeys])) end || {_, N} = ServerId <- ServerIds], %% Restart the servers: the servers should be able to recover from the %% snapshot which was promoted from a checkpoint. @@ -1251,13 +1251,13 @@ stopped_wal_causes_leader_change(Config, RecoverStrat) -> %% kill the wal until the system crashes and the current member is terminated %% and another leader is elected - #{term := Term} = ra:key_metrics(Follower), + #{term := Term} = ra:key_metrics(?SYS, Follower), await_condition(fun () -> WalPid = ct_rpc:call(LeaderNode, erlang, whereis, [ra_log_wal]), true = ct_rpc:call(LeaderNode, erlang, exit, [WalPid, kill]), - #{term := T} = ra:key_metrics(Follower), + #{term := T} = ra:key_metrics(?SYS, Follower), T > Term andalso (begin P = ct_rpc:call(LeaderNode, erlang, whereis, [LeaderName]), diff --git a/test/ra_SUITE.erl b/test/ra_SUITE.erl index 2523edf5a..c40604177 100644 --- a/test/ra_SUITE.erl +++ b/test/ra_SUITE.erl @@ -162,7 +162,7 @@ pipeline_commands(Config) -> ?assertMatch(#{last_index := I, last_applied := I, last_written_index := I, - commit_index := I}, ra:key_metrics(N1)), + commit_index := I}, ra:key_metrics(?SYS, N1)), terminate_cluster([N1]). stop_server_idemp(Config) -> diff --git a/test/ra_kv_SUITE.erl b/test/ra_kv_SUITE.erl index f72d48c00..4d727b239 100644 --- a/test/ra_kv_SUITE.erl +++ b/test/ra_kv_SUITE.erl @@ -198,7 +198,7 @@ snapshot_replication(_Config) -> ra:member_overview(KvId3), Kv1LastIndex == LastIdx end, 100, 100), - ct:pal("counters ~p", [ra_counters:counters(KvId3, [last_applied])]), + ct:pal("counters ~p", [ra_counters:counters(?SYS, KvId3, [last_applied])]), %% ensure Kv3 did not crash during snapshot replication ?assertEqual(KvId3Pid, whereis(Kv3)), @@ -293,13 +293,13 @@ basics(_Config) -> ok = ra_lib:retry( fun () -> #{major_compactions := Maj} = - ra_counters:counters(KvId, [major_compactions]), + ra_counters:counters(?SYS, KvId, [major_compactions]), Maj == 1 end, 10, 100), {ok, {Reads5, _}} = ra_server_proc:read_entries(KvId, [LastIdx | Live], undefined, 1000), ?assertEqual(Reads4, Reads5), - ct:pal("counters ~p", [ra_counters:overview(KvId)]), + ct:pal("counters ~p", [ra_counters:overview(?SYS, KvId)]), ok = ra_kv:remove_member(?SYS, KvId2, KvId), ct:pal("members ~p", [ra:members(KvId)]), ra:delete_cluster([KvId]), diff --git a/test/ra_log_2_SUITE.erl b/test/ra_log_2_SUITE.erl index f5dea8506..0a06a009c 100644 --- a/test/ra_log_2_SUITE.erl +++ b/test/ra_log_2_SUITE.erl @@ -10,6 +10,8 @@ %% %% +-define(SYS, default). + all() -> [ {group, random}, @@ -307,15 +309,15 @@ delete_during_segment_flush(Config) -> ok. read_one(Config) -> - ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS), - Log0 = ra_log_init(Config, #{counter => ra_counters:fetch(?FUNCTION_NAME)}), + ra_counters:new(?SYS, ?FUNCTION_NAME, ?RA_COUNTER_FIELDS), + Log0 = ra_log_init(Config, #{counter => ra_counters:fetch(?SYS, ?FUNCTION_NAME)}), Log1 = append_n(1, 2, 1, Log0), % ensure the written event is delivered Log2 = deliver_all_log_events(Log1, 200), {[_], Log} = ra_log_take(1, 1, Log2), % read out of range #{?FUNCTION_NAME := #{read_mem_table := M1, - read_segment := M2}} = ra_counters:overview(), + read_segment := M2}} = ra_counters:overview(?SYS), % read two entries ?assertEqual(1, M1 + M2), ra_log:close(Log), @@ -341,8 +343,8 @@ take_after_overwrite_and_init(Config) -> validate_sequential_fold(Config) -> - ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS), - Log0 = ra_log_init(Config, #{counter => ra_counters:fetch(?FUNCTION_NAME), + ra_counters:new(?SYS, ?FUNCTION_NAME, ?RA_COUNTER_FIELDS), + Log0 = ra_log_init(Config, #{counter => ra_counters:fetch(?SYS, ?FUNCTION_NAME), max_open_segments => 2}), % write 1000 entries Log1 = append_and_roll(1, 500, 1, Log0), @@ -366,7 +368,7 @@ validate_sequential_fold(Config) -> #{read_mem_table := M1, open_segments := 2, %% as this is the max - read_segment := M4} = O = ra_counters:overview(?FUNCTION_NAME), + read_segment := M4} = O = ra_counters:overview(?SYS, ?FUNCTION_NAME), ct:pal("counters ~p", [O]), ?assertEqual(1000, M1 + M4), @@ -374,9 +376,9 @@ validate_sequential_fold(Config) -> ok. validate_reads_for_overlapped_writes(Config) -> - ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS), - Log0 = ra_log_init(Config, #{counter => ra_counters:fetch(?FUNCTION_NAME) - }), + ra_counters:new(?SYS, ?FUNCTION_NAME, ?RA_COUNTER_FIELDS), + Log0 = ra_log_init(Config, + #{counter => ra_counters:fetch(?SYS, ?FUNCTION_NAME)}), % write a segment and roll 1 - 299 - term 1 Log1 = write_and_roll(1, 300, 1, Log0), % write 300 - 399 in term 1 - no roll @@ -397,11 +399,11 @@ validate_reads_for_overlapped_writes(Config) -> Log8 = validate_fold(200, 550, 2, Log7), #{?FUNCTION_NAME := #{read_mem_table := M1, - read_segment := M2}} = ra_counters:overview(), + read_segment := M2}} = ra_counters:overview(?SYS), ?assertEqual(550, M1 + M2), ra_log:close(Log8), %% re open to test init with overlapping segments - Log = ra_log_init(Config, #{counter => ra_counters:fetch(?FUNCTION_NAME)}), + Log = ra_log_init(Config, #{counter => ra_counters:fetch(?SYS, ?FUNCTION_NAME)}), ra_log:close(Log), ok. @@ -726,9 +728,9 @@ writes_lower_than_snapshot_index_are_dropped(Config) -> ok. updated_segment_can_be_read(Config) -> - ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS), + ra_counters:new(?SYS, ?FUNCTION_NAME, ?RA_COUNTER_FIELDS), Log0 = ra_log_init(Config, - #{counter => ra_counters:fetch(?FUNCTION_NAME), + #{counter => ra_counters:fetch(?SYS, ?FUNCTION_NAME), min_snapshot_interval => 1}), %% append a few entries Log2 = append_and_roll(1, 5, 1, Log0), @@ -742,6 +744,9 @@ updated_segment_can_be_read(Config) -> % this should return all entries {Entries1, _} = ra_log_take(1, 15, Log4), ?assertEqual(15, length(Entries1)), + ct:pal("Entries: ~p", [Entries]), + ct:pal("Entries1: ~p", [Entries1]), + ct:pal("Counters ~p", [ra_counters:overview(?SYS, ?FUNCTION_NAME)]), ?assertEqual(15, length(Entries1)), ok. diff --git a/test/ra_log_ets_SUITE.erl b/test/ra_log_ets_SUITE.erl index bffc37927..3c13e3b57 100644 --- a/test/ra_log_ets_SUITE.erl +++ b/test/ra_log_ets_SUITE.erl @@ -29,6 +29,8 @@ groups() -> {tests, [], all_tests()} ]. +-define(SYS, default). + init_per_suite(Config) -> Config. @@ -48,7 +50,7 @@ init_per_testcase(TestCase, Config) -> SysCfg = (ra_system:default_config())#{data_dir => Dir}, ra_system:store(SysCfg), _ = ra_log_ets:start_link(SysCfg), - ra_counters:init(), + ra_counters:init(?SYS), [{sys_cfg, SysCfg}, {data_dir, Dir} | Config]. end_per_testcase(_TestCase, _Config) -> diff --git a/test/ra_log_segment_writer_SUITE.erl b/test/ra_log_segment_writer_SUITE.erl index aaa33de57..781e0c9bd 100644 --- a/test/ra_log_segment_writer_SUITE.erl +++ b/test/ra_log_segment_writer_SUITE.erl @@ -62,10 +62,10 @@ init_per_testcase(TestCase, Config) -> logger:set_primary_config(level, all), PrivDir = ?config(priv_dir, Config), Dir = filename:join(PrivDir, TestCase), - SysCfg = ra_system:default_config(), + #{name := System} = SysCfg = ra_system:default_config(), ra_system:store(SysCfg), _ = ra_log_ets:start_link(SysCfg), - ra_counters:init(), + ra_counters:init(System), UId = atom_to_binary(TestCase, utf8), ok = ra_directory:register_name(default, UId, self(), undefined, TestCase, TestCase), diff --git a/test/ra_log_segments_SUITE.erl b/test/ra_log_segments_SUITE.erl index eea2d1420..cf5debbed 100644 --- a/test/ra_log_segments_SUITE.erl +++ b/test/ra_log_segments_SUITE.erl @@ -46,13 +46,15 @@ groups() -> {tests, [], all_tests()} ]. +-define(SYS, default). + init_per_testcase(TestCase, Config) -> PrivDir = ?config(priv_dir, Config), Dir = filename:join(PrivDir, TestCase), ok = ra_lib:make_dir(Dir), CompConf = #{max_count => 128, max_size => 128_000}, - ra_counters:init(), + ra_counters:init(?SYS), [{uid, atom_to_binary(TestCase, utf8)}, {comp_conf, CompConf}, {test_case, TestCase}, @@ -653,7 +655,7 @@ major_strategy_num_minors(Config) -> ], - Counters = ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS), + Counters = ra_counters:new(?SYS, ?FUNCTION_NAME, ?RA_COUNTER_FIELDS), CompConf = #{max_count => 16, max_size => 128_000, major_strategy => {num_minors, 2}}, @@ -662,7 +664,7 @@ major_strategy_num_minors(Config) -> _Segs = run_scenario([{seg_conf, SegConf} | Config], Segs0, Scen), ?assertMatch(#{major_compactions := 1, minor_compactions := 2}, - ra_counters:counters(?FUNCTION_NAME, + ra_counters:counters(?SYS, ?FUNCTION_NAME, [minor_compactions, major_compactions])), diff --git a/test/ra_log_wal_SUITE.erl b/test/ra_log_wal_SUITE.erl index 0f4850864..5c1d9216c 100644 --- a/test/ra_log_wal_SUITE.erl +++ b/test/ra_log_wal_SUITE.erl @@ -84,7 +84,7 @@ init_per_group(Group, Config) -> SysCfg = (ra_system:default_config())#{data_dir => Dir}, ra_system:store(SysCfg), ra_directory:init(?SYS), - ra_counters:init(), + ra_counters:init(?SYS), % application:ensure_all_started(lg), SyncMethod = case Group of @@ -108,7 +108,7 @@ init_per_testcase(TestCase, Config) -> Sys = ?config(sys_cfg, Config), Dir = filename:join([PrivDir, M, TestCase]), {ok, Ets} = ra_log_ets:start_link(Sys), - ra_counters:init(), + ra_counters:init(?SYS), UId = atom_to_binary(TestCase, utf8), ok = ra_directory:register_name(default, UId, self(), undefined, TestCase, TestCase),