Merge pull request #545 from rabbitmq/expose-ra-counters

Expose ra counters, remove ra_metrics

Author: kjnilsson

Makefile

@@ -10,7 +10,7 @@ ESCRIPT_EMU_ARGS = -noinput -setcookie ra_fifo_cli
 
 dep_gen_batch_server = hex 0.8.9
 dep_aten = hex 0.6.0
-dep_seshat = hex 0.6.0
+dep_seshat = hex 1.0.0
 DEPS = aten gen_batch_server seshat
 
 TEST_DEPS = proper meck eunit_formatters inet_tcp_proxy

README.md

@@ -53,7 +53,7 @@ which algorithms such as Raft significantly benefit from.
 
 This library was primarily developed as the foundation of a replication layer for
 [quorum queues](https://rabbitmq.com/quorum-queues.html) in RabbitMQ, and today
-also powers [RabbitMQ streams](https://rabbitmq.com/streams.html) and [Khepri](https://github.com/rabbitmq/khepri). 
+also powers [RabbitMQ streams](https://rabbitmq.com/streams.html) and [Khepri](https://github.com/rabbitmq/khepri).
 
 The design it aims to replace uses
 a variant of [Chain Based Replication](https://www.cs.cornell.edu/home/rvr/papers/OSDI04.pdf)
@@ -405,9 +405,14 @@ is available in a separate repository.
     </tr>
     <tr>
         <td>metrics_key</td>
-        <td>Metrics key. The key used to write metrics into the ra_metrics table.</td>
+        <td>DEPRECATED: Metrics key. The key was used to write metrics into the ra_metrics table. Metrics are now in Seshat.</td>
         <td>Atom</td>
     </tr>
+    <tr>
+        <td>metrics_labels</td>
+        <td>Metrics labels. Labels that are later returned with metrics for this server (eg. `#{vhost => ..., queue => ...}` for quorum queues).</td>
+        <td>Map</td>
+    </tr>
     <tr>
         <td>low_priority_commands_flush_size</td>
         <td>

rebar.config

@@ -1,7 +1,7 @@
 {deps, [
         {gen_batch_server, "0.8.9"},
         {aten, "0.6.0"},
-        {seshat, "0.6.0"}
+        {seshat, "1.0.0"}
 ]}.
 
 {profiles,

rebar.lock

@@ -1,14 +1,14 @@
 {"1.2.0",
 [{<<"aten">>,{pkg,<<"aten">>,<<"0.6.0">>},0},
  {<<"gen_batch_server">>,{pkg,<<"gen_batch_server">>,<<"0.8.9">>},0},
- {<<"seshat">>,{pkg,<<"seshat">>,<<"0.6.0">>},0}]}.
+ {<<"seshat">>,{pkg,<<"seshat">>,<<"1.0.0">>},0}]}.
 [
 {pkg_hash,[
  {<<"aten">>, <<"7A57B275A6DAF515AC3683FB9853E280B4D0DCDD74292FD66AC4A01C8694F8C7">>},
  {<<"gen_batch_server">>, <<"1C6BC0F530BF8C17E8B4ACC20C2CC369FFA5BEE2B46DE01E21410745F24B1BC9">>},
- {<<"seshat">>, <<"3172EB1D7A2A4F66108CD6933A4E465AFF80F84AA90ED83F047B92F636123CCD">>}]},
+ {<<"seshat">>, <<"7CFD882C6BD0DA4B8C91AA339351727FE982EF7260A7B4E5B2A3E6551BE30E52">>}]},
 {pkg_hash_ext,[
  {<<"aten">>, <<"5F39A164206AE3F211EF5880B1F7819415686436E3229D30B6A058564FBAA168">>},
  {<<"gen_batch_server">>, <<"C8581FE4A4B6BCCF91E53CE6A8C7E6C27C8C591BAB5408B160166463F5579C22">>},
- {<<"seshat">>, <<"7CEF700F92831DD7CAE6A6DD223CCC55AC88ECCE0631EE9AB0F2B5FB70E79B90">>}]}
+ {<<"seshat">>, <<"95018806AB04E10563157532F01C523F202C50DEC0E72C6C999064CE829CFF24">>}]}
 ].

src/ra.hrl

@@ -389,7 +389,7 @@
           "The last index of the log."},
          {last_written_index, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, counter,
           "The last fully written and fsynced index of the log."},
-         {commit_latency, ?C_RA_SVR_METRIC_COMMIT_LATENCY, gauge,
+         {commit_latency, ?C_RA_SVR_METRIC_COMMIT_LATENCY, {gauge, time_ms},
           "Approximate time taken from an entry being written to the log until it is committed."},
          {term, ?C_RA_SVR_METRIC_TERM, counter, "The current term."},
          {checkpoint_index, ?C_RA_SVR_METRIC_CHECKPOINT_INDEX, counter,

src/ra_bench.erl

@@ -199,9 +199,8 @@ spawn_client(Parent, Leader, Num, DataSize, Pipe, Counter) ->
               end
       end).
 
-print_metrics(Name) ->
+print_metrics(_Name) ->
     io:format("Node: ~w~n", [node()]),
-    io:format("metrics ~p~n", [ets:lookup(ra_metrics, Name)]),
     io:format("counters ~p~n", [ra_counters:overview()]).
 
 
@@ -220,4 +219,3 @@ print_metrics(Name) ->
 %     GzFile = Base ++ ".gz.*",
 %     lg_callgrind:profile_many(GzFile, Base ++ ".out",#{}),
 %     ok.
-

src/ra_counters.erl

@@ -10,6 +10,7 @@
 -export([
          init/0,
          new/2,
+         new/3,
          fetch/1,
          overview/0,
          overview/1,
@@ -32,6 +33,9 @@ init() ->
 new(Name, FieldsSpec) ->
     seshat:new(ra, Name, FieldsSpec).
 
+new(Name, FieldsSpec, MetricLabels) ->
+    seshat:new(ra, Name, FieldsSpec, MetricLabels).
+
 -spec fetch(name()) -> undefined | counters:counters_ref().
 fetch(Name) ->
     seshat:fetch(ra, Name).
@@ -42,11 +46,11 @@ delete(Name) ->
 
 -spec overview() -> #{name() => #{atom() => non_neg_integer()}}.
 overview() ->
-    seshat:overview(ra).
+    seshat:counters(ra).
 
--spec overview(name()) -> #{atom() => non_neg_integer()}.
+-spec overview(name()) -> #{atom() => non_neg_integer()} | undefined.
 overview(Name) ->
-    seshat:overview(ra, Name).
+    seshat:counters(ra, Name).
 
 -spec counters(name(), [atom()]) ->
     #{atom() => non_neg_integer()} | undefined.

src/ra_log_segment_writer.erl

@@ -115,7 +115,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(SegWriterName, ?COUNTER_FIELDS, #{ra_system => System, module => ?MODULE}),
     SegmentConf = maps:get(segment_conf, Conf, #{}),
     maybe_upgrade_segment_file_names(System, DataDir),
     {ok, #state{system = System,
@@ -554,4 +554,3 @@ maybe_upgrade_segment_file_names(System, DataDir) ->
         true ->
             ok
     end.
-

src/ra_log_sup.erl

@@ -56,7 +56,7 @@ init([#{data_dir := DataDir,
 
 
 make_wal_conf(#{data_dir := DataDir,
-                name := _System,
+                name := System,
                 names := #{wal := WalName,
                            segment_writer := SegWriterName} = Names} = Cfg) ->
     WalDir = case Cfg of
@@ -78,6 +78,7 @@ make_wal_conf(#{data_dir := DataDir,
                                ?MIN_BIN_VHEAP_SIZE),
     MinHeapSize = maps:get(wal_min_heap_size, Cfg, ?MIN_HEAP_SIZE),
     #{name => WalName,
+      system => System,
       names => Names,
       dir => WalDir,
       segment_writer => SegWriterName,

src/ra_log_wal.erl

@@ -132,6 +132,7 @@
 
 -type state() :: #state{}.
 -type wal_conf() :: #{name := atom(), %% the name to register the wal as
+                      system := atom(),
                       names := ra_system:names(),
                       dir := file:filename_all(),
                       max_size_bytes => non_neg_integer(),
@@ -254,7 +255,7 @@ start_link(#{name := Name} = Config)
 -spec init(wal_conf()) ->
     {ok, state()} |
     {stop, wal_checksum_validation_failure} | {stop, term()}.
-init(#{dir := Dir} = Conf0) ->
+init(#{dir := Dir, system := System} = Conf0) ->
     #{max_size_bytes := MaxWalSize,
       max_entries := MaxEntries,
       recovery_chunk_size := RecoveryChunkSize,
@@ -278,7 +279,9 @@ init(#{dir := Dir} = Conf0) ->
     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, ?COUNTER_FIELDS),
+    CRef = ra_counters:new(WalName,
+                           ?COUNTER_FIELDS,
+                           #{ra_system => System, module => ?MODULE}),
     Conf = #conf{dir = Dir,
                  segment_writer = SegWriter,
                  compute_checksums = ComputeChecksums,