Move rebalance progress from ns_orchestrator ...

anuthan · anuthan · commit 6a85a17d290c · 2019-03-19T02:18:53.000Z
... to ns_rebalance_observer, so that we can have all rebalance related tracking in one place. Change-Id: Icc2a4ad219770ba794b46c904924c10a1552b907 Reviewed-on: http://review.couchbase.org/101815 Reviewed-by: Aliaksey Artamonau <aliaksey.artamonau@couchbase.com> Tested-by: Abhijeeth Nuthan <abhijeeth.nuthan@couchbase.com> Well-Formed: Build Bot <build@couchbase.com>
diff --git a/src/ns_orchestrator.erl b/src/ns_orchestrator.erl
@@ -25,7 +25,7 @@
 -record(janitor_state, {cleanup_id = undefined :: undefined | pid()}).
 
 -record(rebalancing_state, {rebalancer,
-                            progress,
+                            rebalance_observer = undefined,
                             keep_nodes,
                             eject_nodes,
                             failed_nodes,
@@ -53,7 +53,6 @@
          start_rebalance/3,
          start_rebalance/4,
          stop_rebalance/0,
-         update_progress/2,
          is_rebalance_running/0,
          start_recovery/1,
          stop_recovery/2,
@@ -226,7 +225,20 @@ rebalance_progress_full() ->
                                      {running, [{atom(), float()}]} |
                                      not_running.
 rebalance_progress_full(Timeout) ->
-    gen_statem:call(?SERVER, rebalance_progress, Timeout).
+    case ns_config:search(rebalancer_pid) of
+        false ->
+            not_running;
+        {value, undefined} ->
+            not_running;
+        {value, Pid} when is_pid(Pid) ->
+            case ns_rebalance_observer:get_aggregated_progress(Timeout) of
+                not_running ->
+                    ?log_error("Couldn't reach ns_rebalance_observer"),
+                    not_running;
+                Aggr ->
+                    {running, Aggr}
+            end
+    end.
 
 -spec rebalance_progress() -> {running, [{atom(), float()}]} | not_running.
 rebalance_progress() ->
@@ -463,6 +475,10 @@ handle_info({'EXIT', Pid, Reason}, rebalancing,
             #rebalancing_state{rebalancer = Pid} = State) ->
     handle_rebalance_completion(Reason, State);
 
+handle_info({'EXIT', ObserverPid, Reason}, rebalancing,
+            #rebalancing_state{rebalance_observer = ObserverPid} = State) ->
+    {keep_state, stop_rebalance(State, {rebalance_observer_terminated, Reason})};
+
 handle_info({'EXIT', Pid, Reason}, recovery, #recovery_state{pid = Pid}) ->
     ale:error(?USER_LOGGER,
               "Recovery process ~p terminated unexpectedly: ~p", [Pid, Reason]),
@@ -671,6 +687,15 @@ idle({start_graceful_failover, Node}, From, _State) when is_atom(Node) ->
     {keep_state_and_data,
      [{next_event, {call, From}, {start_graceful_failover, [Node]}}]};
 idle({start_graceful_failover, Nodes}, From, _State) ->
+    ActiveNodes = ns_cluster_membership:active_nodes(),
+    NodesInfo = [{active_nodes, ActiveNodes}],
+    Services = [kv],
+    Type = graceful_failover,
+    {ok, ObserverPid} = ns_rebalance_observer:start_link(
+                          Services,
+                          NodesInfo,
+                          Type),
+
     case ns_rebalancer:start_link_graceful_failover(Nodes) of
         {ok, Pid} ->
             Id = couch_uuids:random(),
@@ -681,27 +706,36 @@ idle({start_graceful_failover, Nodes}, From, _State) ->
             ns_cluster:counter_inc(Type, start),
             set_rebalance_status(Type, running, Pid),
 
-            ActiveNodes = ns_cluster_membership:active_nodes(),
-            Progress = rebalance_progress:init(ActiveNodes, [kv]),
-
             {next_state, rebalancing,
              #rebalancing_state{rebalancer = Pid,
+                                rebalance_observer = ObserverPid,
                                 eject_nodes = [],
                                 keep_nodes = [],
                                 failed_nodes = [],
                                 abort_reason = undefined,
-                                progress = Progress,
                                 type = Type,
                                 rebalance_id = Id},
              [{reply, From, ok}]};
         {error, RV} ->
+            misc:unlink_terminate_and_wait(ObserverPid, kill),
             {keep_state_and_data, [{reply, From, RV}]}
     end;
 idle(rebalance_progress, From, _State) ->
     {keep_state_and_data, [{reply, From, not_running}]};
 %% NOTE: this is not remotely called but is used by maybe_start_rebalance
 idle({start_rebalance, KeepNodes, EjectNodes, FailedNodes, DeltaNodes,
       DeltaRecoveryBuckets, RebalanceId}, From, _State) ->
+    NodesInfo = [{active_nodes, KeepNodes ++ EjectNodes},
+                 {keep_nodes, KeepNodes},
+                 {eject_nodes, EjectNodes},
+                 {delta_nodes, DeltaNodes},
+                 {failed_nodes, FailedNodes}],
+    Type = rebalance,
+    Services = [kv] ++ ns_cluster_membership:topology_aware_services(),
+    {ok, ObserverPid} = ns_rebalance_observer:start_link(
+                          Services,
+                          NodesInfo,
+                          Type),
     case ns_rebalancer:start_link_rebalance(KeepNodes, EjectNodes,
                                             FailedNodes, DeltaNodes,
                                             DeltaRecoveryBuckets) of
@@ -725,14 +759,12 @@ idle({start_rebalance, KeepNodes, EjectNodes, FailedNodes, DeltaNodes,
                              [KeepNodes, EjectNodes, FailedNodes, RebalanceId])
             end,
 
-            Type = rebalance,
             ns_cluster:counter_inc(Type, start),
             set_rebalance_status(Type, running, Pid),
 
             {next_state, rebalancing,
              #rebalancing_state{rebalancer = Pid,
-                                progress = rebalance_progress:init(
-                                             KeepNodes ++ EjectNodes),
+                                rebalance_observer = ObserverPid,
                                 keep_nodes = KeepNodes,
                                 eject_nodes = EjectNodes,
                                 failed_nodes = FailedNodes,
@@ -741,11 +773,22 @@ idle({start_rebalance, KeepNodes, EjectNodes, FailedNodes, DeltaNodes,
                                 rebalance_id = RebalanceId},
              [{reply, From, ok}]};
         {error, no_kv_nodes_left} ->
+            misc:unlink_terminate_and_wait(ObserverPid, kill),
             {keep_state_and_data, [{reply, From, no_kv_nodes_left}]};
         {error, delta_recovery_not_possible} ->
+            misc:unlink_terminate_and_wait(ObserverPid, kill),
             {keep_state_and_data, [{reply, From, delta_recovery_not_possible}]}
     end;
 idle({move_vbuckets, Bucket, Moves}, From, _State) ->
+    KeepNodes = ns_node_disco:nodes_wanted(),
+    Type = move_vbuckets,
+    NodesInfo = [{active_nodes, ns_cluster_membership:active_nodes()},
+                 {keep_nodes, KeepNodes}],
+    Services = [kv],
+    {ok, ObserverPid} = ns_rebalance_observer:start_link(
+                          Services,
+                          NodesInfo,
+                          Type),
     Pid = spawn_link(
             fun () ->
                     ns_rebalancer:move_vbuckets(Bucket, Moves)
@@ -754,16 +797,12 @@ idle({move_vbuckets, Bucket, Moves}, From, _State) ->
     Id = couch_uuids:random(),
     ?log_debug("Moving vBuckets in bucket ~p. Moves ~p. "
                "Operation Id = ~s", [Bucket, Moves, Id]),
-    Type = move_vbuckets,
     ns_cluster:counter_inc(Type, start),
     set_rebalance_status(Type, running, Pid),
 
-    Nodes = ns_cluster_membership:active_nodes(),
-    Progress = rebalance_progress:init(Nodes, [kv]),
-
     {next_state, rebalancing,
      #rebalancing_state{rebalancer = Pid,
-                        progress = Progress,
+                        rebalance_observer = ObserverPid,
                         keep_nodes = ns_node_disco:nodes_wanted(),
                         eject_nodes = [],
                         failed_nodes = [],
@@ -821,11 +860,6 @@ janitor_running(Msg, From, #janitor_state{cleanup_id = ID})
     {next_state, idle, #idle_state{}, [{next_event, {call, From}, Msg}]}.
 
 %% Asynchronous rebalancing events
-rebalancing({update_progress, Service, ServiceProgress},
-            #rebalancing_state{progress = Old} = State) ->
-    NewProgress = rebalance_progress:update(Service, ServiceProgress, Old),
-    {next_state, rebalancing,
-     State#rebalancing_state{progress = NewProgress}};
 rebalancing({timeout, _Tref, stop_timeout},
             #rebalancing_state{rebalancer = Pid} = State) ->
     ?log_debug("Stop rebalance timeout, brutal kill pid = ~p", [Pid]),
@@ -870,11 +904,12 @@ rebalancing(stop_rebalance, From,
             #rebalancing_state{rebalancer = Pid} = State) ->
     ?log_debug("Sending stop to rebalancer: ~p", [Pid]),
     {keep_state, stop_rebalance(State, user_stop), [{reply, From, ok}]};
-rebalancing(rebalance_progress, From,
-            #rebalancing_state{progress = Progress}) ->
-    AggregatedProgress =
-        dict:to_list(rebalance_progress:get_progress(Progress)),
-    {keep_state_and_data, [{reply, From, {running, AggregatedProgress}}]};
+rebalancing(rebalance_progress, From, _State) ->
+    %% Only expect this call if we are pre-madhatter.
+    false = cluster_compat_mode:is_cluster_madhatter(),
+    {keep_state_and_data,
+     [{reply, From,
+       {running, ns_rebalance_observer:get_aggregated_progress(2000)}}]};
 rebalancing(Event, From, _State) ->
     ?log_warning("Got event ~p while rebalancing.", [Event]),
     {keep_state_and_data, [{reply, From, rebalance_running}]}.
@@ -944,10 +979,6 @@ do_request_janitor_run(Item, Fun, FsmState, State) ->
     end,
     {next_state, FsmState, State}.
 
--spec update_progress(service(), dict:dict()) -> ok.
-update_progress(Service, ServiceProgress) ->
-    gen_statem:cast(?SERVER, {update_progress, Service, ServiceProgress}).
-
 wait_for_nodes_loop(Nodes) ->
     receive
         {done, Node} ->
@@ -1142,8 +1173,14 @@ rebalance_completed_next_state({try_autofailover, From, Nodes}) ->
 rebalance_completed_next_state(_) ->
     {next_state, idle, #idle_state{}}.
 
+terminate_observer(#rebalancing_state{rebalance_observer = undefined}) ->
+    ok;
+terminate_observer(#rebalancing_state{rebalance_observer = ObserverPid}) ->
+    misc:unlink_terminate_and_wait(ObserverPid, kill).
+
 handle_rebalance_completion(ExitReason, State) ->
     cancel_stop_timer(State),
+    terminate_observer(State),
     maybe_reset_autofailover_count(ExitReason, State),
     maybe_reset_reprovision_count(ExitReason, State),
     log_rebalance_completion(ExitReason, State),
@@ -1214,6 +1251,11 @@ log_abort_reason({try_autofailover, _, Nodes}, Type, Id) ->
              "~s interrupted due to auto-failover of nodes ~p. "
              "Operation Id = ~s",
              [rebalance_type2text(Type), Nodes, Id]);
+log_abort_reason({rebalance_observer_terminated, Reason}, Type, Id) ->
+    ale:error(?USER_LOGGER,
+              "~s interrupted as observer exited with reason ~p. "
+              "Operation Id = ~s",
+              [rebalance_type2text(Type), Reason, Id]);
 log_abort_reason(user_stop, Type, Id) ->
     ale:info(?USER_LOGGER,
              "~s stopped by user. Operation Id = ~s",
@@ -1272,15 +1314,19 @@ maybe_start_service_upgrader(normal, {changed, OldVersion, NewVersion},
             ale:info(?USER_LOGGER,
                      "Starting upgrade for the following services: ~p",
                      [Services]),
+            Type = service_upgrade,
+            NodesInfo = [{active_nodes, KeepNodes},
+                         {keep_nodes, KeepNodes}],
+            {ok, ObserverPid} = ns_rebalance_observer:start_link(
+                                  Services,
+                                  NodesInfo,
+                                  Type),
             Pid = start_service_upgrader(KeepNodes, Services),
 
-            Type = service_upgrade,
             set_rebalance_status(Type, running, Pid),
             ns_cluster:counter_inc(Type, start),
-            Progress = rebalance_progress:init(KeepNodes, Services),
-
             NewState = State#rebalancing_state{type = Type,
-                                               progress = Progress,
+                                               rebalance_observer = ObserverPid,
                                                rebalancer = Pid},
 
             {started, NewState}
diff --git a/src/ns_rebalance_observer.erl b/src/ns_rebalance_observer.erl
@@ -20,7 +20,10 @@
 
 -include("ns_common.hrl").
 
--export([start_link/1, get_detailed_progress/0]).
+-export([start_link/3,
+         get_detailed_progress/0,
+         get_aggregated_progress/1,
+         update_progress/2]).
 
 %% gen_server callbacks
 -export([code_change/3, init/1, handle_call/3, handle_cast/2, handle_info/2,
@@ -41,22 +44,36 @@
 -record(state, {bucket :: bucket_name() | undefined,
                 buckets_count :: pos_integer(),
                 bucket_number :: non_neg_integer(),
+                progress :: rebalance_progress:progress(),
+                nodes_info :: [{atom(), [node()]}],
+                type :: atom(),
                 done_moves :: [#move_state{}],
                 current_moves :: [#move_state{}],
                 pending_moves :: [#move_state{}]
                }).
 
-start_link(BucketsCount) ->
-    gen_server:start_link(?SERVER, ?MODULE, BucketsCount, []).
+start_link(Services, NodesInfo, Type) ->
+    gen_server:start_link(?SERVER, ?MODULE, {Services, NodesInfo, Type}, []).
 
-get_detailed_progress() ->
+generic_get_call(Call) ->
+    generic_get_call(Call, 10000).
+generic_get_call(Call, Timeout) ->
     try
-        gen_server:call(?SERVER, get_detailed_progress, 10000)
+        gen_server:call(?SERVER, Call, Timeout)
     catch
         exit:_Reason ->
             not_running
     end.
 
+get_detailed_progress() ->
+    generic_get_call(get_detailed_progress).
+
+get_aggregated_progress(Timeout) ->
+    generic_get_call(get_aggregated_progress, Timeout).
+
+update_progress(Service, ServiceProgress) ->
+    gen_server:cast(?SERVER, {update_progress, Service, ServiceProgress}).
+
 is_interesting_master_event({_, bucket_rebalance_started, _Bucket, _Pid}) ->
     fun handle_bucket_rebalance_started/2;
 is_interesting_master_event({_, set_ff_map, _BucketName, _Diff}) ->
@@ -68,8 +85,7 @@ is_interesting_master_event({_, vbucket_move_done, _BucketName, _VBucketId}) ->
 is_interesting_master_event(_) ->
     undefined.
 
-
-init(BucketsCount) ->
+init({Services, NodesInfo, Type}) ->
     Self = self(),
     ns_pubsub:subscribe_link(master_activity_events,
                              fun (Event, _Ignored) ->
@@ -81,11 +97,17 @@ init(BucketsCount) ->
                                      end
                              end, []),
 
+    {active_nodes, ActiveNodes} = lists:keyfind(active_nodes, 1, NodesInfo),
+    Progress = rebalance_progress:init(ActiveNodes, Services),
+    BucketsCount = length(ns_bucket:get_buckets()),
     proc_lib:spawn_link(erlang, apply, [fun docs_left_updater_init/1, [Self]]),
 
     {ok, #state{bucket = undefined,
                 buckets_count = BucketsCount,
                 bucket_number = 0,
+                progress = Progress,
+                nodes_info = NodesInfo,
+                type = Type,
                 done_moves  = [],
                 current_moves = [],
                 pending_moves = []}}.
@@ -94,6 +116,9 @@ handle_call(get, _From, State) ->
     {reply, State, State};
 handle_call(get_detailed_progress, _From, State) ->
     {reply, do_get_detailed_progress(State), State};
+handle_call(get_aggregated_progress, _From,
+            #state{progress = Progress} = State) ->
+    {reply, dict:to_list(rebalance_progress:get_progress(Progress)), State};
 handle_call(Req, From, State) ->
     ?log_error("Got unknown request: ~p from ~p", [Req, From]),
     {reply, unknown_request, State}.
@@ -147,6 +172,11 @@ handle_cast({update_stats, VBucket, NodeToDocsLeft}, State) ->
                         Move#move_state{stats = NewStats}
                 end)};
 
+handle_cast({update_progress, Service, ServiceProgress},
+            #state{progress = Old} = State) ->
+    NewProgress = rebalance_progress:update(Service, ServiceProgress, Old),
+    {noreply, State#state{progress = NewProgress}};
+
 handle_cast(Req, _State) ->
     ?log_error("Got unknown cast: ~p", [Req]),
     erlang:error({unknown_cast, Req}).
diff --git a/src/ns_rebalancer.erl b/src/ns_rebalancer.erl
@@ -607,7 +607,7 @@ rebalance_topology_aware_services(Config, Services, KeepNodesAll, EjectNodesAll)
 rebalance_topology_aware_service(Service, KeepNodes, EjectNodes, DeltaNodes) ->
     ProgressCallback =
         fun (Progress) ->
-                ns_orchestrator:update_progress(Service, Progress)
+                ns_rebalance_observer:update_progress(Service, Progress)
         end,
 
     misc:with_trap_exit(
@@ -758,7 +758,7 @@ make_progress_fun(BucketCompletion, NumBuckets) ->
     end.
 
 update_kv_progress(Progress) ->
-    ns_orchestrator:update_progress(kv, Progress).
+    ns_rebalance_observer:update_progress(kv, Progress).
 
 update_kv_progress(Nodes, Progress) ->
     update_kv_progress(dict:from_list([{N, Progress} || N <- Nodes])).
@@ -777,8 +777,6 @@ rebalance_kv(KeepNodes, EjectNodes, BucketConfigs, DeltaRecoveryBuckets) ->
             exit(Error)
     end,
 
-    {ok, RebalanceObserver} = ns_rebalance_observer:start_link(length(BucketConfigs)),
-
     lists:foreach(fun ({I, {BucketName, BucketConfig}}) ->
                           BucketCompletion = I / NumBuckets,
                           update_kv_progress(LiveKVNodes, BucketCompletion),
@@ -788,8 +786,7 @@ rebalance_kv(KeepNodes, EjectNodes, BucketConfigs, DeltaRecoveryBuckets) ->
                                            KeepKVNodes, EjectNodes, DeltaRecoveryBuckets)
                   end, misc:enumerate(BucketConfigs, 0)),
 
-    update_kv_progress(LiveKVNodes, 1.0),
-    misc:unlink_terminate_and_wait(RebalanceObserver, shutdown).
+    update_kv_progress(LiveKVNodes, 1.0).
 
 rebalance_bucket(BucketName, BucketConfig, ProgressFun,
                  KeepKVNodes, EjectNodes, DeltaRecoveryBuckets) ->
diff --git a/src/rebalance_progress.erl b/src/rebalance_progress.erl
