Periodically re-evaluate max_age retention

This implements periodic re-evaluation of max_age retention specificions
so that streams that have little activity but more than one segments will
evantually be truncated.

Author: kjnilsson

src/osiris_retention.erl

@@ -65,10 +65,11 @@ handle_call(_Request, _From, State) ->
 %% @spec handle_cast(Msg, State) -> {noreply, State} |
 %%                                  {noreply, State, Timeout} |
 %%                                  {stop, Reason, State}
-handle_cast({eval, Name, Dir, Specs, Fun}, State) ->
+handle_cast({eval, _Name, Dir, Specs, Fun} = Eval, State) ->
+    ct:pal("EVAL ~p", [Eval]),
     Result = osiris_log:evaluate_retention(Dir, Specs),
     _ = Fun(Result),
-    {noreply, schedule(osiris_util:normalise_name(Name), Specs, Result, State)}.
+    {noreply, schedule(Eval, Result, State)}.
 
 %% @spec handle_info(Info, State) -> {noreply, State} |
 %%                                   {noreply, State, Timeout} |
@@ -87,6 +88,25 @@ code_change(_OldVsn, State, _Extra) ->
 %%%===================================================================
 %%% Internal functions
 %%%===================================================================
-schedule(_Name, _Specs, _Result, State) ->
-    State.
+schedule({eval, Name, _Dir, Specs, _Fun} = Eval, _Result,
+         #state{scheduled = Scheduled0} = State) ->
+    %% we need to check the scheduled map even if the current specs do not
+    %% include max_age as the retention config could have changed
+    Scheduled = case maps:take(Name, Scheduled0) of
+                    {OldRef, Scheduled1} ->
+                        _ = erlang:cancel_timer(OldRef),
+                        Scheduled1;
+                    error ->
+                        Scheduled0
+                end,
+    case lists:any(fun ({T, _}) -> T == max_age end, Specs) of
+        true ->
+            %% schedule a new eval
+            EvalInterval = application:get_env(osiris, retention_eval_interval,
+                                               ?DEFAULT_SHEDULED_EVAL_TIME),
+            Ref = erlang:send_after(EvalInterval, self(), {'$gen_cast', Eval}),
+            State#state{scheduled = Scheduled#{Name => Ref}};
+        false ->
+            State#state{scheduled = Scheduled}
+    end.
 

test/osiris_SUITE.erl

@@ -53,6 +53,8 @@ all_tests() ->
      replica_unknown_command,
      diverged_replica,
      retention,
+     retention_max_age_eventually,
+     retention_max_age_update_retention,
      retention_add_replica_after,
      retention_overtakes_offset_reader,
      update_retention,
@@ -1033,6 +1035,7 @@ retention(Config) ->
     Num = 150000,
     Name = ?config(cluster_name, Config),
     SegSize = 50000 * 1000,
+    %% silly low setting
     Conf0 =
         #{name => Name,
           epoch => 1,
@@ -1051,6 +1054,58 @@ retention(Config) ->
     osiris:stop_cluster(Conf1),
     ok.
 
+retention_max_age_eventually(Config) ->
+    DataDir = ?config(data_dir, Config),
+    Num = 150000,
+    Name = ?config(cluster_name, Config),
+    SegSize = 50000 * 1000,
+    application:set_env(osiris, retention_eval_interval, 5000),
+    Conf0 =
+        #{name => Name,
+          epoch => 1,
+          leader_node => node(),
+          retention => [{max_age, 5000}],
+          max_segment_size_bytes => SegSize,
+          replica_nodes => []},
+    {ok, #{leader_pid := Leader, replica_pids := []} = Conf1} =
+        osiris:start_cluster(Conf0),
+    timer:sleep(100),
+    write_n(Leader, Num, 0, 1000 * 8, #{}),
+    Wc = filename:join([DataDir, ?FUNCTION_NAME, "*.segment"]),
+    %% one file only
+    await_condition(fun () ->
+                            length(filelib:wildcard(Wc)) == 1
+                    end, 1000, 20),
+    osiris:stop_cluster(Conf1),
+    ok.
+
+retention_max_age_update_retention(Config) ->
+    %% ensure a retention update cancels the current scheduled
+    %% retention evaluation
+    DataDir = ?config(data_dir, Config),
+    Num = 150000,
+    Name = ?config(cluster_name, Config),
+    SegSize = 50000 * 1000,
+    application:set_env(osiris, retention_eval_interval, 5000),
+    Conf0 =
+        #{name => Name,
+          epoch => 1,
+          leader_node => node(),
+          retention => [{max_age, 5000}],
+          max_segment_size_bytes => SegSize,
+          replica_nodes => []},
+    {ok, #{leader_pid := Leader, replica_pids := []} = Conf1} =
+        osiris:start_cluster(Conf0),
+    timer:sleep(100),
+    write_n(Leader, Num, 0, 1000 * 8, #{}),
+    Wc = filename:join([DataDir, ?FUNCTION_NAME, "*.segment"]),
+    ok = osiris:update_retention(Leader, [{max_bytes, SegSize * 10}]),
+    timer:sleep(10000),
+    %% one file only
+    ?assertNot(length(filelib:wildcard(Wc)) == 1),
+    osiris:stop_cluster(Conf1),
+    ok.
+
 retention_add_replica_after(Config) ->
     DataDir = ?config(data_dir, Config),
     Num = 150000,
@@ -1960,4 +2015,19 @@ simple(Bin) ->
     S = byte_size(Bin),
     <<0:1, S:31, Bin/binary>>.
 
+await_condition(_CondFun, _Sleep, 0) ->
+    exit(await_condition_attempts_exceeded);
+await_condition(CondFun, Sleep, Attempt) ->
+    timer:sleep(Sleep),
+    try CondFun() of
+        true ->
+            ok;
+        false ->
+            await_condition(CondFun, Sleep, Attempt-1)
+    catch
+        _:Err ->
+            ct:pal("~s err ~p", [?FUNCTION_NAME, Err]),
+            await_condition(CondFun, Sleep, Attempt-1)
+    end.
+