diff --git a/include/riak_repl.hrl b/include/riak_repl.hrl index 9d27b80e..9fa31c2c 100644 --- a/include/riak_repl.hrl +++ b/include/riak_repl.hrl @@ -49,6 +49,9 @@ -define(BT_META_BUCKET_NAME, bucket_name). -define(BT_META_BLACKLIST, realtime_blacklist). +-define(RIAK_REPL2_RTQ_CONFIG_KEY, {riak_repl2_rtq, config}). +-define(RTSOURCE_REALTIME_CONNECTIONS_KEY, {riak_repl2_rtsource, realtime_connections}). + -type(ip_addr_str() :: string()). -type(ip_portnum() :: non_neg_integer()). -type(repl_addr() :: {ip_addr_str(), ip_portnum()}). diff --git a/src/riak_core_cluster_mgr.erl b/src/riak_core_cluster_mgr.erl index 2462ab8f..08d4f025 100644 --- a/src/riak_core_cluster_mgr.erl +++ b/src/riak_core_cluster_mgr.erl @@ -807,7 +807,7 @@ shuffle(List) -> get_my_remote_ip_list(_Remote, [], _Return) -> {ok, []}; get_my_remote_ip_list(Remote, RemoteUnsorted, Return) -> - case riak_repl2_rtsource_conn_data_mgr:read(active_nodes) of + case riak_repl_util:read_realtime_active_nodes() of no_leader -> {ok, []}; %% SourceNodes = [SourceNode1, SourceNode3, SourceNode2 ...] @@ -871,7 +871,7 @@ build_primary_dict([{Key, Value}| Rest], Dict) -> %% sink nodes to an index which would cause active connections to be dropped and made by other nodes. %% Output: dict -> Key-Value: {Index, SinkNode} link_addrs(AllPrimariesDict, SinkNodes, Remote) -> - case riak_repl2_rtsource_conn_data_mgr:read(realtime_connections, Remote) of + case riak_repl_util:read_realtime_endpoints(Remote) of no_leader -> no_leader; %% ActiveConnsDict: this is a dictionary of active realtime connections diff --git a/src/riak_repl2_rtsink_conn.erl b/src/riak_repl2_rtsink_conn.erl index 60a52843..141227ce 100644 --- a/src/riak_repl2_rtsink_conn.erl +++ b/src/riak_repl2_rtsink_conn.erl @@ -493,14 +493,6 @@ riak_repl2_rtsink_conn_test_() -> }]}. setup() -> - riak_repl_test_util:start_test_ring(), - riak_repl_test_util:abstract_gen_tcp(), - riak_repl_test_util:kill_and_wait(riak_repl2_rt), - {ok, _RT} = riak_repl2_rt:start_link(), - riak_repl_test_util:kill_and_wait(riak_repl2_rtq), - {ok, _} = riak_repl2_rtq:start_link(), - application:set_env(riak_repl, realtime_connection_rebalance_max_delay_secs, 1000), - catch(meck:unload(riak_core_connection_mgr)), meck:new(riak_core_connection_mgr, [passthrough]), meck:expect(riak_core_connection_mgr, disconnect, @@ -508,21 +500,6 @@ setup() -> ok end), - catch(meck:unload(riak_repl2_rtsource_conn_data_mgr)), - meck:new(riak_repl2_rtsource_conn_data_mgr, [passthrough]), - meck:expect(riak_repl2_rtsource_conn_data_mgr, read, fun(active_nodes) -> [node()] - end), - meck:expect(riak_repl2_rtsource_conn_data_mgr, read, - fun(realtime_connections, _Remote) -> - dict:new() - end - ), - meck:expect(riak_repl2_rtsource_conn_data_mgr, write, 4, - fun(_, _, _, _) -> - ok - end - ), - catch(meck:unload(riak_core_cluster_mgr)), meck:new(riak_core_cluster_mgr, [passthrough]), meck:expect(riak_core_cluster_mgr, get_unshuffled_ipaddrs_of_cluster, fun(_Remote) -> [] end ), @@ -552,6 +529,15 @@ setup() -> {reply, ok, #state{transport=Transport, socket=Socket}} end), + + + riak_repl_test_util:start_test_ring(), + riak_repl_test_util:abstract_gen_tcp(), + riak_repl_test_util:kill_and_wait(riak_repl2_rt), + {ok, _RT} = riak_repl2_rt:start_link(), + riak_repl_test_util:kill_and_wait(riak_repl2_rtq), + {ok, _} = riak_repl2_rtq:start_link(), + application:set_env(riak_repl, realtime_connection_rebalance_max_delay_secs, 1000), folsom:start(), ok. @@ -565,7 +551,7 @@ cleanup(_Ctx) -> riak_repl_test_util:stop_test_ring(), riak_repl_test_util:maybe_unload_mecks( [riak_core_service_mgr, - riak_repl2_rtsource_conn_data_mgr, + riak_repl_util, riak_core_connection_mgr, riak_repl_util, riak_core_tcp_mon, @@ -632,6 +618,14 @@ start_source(NegotiatedVer) -> end), + meck:expect(riak_repl_util, read_realtime_active_nodes, fun() -> [node()] end), + meck:expect(riak_repl_util, read_realtime_active_nodes, fun(_) -> [node()] end), + meck:expect(riak_repl_util, read_realtime_endpoints, fun(_) -> dict:new() end), + meck:expect(riak_repl_util, read_realtime_endpoints, fun(_, _) -> dict:new() end), + meck:expect(riak_repl_util, write_realtime_endpoints, fun(_, _) -> ok end), + meck:expect(riak_repl_util, write_realtime_endpoints, fun(_, _, _) -> ok end), + meck:expect(riak_repl_util, write_realtime_endpoints, fun(_, _, _, _) -> ok end), + meck:expect(riak_repl_util, delete_realtime_endpoints, fun(_) -> ok end), {ok, SourcePid} = riak_repl2_rtsource_conn_mgr:start_link("sink_cluster"), receive {sink_started, SinkPid} -> @@ -653,7 +647,7 @@ unload_mecks() -> riak_repl_test_util:maybe_unload_mecks([ stateful, riak_core_ring_manager, riak_core_ring, riak_repl2_rtsink_helper, gen_tcp, fake_source, riak_repl2_rtq, - riak_core_capability, riak_repl2_rtsource_conn_data_mgr]). + riak_core_capability, riak_repl2_rtsource_conn_data_mgr, riak_repl_util]). -endif. diff --git a/src/riak_repl2_rtsource_conn_data_mgr.erl b/src/riak_repl2_rtsource_conn_data_mgr.erl index f24ec818..4b332265 100644 --- a/src/riak_repl2_rtsource_conn_data_mgr.erl +++ b/src/riak_repl2_rtsource_conn_data_mgr.erl @@ -30,8 +30,7 @@ is_leader, connections, active_nodes, - restoration, - core_capability_polling_interval + restoration }). @@ -78,23 +77,27 @@ node_watcher_update(_Services) -> %%%=================================================================== init([]) -> + riak_core_node_watcher_events:add_sup_callback(fun ?MODULE:node_watcher_update/1), + + CoreCapabilityPollingIntervalSecs = app_helper:get_env(riak_repl, realtime_core_capability_polling_interval, 60), + CoreCapabilityPollingInterval = CoreCapabilityPollingIntervalSecs * 1000, + erlang:send_after(CoreCapabilityPollingInterval, self(), poll_core_capability), + Version = riak_core_capability:get({riak_repl, realtime_connections}, legacy), State = case Version of legacy -> legacy_init(#state{}); v1 -> - v1_init(#state{}) + v1_init(#state{}); + v2 -> + #state{version = v2} end, {ok, State}. legacy_init(State) -> - riak_core_node_watcher_events:add_sup_callback(fun ?MODULE:node_watcher_update/1), - CoreCapabilityPollingIntervalSecs = app_helper:get_env(riak_repl, realtime_core_capability_polling_interval, 60), - CoreCapabilityPollingInterval = CoreCapabilityPollingIntervalSecs * 1000, - erlang:send_after(CoreCapabilityPollingInterval, self(), poll_core_capability), State#state{version = legacy, leader_node = undefined, is_leader = false, connections = dict:new(), active_nodes = [], - restoration = false, core_capability_polling_interval = CoreCapabilityPollingInterval}. + restoration = false}. v1_init(State) -> {Leader, IsLeader} = @@ -124,11 +127,7 @@ v1_init(State) -> C = dict:new(), AN = [], - riak_core_node_watcher_events:add_sup_callback(fun ?MODULE:node_watcher_update/1), - CoreCapabilityPollingIntervalSecs = app_helper:get_env(riak_repl, realtime_core_capability_polling_interval, 60), - CoreCapabilityPollingInterval = CoreCapabilityPollingIntervalSecs * 1000, - State#state{version = v1, leader_node = Leader, is_leader = IsLeader, connections = C, active_nodes = AN, restoration = false, - core_capability_polling_interval = CoreCapabilityPollingInterval}. + State#state{version = v1, leader_node = Leader, is_leader = IsLeader, connections = C, active_nodes = AN, restoration = false}. %% -------------------------------------------------- Read ---------------------------------------------------------- %% @@ -182,6 +181,8 @@ handle_call(_Request, _From, State) -> {reply, ok, State}. +handle_cast({set_leader_node, _LeaderNode}, State=#state{version = v2}) -> + {noreply, State}; handle_cast({set_leader_node, LeaderNode}, State=#state{version = V}) -> lager:info("setting leader node as: ~p", [LeaderNode]), case {V, node()} of @@ -190,13 +191,15 @@ handle_cast({set_leader_node, LeaderNode}, State=#state{version = V}) -> {noreply, State#state{leader_node = LeaderNode, is_leader = true}}; {legacy, _} -> {noreply, State#state{leader_node = LeaderNode, is_leader = false}}; - {v1, LeaderNode} -> + {_, LeaderNode} -> gen_server:cast(?SERVER, node_watcher_update), {noreply, become_leader(State, LeaderNode)}; - {v1, _} -> + {_, _} -> {noreply, become_proxy(State, LeaderNode)} end; +handle_cast(node_watcher_update, State=#state{version = v2}) -> + {noreply, State}; handle_cast(node_watcher_update, State=#state{active_nodes = OldActiveNodes, connections = C}) when State#state.is_leader == true -> NewActiveNodes = riak_core_node_watcher:nodes(riak_kv), DownNodes = OldActiveNodes -- NewActiveNodes, @@ -302,8 +305,7 @@ handle_cast(Msg = {write_realtime_connections, Remote, Node, ConnectionList}, St {noreply, State#state{connections = NewConnections}}; false -> - lager:info("data_mgr is proxy sending to leader -> ~p - node = ~p", [Msg, node()]), + lager:info("data_mgr is proxy sending to leader -> ~p node = ~p", [Msg, node()]), proxy_cast(Msg, State), {noreply, State} end; @@ -342,18 +344,26 @@ handle_info(restore_leader_data, State) -> handle_info(reset_restoration_flag, State) -> {noreply, State#state{restoration = false}}; -handle_info(poll_core_capability, State=#state{version = OldVersion, core_capability_polling_interval = PI}) -> - NewVersion = riak_core_capability:get({riak_repl, realtime_connections}, legacy), - case {OldVersion, NewVersion} of - {legacy, legacy} -> +handle_info(poll_core_capability, State=#state{version = OldVersion}) -> + CoreCapabilityPollingIntervalSecs = app_helper:get_env(riak_repl, realtime_core_capability_polling_interval, 60), + PI = CoreCapabilityPollingIntervalSecs * 1000, + case {OldVersion, riak_core_capability:get({riak_repl, realtime_connections}, legacy)} of + {_, v2} -> + case State#state.leader_node == node() of + true -> + riak_core_ring_manager:ring_trans(fun riak_repl_ring:cleanup_realtime_connection_data/2, []); + false -> + ok + end, + {noreply, State#state{version = v2}}; + {v1, v1} -> erlang:send_after(PI, self(), poll_core_capability), {noreply, State}; {legacy, v1} -> - %% send upgrade message to conn_mgr after 30 seconds to give the data manager time to get into the correct state - %% after changing versions - [erlang:send_after(30000, Pid, upgrade_connection_version) || {_Remote, Pid} <- riak_repl2_rtsource_conn_sup:enabled()], - {noreply, v1_init(State)}; - {v1, _} -> + erlang:send_after(PI, self(), poll_core_capability), + {noreply, v1_init(State#state{version = v1})}; + _ -> + erlang:send_after(PI, self(), poll_core_capability), {noreply, State} end; @@ -365,8 +375,7 @@ handle_info(_Info, State) -> terminate(Reason, State) -> - lager:info("riak_repl2_rtsource_conn_data_mgr termianting due to: ~p - State: ~p", [Reason, State]), + lager:info("riak_repl2_rtsource_conn_data_mgr termianting due to: ~p State: ~p", [Reason, State]), ok. code_change(_OldVsn, State, _Extra) -> diff --git a/src/riak_repl2_rtsource_conn_mgr.erl b/src/riak_repl2_rtsource_conn_mgr.erl index 46519108..977485e3 100644 --- a/src/riak_repl2_rtsource_conn_mgr.erl +++ b/src/riak_repl2_rtsource_conn_mgr.erl @@ -1,6 +1,8 @@ -module(riak_repl2_rtsource_conn_mgr). -behaviour(gen_server). +-include("riak_repl.hrl"). + %% API -export([start_link/1]). @@ -20,13 +22,16 @@ stop/1, get_all_status/1, get_all_status/2, - get_source_and_sink_nodes/1, + get_source_and_sink_nodes/2, get_endpoints/1, - get_rtsource_conn_pids/1 + get_rtsource_conn_pids/1, + + node_watcher_update/1, + set_leader/2 ]). -define(SERVER, ?MODULE). - +-define(CURRENT_VERSION, v2). -define(CLIENT_SPEC, {{realtime,[{3,0}, {2,0}, {1,5}]}, {?TCP_OPTIONS, ?SERVER, self()}}). @@ -41,12 +46,11 @@ connection_ref, % reference handed out by connection manager rb_timeout_tref, % Rebalance timeout timer reference rebalance_delay_fun, - rebalance_timer, - max_delay, source_nodes, sink_nodes, remove_endpoint, - endpoints + endpoints, + leader }). %%%=================================================================== @@ -80,6 +84,7 @@ stop(Pid) -> get_all_status(Pid) -> get_all_status(Pid, infinity). + get_all_status(Pid, Timeout) -> gen_server:call(Pid, all_status, Timeout). @@ -89,6 +94,12 @@ get_endpoints(Pid) -> get_rtsource_conn_pids(Pid) -> gen_server:call(Pid, get_rtsource_conn_pids). +node_watcher_update(Pid) -> + gen_server:cast(Pid, node_watcher_update). + +set_leader(Pid, LeaderNode) -> + gen_server:cast(Pid, {set_leader_node, LeaderNode}). + %%%=================================================================== %%% gen_server callbacks @@ -97,42 +108,41 @@ get_rtsource_conn_pids(Pid) -> init([Remote]) -> process_flag(trap_exit, true), - _ = riak_repl2_rtq:register(Remote), % re-register to reset stale deliverfun E = dict:new(), - MaxDelaySecs = app_helper:get_env(riak_repl, realtime_connection_rebalance_max_delay_secs, 120), M = fun(X) -> round(X * crypto:rand_uniform(0, 1000)) end, - RebalanceTimer = app_helper:get_env(riak_repl, realtime_rebalance_on_failure, 5), + {Version, ConnectionType} = + case riak_core_capability:get({riak_repl, realtime_connections}, legacy) of + legacy -> + {legacy, legacy}; + V -> + {V, multi_connection} + end, + IntervalSecs = app_helper:get_env(riak_repl, realtime_core_capability_polling_interval, 60), + Time = IntervalSecs * 1000, + case Version == ?CURRENT_VERSION of + true -> ok; + false -> erlang:send_after(Time, self(), poll_core_capability) + end, - case riak_core_capability:get({riak_repl, realtime_connections}, legacy) of - legacy -> - case riak_core_connection_mgr:connect({rt_repl, Remote}, ?CLIENT_SPEC, legacy) of - {ok, Ref} -> - {ok, #state{version = legacy, remote = Remote, connection_ref = Ref, endpoints = E, rebalance_delay_fun = M, - rebalance_timer=RebalanceTimer*1000, max_delay=MaxDelaySecs, source_nodes = [], sink_nodes = []}}; - {error, Reason}-> - lager:warning("Error connecting to remote, verions: legacy"), - {stop, Reason} - end; - v1 -> - riak_repl2_rtsource_conn_data_mgr:write(realtime_connections, Remote, node(), []), - case riak_core_connection_mgr:connect({rt_repl, Remote}, ?CLIENT_SPEC, multi_connection) of - {ok, Ref} -> - {SourceNodes, SinkNodes} = case get_source_and_sink_nodes(Remote) of - no_leader -> - {[], []}; - {X,Y} -> - {X,Y} - end, - - {ok, #state{version = v1, remote = Remote, connection_ref = Ref, endpoints = E, rebalance_delay_fun = M, - rebalance_timer=RebalanceTimer*1000, max_delay=MaxDelaySecs, source_nodes = SourceNodes, sink_nodes = SinkNodes}}; - {error, Reason}-> - lager:warning("Error connecting to remote, verions: v1"), - {stop, Reason} - end + riak_repl_util:write_realtime_endpoints(Version, Remote, []), + case riak_core_connection_mgr:connect({rt_repl, Remote}, ?CLIENT_SPEC, ConnectionType) of + {ok, Ref} -> + State = #state{version = Version, remote = Remote, connection_ref = Ref, endpoints = E, + rebalance_delay_fun = M}, + {SourceNodes, SinkNodes} = + case get_source_and_sink_nodes(State, Remote) of + no_leader -> + {[], []}; + {X,Y} -> + {X,Y} + end, + {ok, State#state{source_nodes = SourceNodes, sink_nodes = SinkNodes}}; + {error, Reason}-> + lager:warning("Error connecting to remote, verions: v1"), + {stop, Reason} end. %%%===================================================================================================================== @@ -169,10 +179,10 @@ handle_call({connected, Socket, Transport, IPPort, Proto, _Props, Primary}, _Fro legacy -> lager:info("Adding remote connection, however not sending to data_mgr as we are running legacy code base"), ok; - v1 -> + _ -> % save to ring lager:info("Adding remote connections to data_mgr: ~p", [dict:fetch_keys(NewEndpoints)]), - riak_repl2_rtsource_conn_data_mgr:write(realtime_connections, Remote, node(), dict:fetch_keys(NewEndpoints)) + riak_repl_util:write_realtime_endpoints(V, Remote, dict:fetch_keys(NewEndpoints)) end, {reply, ok, NewState#state{endpoints = NewEndpoints}}; @@ -220,14 +230,36 @@ handle_cast({connect_failed, Reason}, State = #state{remote = Remote, endpoints end, {noreply, NewState}; -handle_cast(rebalance_delayed, State=#state{version = Version}) -> - case Version of - legacy -> - {noreply, maybe_rebalance_legacy(State, delayed)}; - v1 -> - {noreply, maybe_rebalance(State, delayed)} +handle_cast(rebalance_delayed, State=#state{version = legacy}) -> + {noreply, maybe_rebalance_legacy(State, delayed)}; + +handle_cast(rebalance_delayed, State) -> + {noreply, maybe_rebalance(State, delayed)}; + +handle_cast({set_leader_node, LeaderNode}, State) -> + lager:info("setting leader node as: ~p", [LeaderNode]), + case node() of + LeaderNode -> + %% we do this incase it was the previous leader that died! + %% we need to remove its connection data from the shared data store between all nodes + handle_cast(node_watcher_update, State#state{leader=true}); + _ -> + {noreply, State#state{leader = false}} end; +handle_cast(node_watcher_update, State=#state{leader = true, version = v2, remote = Remote}) -> + %% If I am the leader, go ahead and loop the data in core_metadata and delete any data for nodes that are now down. + %% ONLY FOR MY REMOTE! + NewNodes = riak_core_node_watcher:nodes(riak_kv), + AllRemoteEndpoints = riak_repl_util:read_realtime_endpoints(v2, Remote), + OldNodes = dict:fetch_keys(AllRemoteEndpoints), + DownNodes = OldNodes -- NewNodes, + lists:foreach(fun(Node) -> riak_repl_util:write_realtime_endpoints(v2, Remote, [], Node) end, DownNodes), + handle_cast(rebalance_delayed, State); + +handle_cast(node_watcher_update, State) -> + handle_cast(rebalance_delayed, State); + handle_cast(Request, State) -> lager:warning("unhandled cast: ~p", [Request]), {noreply, State}. @@ -239,15 +271,7 @@ handle_info({'EXIT', Pid, Reason}, State = #state{endpoints = E, remote = Remote {Key, Pid} -> NewEndpoints = dict:erase(Key, E), State2 = State#state{endpoints = NewEndpoints}, - {IPPort, Primary} = Key, - - case Version of - legacy -> - ok; - v1 -> - riak_repl2_rtsource_conn_data_mgr:delete(realtime_connections, Remote, node(), IPPort, Primary) - end, - + riak_repl_util:write_realtime_endpoints(Version, Remote, dict:fetch_keys(NewEndpoints)), case Reason of normal -> lager:info("riak_repl2_rtsource_conn terminated due to reason nomral, Endpoint: ~p", [Key]); @@ -284,14 +308,28 @@ handle_info(rebalance_now, State=#state{version = Version}) -> case Version of legacy -> {noreply, maybe_rebalance_legacy(State#state{rb_timeout_tref = undefined}, now)}; - v1 -> + _ -> {noreply, maybe_rebalance(State#state{rb_timeout_tref = undefined}, now)} end; -handle_info(upgrade_connection_version, State) -> - cancel_timer(State#state.rb_timeout_tref), - RbTimeoutTref = erlang:send_after(0, self(), rebalance_now), - {noreply, State#state{version = v1, rb_timeout_tref = RbTimeoutTref}}; + +handle_info(poll_core_capability, State=#state{version = OldVersion}) -> + IntervalSecs = app_helper:get_env(riak_repl, realtime_core_capability_polling_interval, 60), + Time = IntervalSecs * 1000, + case riak_core_capability:get({riak_repl, realtime_connections}, legacy) of + ?CURRENT_VERSION -> + cancel_timer(State#state.rb_timeout_tref), + RbTimeoutTref = erlang:send_after(0, self(), rebalance_now), + {noreply, State#state{version = ?CURRENT_VERSION, rb_timeout_tref = RbTimeoutTref}}; + OldVersion -> + erlang:send_after(Time, self(), poll_core_capability), + {noreply, State}; + OtherVersion -> + cancel_timer(State#state.rb_timeout_tref), + RbTimeoutTref = erlang:send_after(0, self(), rebalance_now), + erlang:send_after(Time, self(), poll_core_capability), + {noreply, State#state{version = OtherVersion, rb_timeout_tref = RbTimeoutTref}} + end; handle_info(Info, State) -> lager:warning("unhandled info: ~p", [Info]), @@ -299,8 +337,9 @@ handle_info(Info, State) -> %%%===================================================================================================================== -terminate(Reason, _State=#state{remote = Remote, endpoints = E}) -> +terminate(Reason, _State=#state{version = V, remote = Remote, endpoints = E}) -> lager:info("rtrsource conn mgr terminating, Reason: ~p", [Reason]), + riak_repl_util:write_realtime_endpoints(V, Remote, []), riak_core_connection_mgr:disconnect({rt_repl, Remote}), [catch riak_repl2_rtsource_conn:stop(Pid) || {{{_IP, _Port},_Primary},Pid} <- dict:to_list(E)], ok. @@ -314,12 +353,13 @@ code_change(_OldVsn, State, _Extra) -> %%%=================================================================== %%% Internal functions %%%=================================================================== -maybe_rebalance(State, now) -> - case get_source_and_sink_nodes(State#state.remote) of +maybe_rebalance(#state{version = V} = State, now) -> + RebalanceTimer = app_helper:get_env(riak_repl, realtime_rebalance_on_failure, 5) * 1000, + case get_source_and_sink_nodes(V, State#state.remote) of no_leader -> lager:info("rebalancing triggered, but an error occured with regard to the leader in the data mgr, will rebalance in 5 seconds"), cancel_timer(State#state.rb_timeout_tref), - RbTimeoutTref = erlang:send_after(State#state.rebalance_timer, self(), rebalance_now), + RbTimeoutTref = erlang:send_after(RebalanceTimer, self(), rebalance_now), State#state{rb_timeout_tref = RbTimeoutTref}; {NewSource, NewSink} -> case should_rebalance(State, NewSource, NewSink) of @@ -329,21 +369,21 @@ maybe_rebalance(State, now) -> inconsistent_data -> lager:info("rebalancing triggered, but an inconsistent data was found in realtime connections on data mgr, will rebalance in 5 seconds"), - riak_repl2_rtsource_conn_data_mgr:write(realtime_connections, State#state.remote, node(), dict:fetch_keys(State#state.endpoints)), + riak_repl_util:write_realtime_endpoints(V, State#state.remote, dict:fetch_keys(State#state.endpoints)), cancel_timer(State#state.rb_timeout_tref), - RbTimeoutTref = erlang:send_after(State#state.rebalance_timer, self(), rebalance_now), + RbTimeoutTref = erlang:send_after(RebalanceTimer, self(), rebalance_now), State#state{rb_timeout_tref = RbTimeoutTref}; no_leader -> lager:info("rebalancing triggered, but an error occured with regard to the leader in the data mgr, will rebalance in 5 seconds"), cancel_timer(State#state.rb_timeout_tref), - RbTimeoutTref = erlang:send_after(State#state.rebalance_timer, self(), rebalance_now), + RbTimeoutTref = erlang:send_after(RebalanceTimer, self(), rebalance_now), State#state{rb_timeout_tref = RbTimeoutTref}; rebalance_needed_empty_list_returned -> lager:info("rebalancing triggered but get_ip_addrs_of_cluster returned [], will rebalance in 5 seconds"), cancel_timer(State#state.rb_timeout_tref), - RbTimeoutTref = erlang:send_after(State#state.rebalance_timer, self(), rebalance_now), + RbTimeoutTref = erlang:send_after(RebalanceTimer, self(), rebalance_now), State#state{rb_timeout_tref = RbTimeoutTref}; {true, {equal, DropNodes, ConnectToNodes, _Primary, _Secondary, _ConnectedSinkNodes}} -> @@ -364,7 +404,7 @@ maybe_rebalance(State, now) -> "drop nodes ~p ~n" "connect nodes ~p", [DropNodes, ConnectToNodes]), {_RemoveAllConnections, NewState1} = check_and_drop_connections(State, DropNodes, ConnectedSinkNodes), - riak_repl2_rtsource_conn_data_mgr:write(realtime_connections, NewState1#state.remote, node(), dict:fetch_keys(NewState1#state.endpoints)), + riak_repl_util:write_realtime_endpoints(V, NewState1#state.remote, dict:fetch_keys(NewState1#state.endpoints)), NewState1#state{sink_nodes = NewSink, source_nodes = NewSource}; {true, {nodes_up_and_down, DropNodes, ConnectToNodes, Primary, Secondary, ConnectedSinkNodes}} -> @@ -373,7 +413,7 @@ maybe_rebalance(State, now) -> "connect nodes ~p", [DropNodes, ConnectToNodes]), {RemoveAllConnections, NewState1} = check_and_drop_connections(State, DropNodes, ConnectedSinkNodes), NewState2 = check_remove_endpoint(NewState1, ConnectToNodes), - riak_repl2_rtsource_conn_data_mgr:write(realtime_connections, NewState2#state.remote, node(), dict:fetch_keys(NewState2#state.endpoints)), + riak_repl_util:write_realtime_endpoints(V, NewState2#state.remote, dict:fetch_keys(NewState2#state.endpoints)), case RemoveAllConnections of true -> rebalance_connect(NewState2#state{sink_nodes = NewSink, source_nodes = NewSource}, Primary++Secondary); @@ -383,10 +423,11 @@ maybe_rebalance(State, now) -> end end; -maybe_rebalance(State=#state{rebalance_delay_fun = Fun, max_delay = M}, delayed) -> +maybe_rebalance(State=#state{rebalance_delay_fun = Fun}, delayed) -> case State#state.rb_timeout_tref of undefined -> - RbTimeoutTref = erlang:send_after(Fun(M), self(), rebalance_now), + MaxDelaySecs = app_helper:get_env(riak_repl, realtime_connection_rebalance_max_delay_secs, 120), + RbTimeoutTref = erlang:send_after(Fun(MaxDelaySecs), self(), rebalance_now), State#state{rb_timeout_tref = RbTimeoutTref}; _ -> %% Already sent a "rebalance_now" @@ -394,15 +435,16 @@ maybe_rebalance(State=#state{rebalance_delay_fun = Fun, max_delay = M}, delayed) end. -should_rebalance(State=#state{sink_nodes = OldSink, source_nodes = OldSource, remote = R}, NewSource, NewSink) -> +should_rebalance(State= + #state{sink_nodes = OldSink, source_nodes = OldSource, remote = R, version = V}, NewSource, NewSink) -> {SourceComparison, _SourceNodesDown, _SourceNodesUp} = compare_nodes(OldSource, NewSource), {SinkComparison, _SinkNodesDown, _SinkNodesUp} = compare_nodes(OldSink, NewSink), - RealtimeConnections = riak_repl2_rtsource_conn_data_mgr:read(realtime_connections, R), + RealtimeConnections = riak_repl_util:read_realtime_endpoints(V, R), case RealtimeConnections of no_leader -> no_leader; RTC -> - case check_data_mgr_conn_mgr_data_consistency(RTC, State) of + case check_realtime_connections_consistency(RTC, State) of true -> case {SourceComparison, SinkComparison} of {equal, equal} -> @@ -416,7 +458,7 @@ should_rebalance(State=#state{sink_nodes = OldSink, source_nodes = OldSource, re end end. -check_data_mgr_conn_mgr_data_consistency(RealtimeConnections, #state{endpoints = Endpoints}) -> +check_realtime_connections_consistency(RealtimeConnections, #state{endpoints = Endpoints}) -> case dict:find(node(), RealtimeConnections) of {ok, Conns} -> lists:sort(Conns) == lists:sort(dict:fetch_keys(Endpoints)); @@ -556,28 +598,33 @@ check_and_drop_connections(State=#state{endpoints = E, remote = R, version = V}, {false, State#state{endpoints = NewEndpoints, remove_endpoint = undefined}} end. -remove_connections([], E, _, _) -> + + +remove_connections(IPAddrs, Endpoints, Remote, Version) -> + NewDict = close_rtsource_conn(IPAddrs, Endpoints, Remote), + riak_repl_util:write_realtime_endpoints(Version, Remote, dict:fetch_keys(NewDict)), + NewDict. + +close_rtsource_conn([], E, _) -> E; -remove_connections([Key={Addr, Primary} | Rest], E, Remote, Version) -> +close_rtsource_conn([Key | Rest], E, Remote) -> RtSourcePid = dict:fetch(Key, E), exit(RtSourcePid, {shutdown, rebalance, Key}), lager:info("rtsource_conn is killed ~p", [Key]), - case Version of - legacy -> - ok; - v1 -> - riak_repl2_rtsource_conn_data_mgr:delete(realtime_connections, Remote, node(), Addr, Primary) - end, - remove_connections(Rest, dict:erase(Key, E), Remote, Version). + close_rtsource_conn(Rest, dict:erase(Key, E), Remote). -get_source_and_sink_nodes(Remote) -> - Source = riak_repl2_rtsource_conn_data_mgr:read(active_nodes), - case Source of - no_leader -> - no_leader; - SourceNodes -> - SinkNodes = riak_core_cluster_mgr:get_unshuffled_ipaddrs_of_cluster(Remote), - {SourceNodes, SinkNodes} +get_source_and_sink_nodes(#state{version = V}, Remote) -> + case V of + legacy -> {[], []}; + _ -> + Source = riak_repl_util:read_realtime_active_nodes(V), + case Source of + no_leader -> + no_leader; + SourceNodes -> + SinkNodes = riak_core_cluster_mgr:get_unshuffled_ipaddrs_of_cluster(Remote), + {SourceNodes, SinkNodes} + end end. compare_nodes(Old, New) -> @@ -641,10 +688,11 @@ collect_status_data([Key | Rest], Timeout, Data, E) -> %% Legacy Code %% %% ------------------------------------------------------------------------------------------------------------------ %% -maybe_rebalance_legacy(State=#state{rebalance_delay_fun = Fun, max_delay = M}, delayed) -> +maybe_rebalance_legacy(State=#state{rebalance_delay_fun = Fun}, delayed) -> case State#state.rb_timeout_tref of undefined -> - RbTimeoutTref = erlang:send_after(Fun(M), self(), rebalance_now), + MaxDelaySecs = app_helper:get_env(riak_repl, realtime_connection_rebalance_max_delay_secs, 120), + RbTimeoutTref = erlang:send_after(Fun(MaxDelaySecs), self(), rebalance_now), State#state{rb_timeout_tref = RbTimeoutTref}; _ -> %% Already sent a "rebalance_now" diff --git a/src/riak_repl2_rtsource_conn_sup.erl b/src/riak_repl2_rtsource_conn_sup.erl index a6a63e25..650474ee 100644 --- a/src/riak_repl2_rtsource_conn_sup.erl +++ b/src/riak_repl2_rtsource_conn_sup.erl @@ -6,7 +6,10 @@ start_link/0, enable/1, disable/1, - enabled/0 + enabled/0, + + set_leader/2, + node_watcher_update/1 ]). -export([init/1]). @@ -26,18 +29,31 @@ disable(Remote) -> lager:info("Stopping replication realtime source ~p", [Remote]), _ = supervisor:terminate_child(?MODULE, Remote), _ = supervisor:delete_child(?MODULE, Remote), - riak_repl2_rtsource_conn_data_mgr:delete(realtime_connections, Remote). + riak_repl_util:delete_realtime_endpoints(Remote). enabled() -> - [ {Remote, ConnMgrPid} || {Remote, ConnMgrPid, _, [riak_repl2_rtsource_conn_mgr]} + [{Remote, ConnMgrPid} || {Remote, ConnMgrPid, _, [riak_repl2_rtsource_conn_mgr]} <- supervisor:which_children(?MODULE), is_pid(ConnMgrPid)]. +set_leader(LeaderNode, _LeaderPid) -> + lists:foreach( + fun({_, Pid, _, [riak_repl2_rtsource_conn_mgr]}) -> + riak_repl2_rtsource_conn_mgr:set_leader(Pid, LeaderNode) + end, supervisor:which_children(?MODULE)). + +node_watcher_update(_Services) -> + lists:foreach( + fun({_, Pid, _, [riak_repl2_rtsource_conn_mgr]}) -> + riak_repl2_rtsource_conn_mgr:node_watcher_update(Pid) + end, supervisor:which_children(?MODULE)). + %% @private init([]) -> %% TODO: Move before riak_repl2_rt_sup start %% once connmgr is started by core. Must be started/registered %% before sources are started. riak_repl2_rt:register_remote_locator(), + riak_core_node_watcher_events:add_sup_callback(fun ?MODULE:node_watcher_update/1), {ok, Ring} = riak_core_ring_manager:get_raw_ring(), Remotes = riak_repl_ring:rt_started(Ring), diff --git a/src/riak_repl_app.erl b/src/riak_repl_app.erl index 387522c1..4c699585 100644 --- a/src/riak_repl_app.erl +++ b/src/riak_repl_app.erl @@ -63,7 +63,7 @@ start(_Type, _StartArgs) -> riak_core_capability:register( {riak_repl, realtime_connections}, - [v1, legacy], + [v2, v1, legacy], legacy ), riak_core_capability:register( @@ -108,10 +108,12 @@ start(_Type, _StartArgs) -> riak_repl2_leader:register_notify_fun( fun riak_core_cluster_mgr:set_leader/2), + % TODO: remove in the next release (data_mgr will no longer be used in v2 realtime_connections) + riak_repl2_leader:register_notify_fun( + fun riak_repl2_rtsource_conn_data_mgr:set_leader/2), % rtsource supverisors -> rtsource_conn_mgr will follow the leader riak_repl2_leader:register_notify_fun( - fun riak_repl2_rtsource_conn_data_mgr:set_leader/2 - ), + fun riak_repl2_rtsource_conn_sup:set_leader/2), %% fullsync co-ordincation will follow leader riak_repl2_leader:register_notify_fun( diff --git a/src/riak_repl_util.erl b/src/riak_repl_util.erl index 4b2b1b5a..e75a6a7f 100644 --- a/src/riak_repl_util.erl +++ b/src/riak_repl_util.erl @@ -78,6 +78,89 @@ peer_wire_format/1 ]). +-export([ + delete_realtime_endpoints/1, + write_realtime_endpoints/4, + write_realtime_endpoints/3, + write_realtime_endpoints/2, + read_realtime_endpoints/2, + read_realtime_endpoints/1, + read_realtime_active_nodes/1, + read_realtime_active_nodes/0 + +]). + +% ---------------------------------------------------------------------------------- % +% Realtime Functions % +% ---------------------------------------------------------------------------------- % + +delete_realtime_endpoints(Remote) -> + Version = riak_core_capability:get({riak_repl, realtime_connections}, legacy), + Endpoints = read_realtime_endpoints(Version, Remote), + Keys = dict:fetch_keys(Endpoints), + lists:foreach(fun(Node) -> ?MODULE:write_realtime_endpoints(Version, Remote, Node, []) end, Keys). + +write_realtime_endpoints(Remote, Endpoints) -> + Version = riak_core_capability:get({riak_repl, realtime_connections}, legacy), + ?MODULE:write_realtime_endpoints(Version, Remote, Endpoints). +write_realtime_endpoints(Version, Remote, Endpoints) -> + ?MODULE:write_realtime_endpoints(Version, Remote, Endpoints, node()). +write_realtime_endpoints(v1, Remote, Endpoints, Node) -> + riak_repl2_rtsource_conn_data_mgr:write(realtime_connections, Remote, Node, Endpoints), + ?MODULE:write_realtime_endpoints(v2, Remote, Endpoints, Node); +write_realtime_endpoints(v2, Remote, Endpoints, Node) -> + riak_core_metadata:put(?RTSOURCE_REALTIME_CONNECTIONS_KEY, {Remote, Node}, Endpoints); +write_realtime_endpoints(_, _Remote, _Endpoints, _Node) -> + ok. + +read_realtime_endpoints(Remote) -> + Version = riak_core_capability:get({riak_repl, realtime_connections}, legacy), + read_realtime_endpoints(Version, Remote). +read_realtime_endpoints(legacy, _Remote) -> + []; +read_realtime_endpoints(v1, Remote) -> + riak_repl2_rtsource_conn_data_mgr:read(realtime_connections, Remote); +read_realtime_endpoints(v2, Remote) -> + ActiveNodes = read_realtime_active_nodes(v2), + iterate_realtime_endpoints(riak_core_metadata:iterator(?RTSOURCE_REALTIME_CONNECTIONS_KEY, + [{default, deleted}, {resolver, llw}, {match, {Remote, '_'}}]), [], ActiveNodes); +read_realtime_endpoints(Version, Remote) -> + lager:error("recieved unknown version! Remote: ~p Version: ~p", [Remote, Version]), + dict:new(). + +iterate_realtime_endpoints(Itr, Values, ActiveNodes) -> + case riak_core_metadata:itr_done(Itr) of + true -> + riak_core_metadata:itr_close(Itr), + dict:from_list(Values); + false -> + add_endpoints(ActiveNodes, riak_core_metadata:itr_key_values(Itr), Itr, Values) + end. + +add_endpoints(ActiveNodes, {Key, deleted}, Itr, Values) -> + add_endpoints(ActiveNodes, {Key, []}, Itr, Values); +add_endpoints(ActiveNodes, {_, []}, Itr, Values) -> + iterate_realtime_endpoints(ActiveNodes, riak_core_metadata:itr_next(Itr), Values); +add_endpoints(ActiveNodes, {{_Remote, Node}, SinkConns}, Itr, Values) -> + case lists:member(Node, ActiveNodes) of + true -> + iterate_realtime_endpoints(riak_core_metadata:itr_next(Itr), [{Node, SinkConns}| Values], ActiveNodes); + false -> + iterate_realtime_endpoints(ActiveNodes, riak_core_metadata:itr_next(Itr), Values) + end. + +read_realtime_active_nodes() -> + Version = riak_core_capability:get({riak_repl, realtime_connections}, legacy), + read_realtime_endpoints(Version). +read_realtime_active_nodes(legacy) -> + []; +read_realtime_active_nodes(v1) -> + riak_repl2_rtsource_conn_data_mgr:read(active_nodes); +read_realtime_active_nodes(v2) -> + riak_core_node_watcher:nodes(riak_kv). + +% ---------------------------------------------------------------------------------- % + %% Defines for Wire format encode/decode -define(MAGIC, 42). %% as opposed to 131 for Erlang term_to_binary or 51 for riak_object -define(W1_VER, 1). %% first non-just-term-to-binary wire format diff --git a/test/riak_core_cluster_mgr_tests.erl b/test/riak_core_cluster_mgr_tests.erl index 2c066738..4b35a2d8 100644 --- a/test/riak_core_cluster_mgr_tests.erl +++ b/test/riak_core_cluster_mgr_tests.erl @@ -70,7 +70,7 @@ single_node_test_() -> end}, {"regsiter restore cluster members fun", fun() -> - Fun = fun() -> [{test_name_locator,?REMOTE_CLUSTER_ADDR}] end, + Fun = fun() -> [{cluster_by_name,?REMOTE_CLUSTER_NAME}] end, riak_core_cluster_mgr:register_restore_cluster_targets_fun(Fun), ok end}, @@ -801,14 +801,14 @@ check_secondary([Output|Rest], Expected, true) -> check_secondary(Rest, Expected, lists:member(Output, Expected)). make_deps(ActiveSourceNodes, ActiveConns) -> - meck:expect(riak_repl2_rtsource_conn_data_mgr, read, - fun(active_nodes) -> - lists:reverse(ActiveSourceNodes) - end), - meck:expect(riak_repl2_rtsource_conn_data_mgr, read, - fun(realtime_connections, _Remote) -> - ActiveConns - end). + meck:expect(riak_repl_util, read_realtime_active_nodes, 0, + fun() -> lists:reverse(ActiveSourceNodes) end), + meck:expect(riak_repl_util, read_realtime_active_nodes, 1, + fun(_) -> lists:reverse(ActiveSourceNodes) end), + meck:expect(riak_repl_util, read_realtime_endpoints, 1, + fun(_) -> ActiveConns end), + meck:expect(riak_repl_util, read_realtime_endpoints, 2, + fun(_, _) -> ActiveConns end). wait_for(Fun) -> @@ -1031,6 +1031,9 @@ cleanup({Apps, Pids}) -> riak_core_service_mgr:stop(), riak_core_connection_mgr:stop(), %% tough to stop a supervisor + riak_repl_test_util:maybe_unload_mecks([ + riak_repl_util + ]), catch exit(riak_core_cluster_conn_sup), catch exit(riak_repl2_leader_gs), riak_core_cluster_mgr:stop(),