Packages
ra
3.0.0
3.1.9
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.2
3.0.1
3.0.0
3.0.0-beta.1
2.17.3
2.17.2
2.17.1
2.17.0
2.16.13
2.16.12
2.16.11
2.16.10
2.16.9
2.16.8
2.16.7
2.16.6
2.16.5
2.16.4
2.16.3
2.16.2
2.16.1
2.16.0
2.16.0-pre.12
2.16.0-pre.11
2.16.0-pre.10
2.16.0-pre.9
2.16.0-pre.8
2.16.0-pre.7
2.16.0-pre.6
2.16.0-pre.5
2.16.0-pre.4
2.16.0-pre.3
2.16.0-pre.2
2.16.0-pre.1
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.13.0-pre.1
2.12.0
2.11.0
2.11.0-pre.1
2.10.2-pre.2
2.10.2-pre.1
2.10.1
2.10.0
2.10.0-pre.3
2.10.0-pre.2
2.10.0-pre.1
2.9.10-pre.1
2.9.1
2.9.1-pre.2
2.9.1-pre.1
2.9.0
2.8.0
retired
2.7.3
2.7.2
2.7.1
2.7.0
2.7.0-pre.3
2.7.0-pre.2
2.7.0-pre.1
2.6.3
2.6.2
2.6.1
2.6.0-pre.1
2.5.1
2.5.1-pre.1
2.5.0
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
retired
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.13
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.9.6
0.9.5
0.9.4
0.9.2
0.3.3
retired
0.3.2
retired
0.3.1
retired
Raft library
Current section
Files
Jump to
Current section
Files
src/ra_kv_harness.erl
-module(ra_kv_harness).
-export([
run/1,
run/2,
read_all_keys/0,
teardown_cluster/1,
timestamp/0,
log/2
]).
-include_lib("eunit/include/eunit.hrl").
-define(SYS, default).
-define(CLUSTER_NAME, kv).
-define(TIMEOUT, 30000).
-define(MAX_KEY, 10000). % Limited key space for more conflicts
-define(MIN_VALUE_SIZE, 1).
-define(MAX_VALUE_SIZE, 10_000_000). % 10MB
-define(MAX_NODES, 7). % Maximum number of nodes in the cluster
-type state() :: #{options => map(),
members := #{ra:server_id() => peer:server_ref()},
reference_map := #{binary() => term()},
operations_count := non_neg_integer(),
successful_ops := non_neg_integer(),
failed_ops := non_neg_integer(),
next_node_id := pos_integer(),
remaining_ops := non_neg_integer(),
consistency_failed := boolean(),
partition_state => #{partitioned_node => ra:server_id(),
heal_time => non_neg_integer(),
other_nodes => [node()]}}.
-spec timestamp() -> string().
timestamp() ->
{MegaSecs, Secs, MicroSecs} = os:timestamp(),
{{Year, Month, Day}, {Hour, Min, Sec}} = calendar:now_to_local_time({MegaSecs, Secs, MicroSecs}),
Millisecs = MicroSecs div 1000,
io_lib:format("[~4..0w-~2..0w-~2..0w ~2..0w:~2..0w:~2..0w.~3..0w]",
[Year, Month, Day, Hour, Min, Sec, Millisecs]).
milliseconds() ->
erlang:system_time(millisecond).
-spec log(string(), list()) -> ok.
log(Format, Args) ->
Message = io_lib:format(Format, Args),
io:format("~s", [Message]),
file:write_file("ra_kv_harness.log", Message, [append]).
-spec new_state() -> state().
new_state() ->
#{members => #{},
reference_map => #{},
operations_count => 0,
successful_ops => 0,
failed_ops => 0,
next_node_id => 1,
remaining_ops => 0,
consistency_failed => false}.
-type operation() :: {put, Key :: binary(), Value :: term()} |
{get, Key :: binary()} |
{snapshot} |
{major_compaction} |
{update_almost_all_keys} |
{kill_wal} |
{kill_member} |
{add_member} |
{remove_member} |
{network_partition}.
-spec run(NumOperations :: pos_integer()) ->
{ok, #{successful := non_neg_integer(),
failed := non_neg_integer(),
consistency_checks := non_neg_integer()}} |
{error, term()}.
run(NumOperations) ->
run(NumOperations, #{}).
read_all_keys() ->
[begin
_ = ra_kv:get({?CLUSTER_NAME,
node()},
<<"key_",(integer_to_binary(N))/binary>>, 1000),
ok
end
|| N <- lists:seq(1, ?MAX_KEY)],
ok.
read_all_keys_loop(Members0) when is_list(Members0) ->
receive
stop ->
log("~s Read all keys loop stopped~n", [timestamp()]),
ok
after 0 ->
%% resolve current members
{ok, Members, _} = ra:members(Members0),
Member = hd(Members),
NodeName = element(2, Member),
log("~s Begin reading all keys on member ~0p~n", [timestamp(), Member]),
T1 = erlang:monotonic_time(),
ok = erpc:call(NodeName, ra_kv_harness, read_all_keys, []),
T2 = erlang:monotonic_time(),
Diff = erlang:convert_time_unit(T2 - T1, native, millisecond),
log("~s Read all keys on member ~0p in ~bms~n", [timestamp(), Member, Diff]),
read_all_keys_loop(Members)
end.
-spec run(NumOperations :: pos_integer(),
Options :: map()) ->
{ok, #{successful := non_neg_integer(),
failed := non_neg_integer(),
consistency_checks := non_neg_integer()}} |
{error, term()}.
run(NumOperations, Options) when NumOperations > 0 ->
% Start with a random number of nodes between 1 and 7
NumNodes = rand:uniform(7),
ok = logger:set_primary_config(level, warning),
ok = application:set_env(sasl, sasl_error_logger, false),
ok = application:stop(sasl),
log("~s Starting cluster with ~p nodes~n", [timestamp(), NumNodes]),
case setup_cluster(NumNodes, Options) of
{ok, Members, PeerNodes} ->
MembersMap = maps:from_list(lists:zip(Members, PeerNodes)),
InitialState = (new_state())#{members => MembersMap,
next_node_id => NumNodes + 1,
remaining_ops => NumOperations,
options => Options},
%% keep reading all keys while the other operations are running
ReaderPid = spawn(fun() -> read_all_keys_loop(maps:keys(MembersMap)) end),
try
State = execute_operation(InitialState, {put, <<"never_updated">>, <<"never_updated">>}),
FinalState = run_operations(State, ?CLUSTER_NAME),
ReaderPid ! stop,
case maps:get(consistency_failed, FinalState, false) of
true ->
log("~s EMERGENCY STOP: Leaving cluster running for investigation~n", [timestamp()]),
{error, {consistency_failure, FinalState}};
false ->
ConsistencyChecks = validate_final_consistency(FinalState),
teardown_cluster(FinalState),
{ok, #{successful => maps:get(successful_ops, FinalState),
failed => maps:get(failed_ops, FinalState),
consistency_checks => ConsistencyChecks}}
end
catch
Class:Reason:Stack ->
ReaderPid ! stop,
teardown_cluster(InitialState),
{error, {Class, Reason, Stack}}
end;
{error, Reason} ->
{error, Reason}
end.
setup_cluster(NumNodes, Opts) when NumNodes > 0 ->
% Start peer nodes
case start_peer_nodes(NumNodes, Opts) of
{ok, PeerNodes, NodeNames} ->
Members = [{?CLUSTER_NAME, NodeName} || NodeName <- NodeNames],
case ra_kv:start_cluster(?SYS, ?CLUSTER_NAME, #{members => Members}) of
{ok, StartedMembers, _} ->
log("~s Started cluster with ~p members~n", [timestamp(), length(StartedMembers)]),
{ok, StartedMembers, PeerNodes};
{error, Reason} ->
[peer:stop(PeerRef) || PeerRef <- PeerNodes],
{error, Reason}
end;
{error, Reason} ->
{error, Reason}
end.
start_peer_nodes(NumNodes, Opts) ->
start_peer_nodes(NumNodes, [], [], Opts).
start_peer_nodes(0, PeerRefs, NodeNames, _Opts) ->
{ok, lists:reverse(PeerRefs), lists:reverse(NodeNames)};
start_peer_nodes(N, PeerRefs, NodeNames, Opts) when N > 0 ->
case start_single_peer_node(N, Opts) of
{ok, PeerRef, NodeName} ->
start_peer_nodes(N - 1, [PeerRef | PeerRefs], [NodeName | NodeNames], Opts);
{error, Reason} ->
% Clean up any already started peers
[peer:stop(PeerRef) || PeerRef <- PeerRefs],
{error, Reason}
end.
start_single_peer_node(NodeId, Opts) ->
NodeName = list_to_atom("ra_test_" ++ integer_to_list(NodeId) ++ "@" ++
inet_db:gethostname()),
log("~s Starting node ~p nodes~n", [timestamp(), NodeName]),
% Get all code paths from current node
CodePaths = code:get_path(),
PaArgs = lists:flatmap(fun(Path) -> ["-pa", Path] end, CodePaths),
BufferSize = ["+zdbbl", "102400"],
% Check if inet_tcp_proxy_dist is available
% If yes, add -proto_dist argument at the front with its path, like erlang_node_helpers does
ProtoDistArgs = case code:where_is_file("inet_tcp_proxy_dist.beam") of
non_existing ->
% inet_tcp_proxy not available, don't use it
log("~s WARNING: inet_tcp_proxy_dist not found, starting peer without it~n", [timestamp()]),
[];
BeamPath ->
DistModPath = filename:dirname(BeamPath),
["-pa", DistModPath, "-proto_dist", "inet_tcp_proxy",
"-kernel", "prevent_overlapping_partitions", "false"]
end,
case peer:start_link(#{name => NodeName,
args => ProtoDistArgs ++ PaArgs ++ BufferSize,
connection => standard_io}) of
{ok, PeerRef, NodeName} ->
BaseDir = maps:get(dir, Opts, ""),
erpc:call(NodeName, logger, set_primary_config, [level, warning]),
erpc:call(NodeName, application, set_env, [sasl, sasl_error_logger, false]),
erpc:call(NodeName, application, stop, [sasl]),
Dir = filename:join(BaseDir, NodeName),
{ok, _} = erpc:call(NodeName, ra, start_in, [Dir]),
% Set logger level to reduce verbosity on peer node
% Start ra application on the new peer node
{ok, PeerRef, NodeName};
{error, Reason} ->
{error, Reason}
end.
start_new_peer_node(NodeId, Opts) ->
start_single_peer_node(NodeId, Opts).
-spec teardown_cluster(state()) -> ok.
teardown_cluster(#{members := Members}) ->
% Stop Ra servers on each node and stop peer nodes
maps:foreach(fun(Member, PeerRef) ->
NodeName = element(2, Member),
log("~s Stopping member ~p~n", [timestamp(), Member]),
catch erpc:call(NodeName, ra, stop_server, [?SYS, Member]),
catch peer:stop(PeerRef),
ok
end, Members),
ok.
run_operations(State, _ClusterName) ->
RemainingOps = maps:get(remaining_ops, State),
case RemainingOps =< 0 of
true ->
State;
false ->
case RemainingOps rem 1000 of
0 -> log("~s ~p operations remaining~n", [timestamp(), RemainingOps]);
_ -> ok
end,
Operation = generate_operation(),
NewState = execute_operation(State, Operation),
% Update remaining operations count
UpdatedState = NewState#{remaining_ops => RemainingOps - 1},
% Validate consistency every 100 operations
ValidationState = case maps:get(operations_count, UpdatedState) rem 100 of
0 -> validate_consistency(UpdatedState);
_ -> UpdatedState
end,
run_operations(ValidationState, _ClusterName)
end.
-spec generate_operation() -> operation().
generate_operation() ->
case rand:uniform(100) of
1 -> % 1% update almost all keys
{update_almost_all_keys};
2 -> % 1% add member
{add_member};
3 -> % 1% remove member
{remove_member};
4 -> % 1% kill WAL
{kill_wal};
5 -> % 1% kill member
{kill_member};
6 -> % 1% network partition
{network_partition};
N when N =< 10 -> % 4% snapshot
{snapshot};
N when N =< 12 -> % 2% major compactions
{major_compaction};
N when N =< 80 ->
Key = generate_key(),
Value = generate_value(),
{put, Key, Value};
_ ->
Key = generate_key(),
{get, Key}
end.
key(N) when is_integer(N) ->
<<"key_", (integer_to_binary(N))/binary>>.
generate_key() ->
KeyNum = rand:uniform(?MAX_KEY), % Limited key space for more conflicts
key(KeyNum).
-spec generate_value() -> binary().
generate_value() ->
Size = rand:uniform(?MAX_VALUE_SIZE - ?MIN_VALUE_SIZE) + ?MIN_VALUE_SIZE,
rand:bytes(Size).
-spec execute_operation(state(), operation()) -> state().
execute_operation(State, {put, Key, Value}) ->
Members = maps:get(members, State),
RefMap = maps:get(reference_map, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
FailedOps = maps:get(failed_ops, State),
% Pick a random cluster member to send the operation to
MembersList = maps:keys(Members),
Member = lists:nth(rand:uniform(length(MembersList)), MembersList),
case ra_kv:put(Member, Key, Value, ?TIMEOUT) of
{ok, _Meta} ->
NewRefMap = RefMap#{Key => Value},
State#{reference_map => NewRefMap,
operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
{error, _Reason} ->
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1};
{timeout, _ServerId} ->
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1}
end;
execute_operation(State, {get, Key}) ->
Members = maps:get(members, State),
RefMap = maps:get(reference_map, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
FailedOps = maps:get(failed_ops, State),
% Pick a random cluster member to send the operation to
MembersList = maps:keys(Members),
Member = lists:nth(rand:uniform(length(MembersList)), MembersList),
NodeName = element(2, Member),
RefValue = maps:get(Key, RefMap, not_found),
log("~s ra_kv:get/3 from node ~w ~n", [timestamp(), NodeName]),
case erpc:call(NodeName, ra_kv, get, [Member, Key, ?TIMEOUT]) of
{error, not_found} when RefValue =:= not_found ->
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
{error, not_found} when RefValue =/= not_found ->
log("~s CONSISTENCY ERROR: Key ~p should exist but not found~n", [timestamp(), Key]),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1};
{ok, _Meta, Value} when RefValue =:= Value ->
log("~s ra_kv:get/3 from node ~w ok! ~n", [timestamp(), NodeName]),
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
{ok, _Meta, Value} when RefValue =/= Value ->
log("~s CONSISTENCY ERROR: Key ~p, Expected ~p, Got ~p~n",
[timestamp(), Key, RefValue, Value]),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1}
end;
execute_operation(State, {update_almost_all_keys}) ->
RefMap = maps:get(reference_map, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
Member = random_non_partitioned_member(State),
% Update a random percentage (10-90%) of keys
PercentToUpdate = rand:uniform(81) + 9, % Random number between 10 and 90
AllKeys = lists:seq(1, ?MAX_KEY),
NumKeysToUpdate = (?MAX_KEY * PercentToUpdate) div 100,
KeysToUpdate = lists:sublist([N || {_, N} <- lists:sort([{rand:uniform(), K} || K <- AllKeys])],
NumKeysToUpdate),
{T, Results} = timer:tc(fun() ->
[begin
case ra_kv:put(Member, key(N), OpCount, ?TIMEOUT) of
{ok, _} -> {ok, N};
{timeout, _} -> {timeout, N};
{error, _Reason} -> {error, N}
end
end || N <- KeysToUpdate]
end),
% Count successful and failed updates
SuccessfulKeys = [N || {ok, N} <- Results],
FailedResults = [R || R <- Results, element(1, R) =/= ok],
case FailedResults of
[] ->
log("~s Updated ~p% of the ~p keys (~p successful) in ~bms~n",
[timestamp(), PercentToUpdate, ?MAX_KEY, length(SuccessfulKeys), T div 1000]);
_ ->
log("~s Updated ~p/~p keys (~p% target) in ~bms (aborted due to ~p failures/timeouts)~n",
[timestamp(), length(SuccessfulKeys), length(KeysToUpdate), PercentToUpdate, T div 1000, length(FailedResults)])
end,
% Only update reference map for successfully updated keys
NewRefMap = maps:merge(RefMap, maps:from_list([{key(N), OpCount} || N <- SuccessfulKeys])),
State#{reference_map => NewRefMap,
operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
execute_operation(State, {snapshot}) ->
Members = maps:get(members, State),
% Pick a random cluster member to send snapshot command to
MembersList = maps:keys(Members),
Member = lists:nth(rand:uniform(length(MembersList)), MembersList),
NodeName = element(2, Member),
case erpc:call(NodeName, erlang, whereis, [?CLUSTER_NAME]) of
undefined ->
State;
_Pid ->
log("~s Rollover/snapshot on node ~p...~n", [timestamp(), NodeName]),
erpc:call(NodeName, ra_log_wal, force_roll_over, [ra_log_wal]),
erpc:call(NodeName, ra, aux_command, [Member, take_snapshot]),
State
end;
execute_operation(State, {major_compaction}) ->
Members = maps:get(members, State),
% Pick a random cluster member to send snapshot command to
MembersList = maps:keys(Members),
Member = lists:nth(rand:uniform(length(MembersList)), MembersList),
NodeName = element(2, Member),
case erpc:call(NodeName, erlang, whereis, [?CLUSTER_NAME]) of
undefined ->
State;
_Pid ->
log("~s Triggering major compaction on node ~p...~n", [timestamp(), NodeName]),
erpc:call(NodeName, ra, trigger_compaction, [Member]),
State
end;
execute_operation(#{options := Opts} = State, {add_member}) ->
Members = maps:get(members, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
FailedOps = maps:get(failed_ops, State),
NextNodeId = maps:get(next_node_id, State),
% Don't add members if we already have 7 (maximum 7 nodes)
case maps:size(Members) >= ?MAX_NODES of
true ->
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1};
false ->
log("~s Adding member on node ~p.~n",
[timestamp(), NextNodeId]),
case start_new_peer_node(NextNodeId, Opts) of
{ok, PeerRef, NodeName} ->
NewMember = {?CLUSTER_NAME, NodeName},
ExistingMember = random_non_partitioned_member(State),
case ra_kv:add_member(?SYS, NewMember, ExistingMember) of
ok ->
NewMembers = Members#{NewMember => PeerRef},
NewMembersList = maps:keys(NewMembers),
log("~s Added member ~p. Cluster now has ~p members: ~0p; partitioned node: ~0p~n",
[timestamp(), NewMember,
length(NewMembersList),
NewMembersList,
partitioned_node_name(State)]),
State#{members => NewMembers,
operations_count => OpCount + 1,
successful_ops => SuccessOps + 1,
next_node_id => NextNodeId + 1};
{timeout, _ServerId} ->
log("~s Timeout adding member ~p~n",
[timestamp(), NewMember]),
% Clean up the peer node since add failed
catch peer:stop(PeerRef),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1,
next_node_id => NextNodeId + 1};
{error, Reason} ->
log("~s Failed to add member ~p: ~p~n",
[timestamp(), NewMember, Reason]),
% Clean up the peer node since add failed
catch peer:stop(PeerRef),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1,
next_node_id => NextNodeId + 1}
end;
{error, Reason} ->
log("~s Failed to start peer node: ~p~n", [timestamp(), Reason]),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1,
next_node_id => NextNodeId + 1}
end
end;
execute_operation(State, {remove_member}) ->
Members = maps:get(members, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
FailedOps = maps:get(failed_ops, State),
% Don't remove members if we only have one left (minimum 1 node)
case maps:size(Members) =< 1 of
true ->
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1};
false ->
% Pick a random member to remove
MembersList = maps:keys(Members),
MemberToRemove = random_non_partitioned_member(State),
% Pick a different member to send the remove command to
RemainingMembers = MembersList -- [MemberToRemove],
CommandTarget = random_non_partitioned_member(RemainingMembers, State),
log("~s Removing member ~w... command target ~w~n",
[timestamp(), MemberToRemove, CommandTarget]),
case ra_kv:remove_member(?SYS, MemberToRemove, CommandTarget) of
ok ->
% Stop the peer node for the removed member
case maps:get(MemberToRemove, Members, undefined) of
undefined ->
ok;
PeerRef ->
catch peer:stop(PeerRef)
end,
NewMembers = maps:remove(MemberToRemove, Members),
NewMembersList = maps:keys(NewMembers),
log("~s Member ~w removed. Cluster now has ~p members: ~0p; paritioned node: ~p~n",
[timestamp(),
MemberToRemove,
length(NewMembersList),
NewMembersList,
partitioned_node_name(State)]),
State#{members => NewMembers,
operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
{timeout, _ServerId} ->
log("~s Timeout removing member ~p~n",
[timestamp(), MemberToRemove]),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1};
{error, Reason} ->
log("~s Failed to remove member ~p: ~p~n",
[timestamp(), MemberToRemove, Reason]),
State#{operations_count => OpCount + 1,
failed_ops => FailedOps + 1}
end
end;
execute_operation(State, {kill_wal}) ->
Members = maps:get(members, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
% Pick a node to kill WAL on
MembersList = maps:keys(Members),
Rnd = rand:uniform(?MAX_NODES),
case Rnd > length(MembersList) of
true ->
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
false ->
Member = lists:nth(Rnd, MembersList),
NodeName = element(2, Member),
log("~s Killing WAL on member ~w...~n", [timestamp(), NodeName]),
case erpc:call(NodeName, erlang, whereis, [ra_log_wal]) of
Pid when is_pid(Pid) ->
erpc:call(NodeName, erlang, exit, [Pid, kill]),
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
_ ->
State
end
end;
execute_operation(State, {kill_member}) ->
Members = maps:get(members, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
Kills = maps:get(kills, State, #{}),
% Pick a random member to kill
MembersList = maps:keys(Members),
Rnd = rand:uniform(?MAX_NODES),
case Rnd > length(MembersList) of
true ->
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1};
false ->
Member = lists:nth(rand:uniform(length(MembersList)), MembersList),
NodeName = element(2, Member),
Now = milliseconds(),
LastKill = maps:get(Member, Kills, Now - 10000),
if(Now > LastKill + 5500) ->
case erpc:call(NodeName, erlang, whereis, [?CLUSTER_NAME]) of
Pid when is_pid(Pid) ->
log("~s Killing member ~w...~n", [timestamp(), Member]),
erpc:call(NodeName, erlang, exit, [Pid, kill]),
%% give it a bit of time after a kill in case this member is chosen
%% for the next operation
timer:sleep(100),
State#{operations_count => OpCount + 1,
kills => Kills#{Member => Now},
successful_ops => SuccessOps + 1};
_ ->
State
end;
true ->
log("~s Not killing member ~w...~n", [timestamp(), Member]),
State
end
end;
execute_operation(State, {network_partition}) ->
Members = maps:get(members, State),
OpCount = maps:get(operations_count, State),
SuccessOps = maps:get(successful_ops, State),
PartitionState = maps:get(partition_state, State, #{}),
Now = milliseconds(),
case PartitionState of
#{heal_time := HealTime} when Now < HealTime ->
% There's an active partition already
State;
#{partitioned_node := PartitionedNode, heal_time := HealTime, other_nodes := OtherNodes} when Now >= HealTime ->
% Time to heal the partition
PartitionedNodeName = element(2, PartitionedNode),
log("~s Healing network partition; node ~w will rejoin the rest~n", [timestamp(), PartitionedNode]),
% Allow communication between partitioned node and other nodes
% Call allow on each node to restore bidirectional communication
% We use rpc:call with infinity timeout since nodes might be reconnecting
[begin
spawn(fun() ->
case rpc:call(PartitionedNodeName, inet_tcp_proxy_dist, allow, [OtherNode], 5000) of
ok -> ok;
Err ->
log("~s rpc:call(~p, inet_tcp_proxy_dist, allow, [~p]) failed with ~p~n",
[timestamp(),
PartitionedNodeName,
OtherNode,
Err]),
ok
end
end),
spawn(fun() ->
case rpc:call(OtherNode, inet_tcp_proxy_dist, allow, [PartitionedNodeName], 5000) of
ok -> ok;
Err ->
log("~s rpc:call(~p, inet_tcp_proxy_dist, allow, [~p]) failed with ~p~n",
[timestamp(),
OtherNode,
PartitionedNodeName,
Err]),
ok
end
end)
end || OtherNode <- OtherNodes],
% Give nodes time to reconnect
timer:sleep(5000),
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1,
partition_state => #{}};
_ ->
MembersList = maps:keys(Members),
case length(MembersList) < 3 of
true ->
% Need at least 3 nodes for partition
State;
false ->
% No active partition, create a new one
% Pick a random node to partition
NodeToPartition = lists:nth(rand:uniform(length(MembersList)), MembersList),
NodeToPartitionName = element(2, NodeToPartition),
% Get the other nodes (excluding the one to partition)
OtherNodes = [element(2, M) || M <- MembersList, M =/= NodeToPartition],
% Choose partition duration: 3s, 15s, 55s, or 90s
PartitionDurations = [3000, 15000, 55000, 90000],
Duration = lists:nth(rand:uniform(length(PartitionDurations)), PartitionDurations),
HealTime = Now + Duration,
log("~s Creating network partition: isolating node ~w from ~w for ~wms~n",
[timestamp(), NodeToPartitionName, OtherNodes, Duration]),
% Block communication between partitioned node and other nodes
% The harness node (current node) maintains access to all nodes
[begin
_ = erpc:call(NodeToPartitionName, inet_tcp_proxy_dist, block, [OtherNode]),
_ = erpc:call(OtherNode, inet_tcp_proxy_dist, block, [NodeToPartitionName]),
ok
end || OtherNode <- OtherNodes],
State#{operations_count => OpCount + 1,
successful_ops => SuccessOps + 1,
partition_state => #{partitioned_node => NodeToPartition,
heal_time => HealTime,
other_nodes => OtherNodes}}
end
end.
-spec wait_for_applied_index_convergence([ra:server_id()], non_neg_integer()) -> ok.
wait_for_applied_index_convergence(Members, MaxRetries) when MaxRetries > 0 ->
IndicesMap = get_applied_indices(Members),
Indices = maps:values(IndicesMap),
case lists:uniq(Indices) of
[_SingleIndex] ->
ok; % All nodes have converged
_MultipleIndices ->
timer:sleep(100), % Wait 100ms before retry
wait_for_applied_index_convergence(Members, MaxRetries - 1)
end;
wait_for_applied_index_convergence(Members, 0) ->
IndicesMap = get_applied_indices(Members),
log("~s WARNING: Applied index convergence timeout. Reported values: ~0p~n",
[timestamp(), IndicesMap]),
ok.
-spec get_applied_indices([ra:server_id()]) -> #{ra:server_id() => ra:index() | undefined}.
get_applied_indices(Members) ->
maps:from_list([{Member, case ra:member_overview(Member, 1000) of
{ok, #{last_applied := Index}, _} ->
Index;
_ ->
undefined
end} || Member <- Members]).
-spec validate_consistency(state()) -> state().
validate_consistency(State) ->
Members = maps:get(members, State),
PartitionState = maps:get(partition_state, State, #{}),
MembersList = maps:keys(Members),
% Determine which members to validate based on active partition
MembersToValidate = case PartitionState of
#{partitioned_node := PartitionedNode} ->
% Active partition - exclude the partitioned node
log("~s Consistency check during partition; skipping validation of partitioned node ~0p~n",
[timestamp(), partitioned_node_name(State)]),
lists:delete(PartitionedNode, MembersList);
_ ->
% No active partition, validate all members
MembersList
end,
% Perform consistency check on selected members
perform_consistency_check(State, MembersToValidate).
-spec perform_consistency_check(state(), [ra:server_id()]) -> state().
perform_consistency_check(State, MembersToValidate) ->
RefMap = maps:get(reference_map, State),
% Wait for all nodes to converge to the same applied index
wait_for_applied_index_convergence(MembersToValidate, 300), % Wait up to 30 seconds
% Check that all members have the same view
ValidationResults = [validate_member_consistency(Member, RefMap)
|| Member <- MembersToValidate],
Result1 = hd(ValidationResults),
case lists:all(fun(Result) ->
is_map(Result) andalso
is_map(Result1) andalso
lists:sort(maps:get(live_indexes, Result)) =:=
lists:sort(maps:get(live_indexes, Result1))
end, ValidationResults) of
true ->
State;
false ->
% Brief console output with live_indexes summary
LiveIndexesSummary = [{Member, case Result of
#{live_indexes := LI,
log := #{last_index := LastIndex}} ->
{length(LI), LastIndex};
_ -> error
end} || {Member, Result} <-
lists:zip(MembersToValidate, ValidationResults)],
log("~s Consistency check failed. Live indexes per node: ~p~n",
[timestamp(), LiveIndexesSummary]),
log("~s STOPPING: No more operations will be performed due to consistency failure~n", [timestamp()]),
% Write full details to log file with difference analysis
LogEntry = format_consistency_failure(MembersToValidate, ValidationResults),
ok = file:write_file("ra_kv_harness.log", LogEntry, [append]),
FailedOps = maps:get(failed_ops, State),
State#{failed_ops => FailedOps + 1, remaining_ops => 0, consistency_failed => true}
end.
-spec format_consistency_failure([ra:server_id()], [map() | error]) -> iolist().
format_consistency_failure(Members, Results) ->
MemberResults = lists:zip(Members, Results),
% Extract all unique results for comparison
UniqueResults = lists:usort([R || {_, R} <- MemberResults, R =/= error]),
Header = io_lib:format("~s Consistency check failed:~n", [timestamp()]),
% Log raw data
RawData = [io_lib:format(" Member ~p: ~p~n", [Member, Result]) || {Member, Result} <- MemberResults],
% Analyze differences
DiffAnalysis = case UniqueResults of
[] ->
[" ANALYSIS: All members returned errors\n"];
[_SingleResult] ->
[" ANALYSIS: All successful members have identical results (errors may exist)\n"];
MultipleResults ->
[" ANALYSIS: Found ~p different result patterns:\n" |
[io_lib:format(" Pattern ~p: ~p\n", [I, Pattern]) ||
{I, Pattern} <- lists:zip(lists:seq(1, length(MultipleResults)), MultipleResults)] ++
[" DIFFERENCES:\n"] ++
analyze_field_differences(MultipleResults)]
end,
[Header, RawData, DiffAnalysis, "\n"].
-spec analyze_field_differences([map()]) -> iolist().
analyze_field_differences(Results) ->
% Extract live_indexes and num_keys for comparison
LiveIndexes = [maps:get(live_indexes, R, undefined) || R <- Results, is_map(R)],
NumKeys = [maps:get(num_keys, R, undefined) || R <- Results, is_map(R)],
LiveIndexDiff = case lists:usort(LiveIndexes) of
[_] -> [];
MultipleLI -> [io_lib:format(" live_indexes differ: ~p\n", [MultipleLI])]
end,
NumKeysDiff = case lists:usort(NumKeys) of
[_] -> [];
MultipleNK -> [io_lib:format(" num_keys differ: ~p\n", [MultipleNK])]
end,
[LiveIndexDiff, NumKeysDiff].
-spec validate_member_consistency(ra:server_id(), map()) -> map() | error.
validate_member_consistency(Member, _RefMap) ->
case ra_kv:member_overview(Member) of
#{log := Log,
machine := #{live_indexes := Live, num_keys := Num}} ->
%io:format("Member ~p overview: Live indexes ~p, Num keys ~p", [Member, Live, Num]),
#{log => Log,
live_indexes => Live,
num_keys => Num};
Error ->
log("~s Member ~p failed overview check: ~p~n",
[timestamp(), Member, Error]),
error
end.
-spec validate_final_consistency(state()) -> non_neg_integer().
validate_final_consistency(State) ->
Members = maps:get(members, State),
RefMap = maps:get(reference_map, State),
log("~s Performing final consistency validation...~n", [timestamp()]),
log("~s Reference map has ~p keys~n", [timestamp(), maps:size(RefMap)]),
% Wait for all nodes to converge before final validation
MembersList = maps:keys(Members),
log("~s Waiting for applied index convergence...~n", [timestamp()]),
wait_for_applied_index_convergence(MembersList, 100), % Wait up to 10 seconds for final check
% Validate all keys across all members
Keys = maps:keys(RefMap),
MembersList = maps:keys(Members),
ValidationCount = lists:foldl(
fun(Key, Acc) ->
RefValue = maps:get(Key, RefMap),
case validate_key_across_members(Key, RefValue, MembersList) of
ok -> Acc + 1;
error -> Acc
end
end, 0, Keys),
log("~s Final consistency check: ~p/~p keys validated successfully~n",
[timestamp(), ValidationCount, length(Keys)]),
ValidationCount.
-spec validate_key_across_members(binary(), term(), [ra:server_id()]) -> ok | error.
validate_key_across_members(Key, ExpectedValue, Members) ->
Results = [begin
case erpc:call(Node, ra_kv, get, [Member, Key, ?TIMEOUT]) of
{ok, _Meta, Value} when Value =:= ExpectedValue -> ok;
{ok, _Meta, Value} ->
log("~s Key ~p mismatch on ~p: expected ~p, got ~p~n",
[timestamp(), Key, Member, ExpectedValue, Value]),
error;
{error, not_found} ->
log("~s Key ~p not found on ~p but should exist~n", [timestamp(), Key, Member]),
error;
Other ->
log("~s Key ~p query failed on ~p: ~p~n", [timestamp(), Key, Member, Other]),
error
end
end || {_, Node} = Member <- Members],
case lists:all(fun(R) -> R =:= ok end, Results) of
true -> ok;
false -> error
end.
partitioned_node_name(#{partition_state := #{partitioned_node := {_, Node}}}) ->
Node;
partitioned_node_name(_State) ->
none.
random_non_partitioned_member(MemberList, State) ->
PartitionState = maps:get(partition_state, State, #{}),
PartitionedNode = maps:get(partitioned_node, PartitionState, none),
MajorityMembers = lists:delete(PartitionedNode, MemberList),
lists:nth(rand:uniform(length(MajorityMembers)), MajorityMembers).
random_non_partitioned_member(State) ->
Members = maps:get(members, State, #{}),
PartitionState = maps:get(partition_state, State, #{}),
PartitionedNode = maps:get(partitioned_node, PartitionState, none),
MajorityMembers = lists:delete(PartitionedNode, maps:keys(Members)),
lists:nth(rand:uniform(length(MajorityMembers)), MajorityMembers).