Merge pull request #7 from helium/bugfix/cluster-ops

Tie up some cluster management loose ends

Reviewed-by: kellymclaughlin

Author: he1000

src/plumtree_broadcast.erl

@@ -278,7 +278,8 @@ handle_cast({graft, MessageId, Mod, Round, Root, From}, State) ->
     State1 = handle_graft(Result, MessageId, Mod, Round, Root, From, State),
     {noreply, State1};
 handle_cast({update, LocalState}, State=#state{all_members=BroadcastMembers}) ->
-    CurrentMembers = ordsets:from_list(LocalState),
+    Members = riak_dt_orswot:value(LocalState),
+    CurrentMembers = ordsets:from_list(Members),
     New = ordsets:subtract(CurrentMembers, BroadcastMembers),
     Removed = ordsets:subtract(BroadcastMembers, CurrentMembers),
     State1 = case ordsets:size(New) > 0 of

src/plumtree_peer_service.erl

@@ -25,7 +25,7 @@
          join/3,
          attempt_join/1,
          attempt_join/2,
-         leave/0,
+         leave/1,
          stop/0,
          stop/1
         ]).
@@ -67,7 +67,7 @@ attempt_join(Node, Local) ->
     _ = [gen_server:cast({plumtree_peer_service_gossip, P}, {receive_state, Merged}) || P <- Members, P /= node()],
     ok.
 
-leave() ->
+leave(_Args) when is_list(_Args) ->
     {ok, Local} = plumtree_peer_service_manager:get_local_state(),
     {ok, Actor} = plumtree_peer_service_manager:get_actor(),
     {ok, Leave} = riak_dt_orswot:update({remove, node()}, Actor, Local),
@@ -81,14 +81,16 @@ leave() ->
             case [P || P <- Remote2List, P =:= node()] of
                 [] ->
                     %% leaving the cluster shuts down the node
-                    plumtree_peer_service_manager:reset_state(),
+                    plumtree_peer_service_manager:delete_state(),
                     stop("Leaving cluster");
                 _ ->
-                    leave()
+                    leave([])
             end;
         {error, singleton} ->
             lager:warning("Cannot leave, not a member of a cluster.")
-    end.
+    end;
+leave(_Args) ->
+    leave([]).
 
 stop() ->
     stop("received stop request").

src/plumtree_peer_service_console.erl

@@ -0,0 +1,35 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2015 Helium Systems, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you under the Apache License,
+%% Version 2.0 (the "License"); you may not use this file
+%% except in compliance with the License.  You may obtain
+%% a copy of the License at
+%%
+%%   http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% -------------------------------------------------------------------
+
+-module(plumtree_peer_service_console).
+
+-export([members/1]).
+
+members([]) ->
+    {ok, LocalState} = plumtree_peer_service_manager:get_local_state(),
+    Members = riak_dt_orswot:value(LocalState),
+    print_members(Members).
+
+print_members(Members) ->
+    _ = io:format("~29..=s Cluster Membership ~30..=s~n", ["",""]),
+    _ = io:format("Connected Nodes:~n~n", []),
+    _ = [io:format("~p~n", [Node]) || Node <- Members],
+    _ = io:format("~79..=s~n", [""]),
+    ok.

src/plumtree_peer_service_events.erl

@@ -65,8 +65,7 @@ update(LocalState) ->
 
 init([Fn]) ->
     {ok, LocalState} = plumtree_peer_service_manager:get_local_state(),
-    Members = riak_dt_orswot:value(LocalState),
-    Fn(Members),
+    Fn(LocalState),
     {ok, #state { callback = Fn }}.
 
 handle_event({update, LocalState}, State) ->

src/plumtree_peer_service_manager.erl

@@ -22,7 +22,7 @@
 
 -define(TBL, cluster_state).
 
--export([init/0, get_local_state/0, get_actor/0, update_state/1, reset_state/0]).
+-export([init/0, get_local_state/0, get_actor/0, update_state/1, delete_state/0]).
 
 init() ->
     %% setup ETS table for cluster_state
@@ -61,11 +61,8 @@ update_state(State) ->
     write_state_to_disk(State),
     ets:insert(?TBL, {cluster_state, State}).
 
-reset_state() ->
-    add_self(),
-    {ok, State} = get_local_state(),
-    write_state_to_disk(State),
-    ets:insert(?TBL, {cluster_state, State}).
+delete_state() ->
+    delete_state_from_disk().
 
 %%% ------------------------------------------------------------------
 %%% internal functions
@@ -106,6 +103,21 @@ write_state_to_disk(State) ->
                                  riak_dt_orswot:to_binary(State))
     end.
 
+delete_state_from_disk() ->
+    case data_root() of 
+        undefined ->
+            ok;
+        Dir ->
+            File = filename:join(Dir, "cluster_state"),
+            ok = filelib:ensure_dir(File),
+            case file:delete(File) of
+                ok ->
+                    lager:info("Leaving cluster, removed cluster_state");
+                {error, Reason} ->
+                    lager:info("Unable to remove cluster_state for reason ~p", [Reason])
+            end
+    end.
+
 maybe_load_state_from_disk() ->
     case data_root() of
         undefined ->

test/cluster_membership_SUITE.erl

@@ -35,6 +35,7 @@
     join_nonexistant_node_test/1,
     join_self_test/1,
     leave_test/1,
+    leave_rejoin_test/1,
     sticky_membership_test/1
         ]).
 
@@ -76,7 +77,7 @@ end_per_testcase(_, _Config) ->
 
 all() ->
     [singleton_test, join_test, join_nonexistant_node_test, join_self_test,
-    leave_test, sticky_membership_test].
+    leave_test, leave_rejoin_test, sticky_membership_test].
 
 singleton_test(Config) ->
     Nodes = proplists:get_value(nodes, Config),
@@ -118,7 +119,7 @@ leave_test(Config) ->
     [?assertEqual({Node, Expected}, {Node,
                                      lists:sort(plumtree_test_utils:get_cluster_members(Node))})
      || Node <- Nodes],
-    ?assertEqual(ok, rpc:call(Node1, plumtree_peer_service, leave, [])),
+    ?assertEqual(ok, rpc:call(Node1, plumtree_peer_service, leave, [[]])),
     Expected2 = lists:sort(OtherNodes),
     ok = plumtree_test_utils:wait_until_left(OtherNodes, Node1),
     %% should be a 3 node cluster now
@@ -129,6 +130,34 @@ leave_test(Config) ->
     ?assertEqual(pang, net_adm:ping(Node1)),
     ok.
 
+leave_rejoin_test(Config) ->
+    [Node1|OtherNodes] = Nodes = proplists:get_value(nodes, Config),
+    [Node2|_Rest] = OtherNodes,
+    [?assertEqual(ok, rpc:call(Node, plumtree_peer_service, join, [Node1]))
+     || Node <- OtherNodes],
+    Expected = lists:sort(Nodes),
+    ok = plumtree_test_utils:wait_until_joined(Nodes, Expected),
+    [?assertEqual({Node, Expected}, {Node,
+                                     lists:sort(plumtree_test_utils:get_cluster_members(Node))})
+     || Node <- Nodes],
+    ?assertEqual(ok, rpc:call(Node1, plumtree_peer_service, leave, [[]])),
+    Expected2 = lists:sort(OtherNodes),
+    ok = plumtree_test_utils:wait_until_left(OtherNodes, Node1),
+    %% should be a 3 node cluster now
+    [?assertEqual({Node, Expected2}, {Node,
+                                      lists:sort(plumtree_test_utils:get_cluster_members(Node))})
+     || Node <- OtherNodes],
+    %% node1 should be offline
+    ?assertEqual(pang, net_adm:ping(Node1)),
+    plumtree_test_utils:start_node(jaguar, Config, leave_rejoin_test),
+    %% rejoin cluster
+    ?assertEqual(ok, rpc:call(Node1, plumtree_peer_service, join, [Node2])),
+    ok = plumtree_test_utils:wait_until_joined(Nodes, Expected),
+    [?assertEqual({Node, Expected}, {Node, 
+                                     lists:sort(plumtree_test_utils:get_cluster_members(Node))})
+     || Node <- Nodes],
+    ok.
+
 sticky_membership_test(Config) ->
     [Node1|OtherNodes] = Nodes = proplists:get_value(nodes, Config),
     [?assertEqual(ok, rpc:call(Node, plumtree_peer_service, join, [Node1]))
@@ -150,7 +179,7 @@ sticky_membership_test(Config) ->
     ct_slave:stop(jaguar),
     ok = plumtree_test_utils:wait_until_offline(Node1),
     [Node2|LastTwo] = OtherNodes,
-    ?assertEqual(ok, rpc:call(Node2, plumtree_peer_service, leave, [])),
+    ?assertEqual(ok, rpc:call(Node2, plumtree_peer_service, leave, [[]])),
     ok = plumtree_test_utils:wait_until_left(LastTwo, Node2),
     ok = plumtree_test_utils:wait_until_offline(Node2),
     Expected2 = lists:sort(Nodes -- [Node2]),