compaction counters

Author: kjnilsson

src/ra.erl

@@ -73,7 +73,8 @@
          member_overview/1,
          member_overview/2,
          key_metrics/1,
-         key_metrics/2
+         key_metrics/2,
+         trigger_compaction/1
         ]).
 
 -define(START_TIMEOUT, ?DEFAULT_TIMEOUT).
@@ -1192,6 +1193,13 @@ key_metrics({Name, N} = ServerId, _Timeout) when N == node() ->
 key_metrics({_, N} = ServerId, Timeout) ->
     erpc:call(N, ?MODULE, ?FUNCTION_NAME, [ServerId], Timeout).
 
+%% @doc Potentially triggers a major compaction for the provided member
+%% @param ServerId the Ra server to send the request to
+%% @end
+-spec trigger_compaction(ra_server_id()) -> ok.
+trigger_compaction(ServerRef) ->
+    gen_statem:cast(ServerRef, {ra_log_event, major_compaction}).
+
 %% internal
 
 -spec usr(UserCommand, ReplyMode) -> Command when

src/ra.hrl

@@ -278,8 +278,17 @@
           "Number of checkpoint bytes written"},
          {checkpoints_promoted, ?C_RA_LOG_CHECKPOINTS_PROMOTED, counter,
           "Number of checkpoints promoted to snapshots"},
+         {major_compactions, ?C_RA_LOG_COMPACTIONS_MAJOR_COUNT, counter,
+          "Number of requested major compactions"},
+         {major_compaction_segments_written,
+          ?C_RA_LOG_COMPACTIONS_SEGMENTS_WRITTEN, counter,
+          "Number of segments written during major compactions"},
+         {major_compaction_segments_compacted,
+          ?C_RA_LOG_COMPACTIONS_SEGMENTS_COMPACTED, counter,
+          "Number of segments compacted during major compactions"},
          {reserved_1, ?C_RA_LOG_RESERVED, counter, "Reserved counter"}
          ]).
+
 -define(C_RA_LOG_WRITE_OPS, 1).
 -define(C_RA_LOG_WRITE_RESENDS, 2).
 -define(C_RA_LOG_READ_OPS, 3).
@@ -295,7 +304,10 @@
 -define(C_RA_LOG_CHECKPOINTS_WRITTEN, 13).
 -define(C_RA_LOG_CHECKPOINT_BYTES_WRITTEN, 14).
 -define(C_RA_LOG_CHECKPOINTS_PROMOTED, 15).
--define(C_RA_LOG_RESERVED, 16).
+-define(C_RA_LOG_COMPACTIONS_MAJOR_COUNT, 16).
+-define(C_RA_LOG_COMPACTIONS_SEGMENTS_WRITTEN, 17).
+-define(C_RA_LOG_COMPACTIONS_SEGMENTS_COMPACTED, 18).
+-define(C_RA_LOG_RESERVED, 19).
 
 -define(C_RA_SRV_AER_RECEIVED_FOLLOWER, ?C_RA_LOG_RESERVED + 1).
 -define(C_RA_SRV_AER_REPLIES_SUCCESS, ?C_RA_LOG_RESERVED + 2).

src/ra_lib.erl

@@ -337,9 +337,9 @@ retry(Func, Attempt, Sleep) ->
             ok;
         true ->
             ok;
-        _ ->
+        _Err ->
             timer:sleep(Sleep),
-            retry(Func, Attempt - 1)
+            retry(Func, Attempt - 1, Sleep)
     end.
 
 -spec write_file(file:name_all(), iodata()) ->

src/ra_log.erl

@@ -71,6 +71,8 @@
                       {snapshot_written, ra_idxterm(),
                        LiveIndexes :: ra_seq:state(),
                        ra_snapshot:kind()} |
+                      {compaction_result, term()} |
+                      major_compaction |
                       {down, pid(), term()}.
 
 -type event() :: {ra_log_event, event_body()}.
@@ -818,6 +820,17 @@ handle_event({compaction_result, Result},
              #?MODULE{reader = Reader0} = State) ->
     {Reader, Effs} = ra_log_segments:handle_compaction_result(Result, Reader0),
     {State#?MODULE{reader = Reader}, Effs};
+handle_event(major_compaction, #?MODULE{reader = Reader0,
+                                        live_indexes = LiveIndexes,
+                                        snapshot_state = SS} = State) ->
+    case ra_snapshot:current(SS) of
+        {SnapIdx, _} ->
+            Effs = ra_log_segments:schedule_compaction(major,SnapIdx,
+                                                       LiveIndexes, Reader0),
+            {State, Effs};
+        _ ->
+            {State, []}
+    end;
 handle_event({snapshot_written, {SnapIdx, _} = Snap, LiveIndexes, SnapKind},
              #?MODULE{cfg = #cfg{uid = UId,
                                  names = Names} = Cfg,

src/ra_log_segments.erl

@@ -150,7 +150,7 @@ update_segments(NewSegmentRefs, #?STATE{open_segments = Open0,
     [ra_server:effect()].
 schedule_compaction(Type, SnapIdx, LiveIndexes,
                     #?MODULE{cfg = #cfg{log_id = LogId,
-                                        directory = Dir}} = State) ->
+                                        directory = Dir} = Cfg} = State) ->
     case compactable_segrefs(SnapIdx, State) of
         [] ->
             [];
@@ -170,6 +170,7 @@ schedule_compaction(Type, SnapIdx, LiveIndexes,
         SegRefs ->
             Self = self(),
             Fun = fun () ->
+                          ok = incr_counter(Cfg, ?C_RA_LOG_COMPACTIONS_MAJOR_COUNT, 1),
                           MajConf = #{dir => Dir},
                           Result = major_compaction(MajConf, SegRefs,
                                                     LiveIndexes),
@@ -200,7 +201,7 @@ handle_compaction_result(#compaction_result{unreferenced = [],
 handle_compaction_result(#compaction_result{unreferenced = Unreferenced,
                                             linked = Linked,
                                             compacted = Compacted},
-                         #?STATE{cfg = #cfg{directory = Dir},
+                         #?STATE{cfg = #cfg{directory = Dir} = Cfg,
                                  open_segments = Open0,
                                  segment_refs = SegRefs0} = State) ->
     SegRefs1 = maps:from_list(ra_lol:to_list(SegRefs0)),
@@ -214,6 +215,10 @@ handle_compaction_result(#compaction_result{unreferenced = Unreferenced,
                    || F <- Unreferenced],
                   ok
           end,
+    ok = incr_counter(Cfg, ?C_RA_LOG_COMPACTIONS_SEGMENTS_WRITTEN,
+                      length(Compacted)),
+    ok = incr_counter(Cfg, ?C_RA_LOG_COMPACTIONS_SEGMENTS_COMPACTED,
+                      length(Linked) + length(Compacted)),
     {State#?MODULE{segment_refs = ra_lol:from_list(fun seg_ref_gt/2,
                                                    SegmentRefs),
                    open_segments = Open},

test/ra_kv_SUITE.erl

@@ -3,14 +3,11 @@
 -compile(nowarn_export_all).
 -compile(export_all).
 
--export([
-         ]).
-
--include_lib("src/ra.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 
 -define(SYS, default).
+
 %%%===================================================================
 %%% Common Test callbacks
 %%%===================================================================
@@ -118,7 +115,6 @@ basics(_Config) ->
     {ok, {Reads3, _}} = ra_server_proc:read_entries(KvId2, [LastIdx | Live],
                                                     undefined, 1000),
     ct:pal("ReadRes3 ~p", [Reads3]),
-    % ct:pal("overview3 ~p", [ra:member_overview(KvId2)]),
     ?assertEqual(3, map_size(Reads3)),
 
     %% TODO: test recovery of kv
@@ -128,4 +124,16 @@ basics(_Config) ->
                                                     undefined, 1000),
 
     ?assertEqual(3, map_size(Reads4)),
+    ra:trigger_compaction(KvId),
+    %% wait for compaction by querying counters
+    ok = ra_lib:retry(
+           fun () ->
+                   #{major_compactions := Maj} =
+                       ra_counters:counters(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)]),
     ok.

test/ra_log_2_SUITE.erl

@@ -256,7 +256,6 @@ read_one(Config) ->
     ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS),
     Log0 = ra_log_init(Config, #{counter => ra_counters:fetch(?FUNCTION_NAME)}),
     Log1 = append_n(1, 2, 1, Log0),
-    % Log1 = ra_log:append({1, 1, <<1:64/integer>>}, Log0),
     % ensure the written event is delivered
     Log2 = deliver_all_log_events(Log1, 200),
     {[_], Log} = ra_log_take(1, 1, Log2),