Make sure stop and delete servers handle both name formats.

Also normalise Name to binary inside writer and replica.

Author: kjnilsson

src/osiris_log.erl

@@ -2653,23 +2653,25 @@ read_header0(#?MODULE{cfg = #cfg{directory = Dir,
     end.
 
 trigger_retention_eval(#?MODULE{cfg =
-                                    #cfg{directory = Dir,
+                                    #cfg{name = Name,
+                                         directory = Dir,
                                          retention = RetentionSpec,
                                          counter = Cnt,
                                          shared = Shared}} = State) ->
 
     %% updates first offset and first timestamp
     %% after retention has been evaluated
-    EvalFun = fun ({{FstOff, _}, FstTs, Seg}) when is_integer(FstOff),
-                                                   is_integer(FstTs) ->
+    EvalFun = fun ({{FstOff, _}, FstTs, NumSegLeft})
+                    when is_integer(FstOff),
+                         is_integer(FstTs) ->
                       osiris_log_shared:set_first_chunk_id(Shared, FstOff),
                       counters:put(Cnt, ?C_FIRST_OFFSET, FstOff),
                       counters:put(Cnt, ?C_FIRST_TIMESTAMP, FstTs),
-                      counters:put(Cnt, ?C_SEGMENTS, Seg);
+                      counters:put(Cnt, ?C_SEGMENTS, NumSegLeft);
                   (_) ->
                       ok
               end,
-    ok = osiris_retention:eval(Dir, RetentionSpec, EvalFun),
+    ok = osiris_retention:eval(Name, Dir, RetentionSpec, EvalFun),
     State.
 
 next_location(undefined) ->

src/osiris_replica.erl

@@ -158,9 +158,11 @@ await(Server) ->
 init(Config) ->
     {ok, undefined, {continue, Config}}.
 
-handle_continue(#{name := Name,
+handle_continue(#{name := Name0,
                   leader_pid := LeaderPid,
-                  reference := ExtRef} = Config, undefined) ->
+                  reference := ExtRef} = Config, undefined)
+  when ?IS_STRING(Name0) ->
+    Name = osiris_util:normalise_name(Name0),
     process_flag(trap_exit, true),
     process_flag(message_queue_data, off_heap),
     Node = node(LeaderPid),

src/osiris_retention.erl

@@ -11,7 +11,7 @@
 
 %% API functions
 -export([start_link/0,
-         eval/3]).
+         eval/4]).
 %% gen_server callbacks
 -export([init/1,
          handle_call/3,
@@ -20,6 +20,8 @@
          terminate/2,
          code_change/3]).
 
+-define(DEFAULT_SHEDULED_EVAL_TIME, 1000 * 60 * 60). %% 1HR
+
 -record(state, {scheduled = #{} :: #{osiris:name() => timer:tref()}}).
 
 %%%===================================================================
@@ -30,13 +32,13 @@
 start_link() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
--spec eval(file:name_all(), [osiris:retention_spec()],
+-spec eval(osiris:name(), file:name_all(), [osiris:retention_spec()],
            fun((osiris_log:range()) -> ok)) ->
-              ok.
-eval(_Dir, [], _Fun) ->
+    ok.
+eval(_Name, _Dir, [], _Fun) ->
     ok;
-eval(Dir, Specs, Fun) ->
-    gen_server:cast(?MODULE, {eval, Dir, Specs, Fun}).
+eval(Name, Dir, Specs, Fun) ->
+    gen_server:cast(?MODULE, {eval, Name, Dir, Specs, Fun}).
 
 %%%===================================================================
 %%% gen_server callbacks
@@ -63,10 +65,10 @@ handle_call(_Request, _From, State) ->
 %% @spec handle_cast(Msg, State) -> {noreply, State} |
 %%                                  {noreply, State, Timeout} |
 %%                                  {stop, Reason, State}
-handle_cast({eval, Dir, Specs, Fun}, State) ->
+handle_cast({eval, Name, Dir, Specs, Fun}, State) ->
     Result = osiris_log:evaluate_retention(Dir, Specs),
     _ = Fun(Result),
-    {noreply, State}.
+    {noreply, schedule(osiris_util:normalise_name(Name), Specs, Result, State)}.
 
 %% @spec handle_info(Info, State) -> {noreply, State} |
 %%                                   {noreply, State, Timeout} |
@@ -85,3 +87,6 @@ code_change(_OldVsn, State, _Extra) ->
 %%%===================================================================
 %%% Internal functions
 %%%===================================================================
+schedule(_Name, _Specs, _Result, State) ->
+    State.
+

src/osiris_server_sup.erl

@@ -21,15 +21,23 @@ init([]) ->
     Procs = [],
     {ok, {{one_for_one, 1, 5}, Procs}}.
 
-stop_child(Node, CName) ->
+stop_child(Node, Name) ->
     try
-        case supervisor:terminate_child({?MODULE, Node}, CName) of
+        %% as replicas are temporary we don't have to explicitly
+        %% delete them
+        case supervisor:terminate_child({?MODULE, Node}, Name) of
             ok ->
-                %% as replicas are temporary we don't have to explicitly
-                %% delete them
                 ok;
             {error, not_found} ->
-                ok;
+                OthName = flip_name(Name),
+                case supervisor:terminate_child({?MODULE, Node}, OthName) of
+                    ok ->
+                        ok;
+                    {error, not_found} ->
+                        ok;
+                    Err ->
+                        Err
+                end;
             Err ->
                 Err
         end
@@ -39,17 +47,29 @@ stop_child(Node, CName) ->
             ok
     end.
 
-delete_child(Node, #{name := CName} = Config) ->
+delete_child(Node, #{name := Name} = Config) ->
     try
-        case supervisor:get_childspec({?MODULE, Node}, CName) of
+        case supervisor:get_childspec({?MODULE, Node}, Name) of
             {ok, _} ->
-                _ = stop_child(Node, CName),
+                _ = stop_child(Node, Name),
                 rpc:call(Node, osiris_log, delete_directory, [Config]);
             {error, not_found} ->
-                ok
+                OthName = flip_name(Name),
+                case supervisor:get_childspec({?MODULE, Node}, OthName) of
+                    {ok, _} ->
+                        _ = stop_child(Node, OthName),
+                        rpc:call(Node, osiris_log, delete_directory, [Config]);
+                    {error, not_found} ->
+                        ok
+                end
         end
     catch
         _:{noproc, _} ->
             %% Whole supervisor or app is already down - i.e. stop_app
             ok
     end.
+
+flip_name(N) when is_binary(N) ->
+    binary_to_list(N);
+flip_name(N) when is_list(N) ->
+    list_to_binary(N).

src/osiris_util.erl

@@ -17,7 +17,8 @@
          get_replication_configuration_from_tls_dist/0,
          get_replication_configuration_from_tls_dist/1,
          get_replication_configuration_from_tls_dist/2,
-         partition_parallel/3
+         partition_parallel/3,
+         normalise_name/1
         ]).
 
 %% For testing
@@ -43,8 +44,8 @@ validate_base64uri(Str) when ?IS_STRING(Str) ->
               string:is_empty(Str) == false
           end.
 
--spec to_base64uri(string()) -> string().
-to_base64uri(Str) when is_list(Str) ->
+-spec to_base64uri(string() | binary()) -> string().
+to_base64uri(Str) when ?IS_STRING(Str) ->
     lists:foldr(fun(G, Acc) ->
                    case lists:member(G, ?BASE64_URI_CHARS) of
                        true -> [G | Acc];
@@ -261,3 +262,8 @@ collect([{{Pid, MRef}, E} | Next], {Left, Right}, Timeout) ->
     after Timeout ->
               exit(partition_parallel_timeout)
     end.
+
+normalise_name(Name) when is_binary(Name) ->
+    Name;
+normalise_name(Name) when is_list(Name) ->
+    list_to_binary(Name).

src/osiris_writer.erl

@@ -72,13 +72,14 @@
 
 start(Config = #{name := Name, leader_node := Leader}) ->
     supervisor:start_child({?SUP, Leader},
-                           #{id => Name,
+                           #{id => osiris_util:normalise_name(Name),
                              start => {?MODULE, start_link, [Config]},
                              restart => temporary,
                              shutdown => 5000,
                              type => worker}).
 
 stop(#{name := Name, leader_node := Node}) ->
+    %% stop child handles name normalisation
     ?SUP:stop_child(Node, Name).
 
 delete(#{leader_node := Node} = Config) ->
@@ -150,12 +151,13 @@ query_replication_state(Pid) when is_pid(Pid) ->
 init(Config) ->
     {ok, undefined, {continue, Config}}.
 
-handle_continue(#{name := Name,
+handle_continue(#{name := Name0,
                   epoch := Epoch,
                   reference := ExtRef,
                   replica_nodes := Replicas} =
                 Config, undefined)
-    when ?IS_STRING(Name) ->
+  when ?IS_STRING(Name0) ->
+    Name = osiris_util:normalise_name(Name0),
     Dir = osiris_log:directory(Config),
     process_flag(trap_exit, true),
     process_flag(message_queue_data, off_heap),

test/osiris_SUITE.erl

@@ -106,7 +106,6 @@ init_per_testcase(TestCase, Config) ->
     ok = extra_init(TestCase),
     {ok, Apps} = application:ensure_all_started(osiris),
     ok = logger:set_primary_config(level, all),
-    % file:make_dir(Dir),
     [{data_dir, Dir},
      {test_case, TestCase},
      {cluster_name, atom_to_binary(TestCase, utf8)},
@@ -769,16 +768,24 @@ cluster_restart_new_leader(Config) ->
     ok.
 
 cluster_delete(Config) ->
-    PrivDir = ?config(data_dir, Config),
     Name = ?config(cluster_name, Config),
+    ok = cluster_delete(Config, Name, Name),
+    ok = cluster_delete(Config, Name, binary_to_list(Name)),
+    ok = cluster_delete(Config, binary_to_list(Name), binary_to_list(Name)),
+    ok = cluster_delete(Config, binary_to_list(Name), Name),
+    ok.
+
+cluster_delete(Config, StartName, DeleteName) ->
+    PrivDir = ?config(data_dir, Config),
     PeerStates = [start_child_node(N, PrivDir) || N <- [s1, s2, s3]],
     [LeaderNode | Replicas] = [NodeName || {_Ref, NodeName} <- PeerStates],
     Conf0 =
-        #{name => Name,
+        #{name => StartName,
           epoch => 1,
           leader_node => LeaderNode,
           replica_nodes => Replicas},
-    {ok, #{leader_pid := Leader} = Conf} = osiris:start_cluster(Conf0),
+    {ok, #{leader_pid := Leader,
+           replica_pids := ReplicaPids} = Conf} = osiris:start_cluster(Conf0),
     ok = osiris:write(Leader, undefined, 42, <<"before-restart">>),
     receive
         {osiris_written, _, _WriterId, [42]} ->
@@ -788,7 +795,11 @@ cluster_delete(Config) ->
         exit(osiris_written_timeout)
     end,
 
-    osiris:delete_cluster(Conf),
+    osiris:delete_cluster(Conf#{name => DeleteName}),
+    %% validate pids are gone
+    [begin
+         ?assertEqual(false, erpc:call(node(Pid), erlang,is_process_alive, [Pid]))
+     end || Pid <- [Leader | ReplicaPids]],
     [stop_peer(Ref) || {Ref, _} <- PeerStates],
     ok.
 
@@ -1732,7 +1743,7 @@ start_child_node(NodeNamePrefix, PrivDir) ->
 start_child_node(NodeName, PrivDir, ExtraAppConfig0) ->
     _ = file:make_dir(PrivDir),
     Dir = filename:join(PrivDir, NodeName),
-    ok = file:make_dir(Dir),
+    _ = file:make_dir(Dir),
 
     %% make sure the data dir computed above is passed on to the peers
     ExtraAppConfig1 = proplists:delete(data_dir, ExtraAppConfig0),