Introduce new master activity events for ...

1. Signifying start of a stage of rebalance.
2. Signifying completion of a stage of rebalance.
3. Signifying a notable event that does not cause rebalance to fail,
   during a particular stage.

A stage of rebalance can be a nested stage, for example, delta
recovery is a sub stage of kv rebalance stage.

Helps with,
MB-27463: Per-node details for certain stages
MB-12000: Delta recovery information

Part of EPIC,
MB-30894: Rebalance visibility and reporting

Change-Id: I28966768ecf3a3f6d832f131a414629a5b425005
Reviewed-on: http://review.couchbase.org/101816
Reviewed-by: Aliaksey Artamonau <[email protected]>
Tested-by: Abhijeeth Nuthan <[email protected]>

Author: anuthan

src/master_activity_events.erl

@@ -59,14 +59,31 @@
          note_set_service_map/2,
          note_autofailover_node_state_change/4,
          note_autofailover_server_group_state_change/4,
-         note_autofailover_done/2
+         note_autofailover_done/2,
+         note_rebalance_stage_started/1,
+         note_rebalance_stage_completed/1,
+         note_rebalance_stage_event/2
         ]).
 
 -export([stream_events/2]).
 
 submit_cast(Arg) ->
     (catch gen_event:notify(master_activity_events_ingress, {submit_master_event, Arg})).
 
+get_stage_list(Stage) when is_atom(Stage) ->
+    [Stage];
+get_stage_list(Stage) when is_list(Stage) ->
+    Stage.
+
+note_rebalance_stage_started(Stage) ->
+    submit_cast({rebalance_stage_started, get_stage_list(Stage)}).
+
+note_rebalance_stage_completed(Stage) ->
+    submit_cast({rebalance_stage_completed, get_stage_list(Stage)}).
+
+note_rebalance_stage_event(Stage, Text) ->
+    submit_cast({rebalance_stage_event, get_stage_list(Stage), Text}).
+
 note_vbucket_state_change(Bucket, Node, VBucketId, NewState) ->
     submit_cast({vbucket_state_change, Bucket, Node, VBucketId, NewState}).
 
@@ -375,6 +392,22 @@ maybe_get_pids_node(Pid) when is_pid(Pid) ->
 maybe_get_pids_node(_PerhapsBinary) ->
     skip_this_pair_please.
 
+event_to_jsons({TS, rebalance_stage_started, Stage}) ->
+    [format_simple_plist_as_json([{type, rebalanceStageStarted},
+                                  {ts, misc:time_to_epoch_float(TS)},
+                                  {stage, {list, Stage}}])];
+
+event_to_jsons({TS, rebalance_stage_completed, Stage}) ->
+    [format_simple_plist_as_json([{type, rebalanceStageCompleted},
+                                  {ts, misc:time_to_epoch_float(TS)},
+                                  {stage, {list, Stage}}])];
+
+event_to_jsons({TS, rebalance_stage_event, Stage, Text}) ->
+    [format_simple_plist_as_json([{type, rebalanceStageEvent},
+                                  {ts, misc:time_to_epoch_float(TS)},
+                                  {stage, {list, Stage}},
+                                  {event, Text}])];
+
 event_to_jsons({TS, vbucket_state_change, Bucket, Node, VBucketId, NewState}) ->
     [format_simple_plist_as_json([{type, vbucketStateChange},
                                   {ts, misc:time_to_epoch_float(TS)},

src/ns_rebalancer.erl

@@ -540,9 +540,13 @@ rebalance_simple_services(Config, Services, KeepNodes) ->
         true ->
             lists:filtermap(
               fun (Service) ->
+                      master_activity_events:note_rebalance_stage_started(
+                        Service),
                       ServiceNodes = ns_cluster_membership:service_nodes(KeepNodes, Service),
                       Updated = update_service_map_with_config(Config, Service, ServiceNodes),
 
+                      master_activity_events:note_rebalance_stage_completed(
+                        Service),
                       case Updated of
                           false ->
                               false;
@@ -596,10 +600,14 @@ rebalance_topology_aware_services(Config, Services, KeepNodesAll, EjectNodesAll)
                   [] ->
                       false;
                   _ ->
+                      master_activity_events:note_rebalance_stage_started(
+                        Service),
                       update_service_map_with_config(Config, Service, AllNodes),
                       ok = rebalance_topology_aware_service(Service, KeepNodes,
                                                             EjectNodes, DeltaNodes),
                       update_service_map(Service, AllNodes, KeepNodes),
+                      master_activity_events:note_rebalance_stage_completed(
+                        Service),
                       {true, {Service, os:timestamp()}}
               end
       end, Services).
@@ -708,6 +716,7 @@ rebalance_body(KeepNodes,
 
     ok = drop_old_2i_indexes(KeepNodes),
 
+    master_activity_events:note_rebalance_stage_started(kv),
     %% wait till all bucket shutdowns are done on nodes we're
     %% adding (or maybe adding).
     do_wait_buckets_shutdown(KeepNodes),
@@ -721,9 +730,13 @@ rebalance_body(KeepNodes,
                           run_janitor_pre_rebalance(Bucket)
                   end, BucketConfigs),
 
+    master_activity_events:note_rebalance_stage_started(
+      [kv, kv_delta_recovery]),
     ok = apply_delta_recovery_buckets(DeltaRecoveryBuckets,
                                       KVDeltaNodes, BucketConfigs),
     ok = maybe_clear_recovery_type(KeepNodes),
+    master_activity_events:note_rebalance_stage_completed(
+      [kv, kv_delta_recovery]),
     ok = service_janitor:cleanup(),
 
     ok = leader_activities:activate_quorum_nodes(KeepNodes),
@@ -738,6 +751,7 @@ rebalance_body(KeepNodes,
     ok = check_test_condition(rebalance_cluster_nodes_active),
 
     rebalance_kv(KeepNodes, EjectNodesAll, BucketConfigs, DeltaRecoveryBuckets),
+    master_activity_events:note_rebalance_stage_completed(kv),
     rebalance_services(KeepNodes, EjectNodesAll),
 
     ok = leader_activities:deactivate_quorum_nodes(EjectNodesAll),
@@ -1357,10 +1371,13 @@ do_run_graceful_failover_moves(Nodes, BucketName, BucketConfig, I, N) ->
     Map = proplists:get_value(map, BucketConfig, []),
     Map1 = mb_map:promote_replicas_for_graceful_failover(Map, Nodes),
 
+    master_activity_events:note_rebalance_stage_started(kv),
     ProgressFun = make_progress_fun(I, N),
-    run_mover(BucketName, BucketConfig,
-              proplists:get_value(servers, BucketConfig),
-              ProgressFun, Map, Map1).
+    RV = run_mover(BucketName, BucketConfig,
+                   proplists:get_value(servers, BucketConfig),
+                   ProgressFun, Map, Map1),
+    master_activity_events:note_rebalance_stage_completed(kv),
+    RV.
 
 check_graceful_failover_possible(Nodes, BucketsAll) ->
     %% No graceful failovers for non KV node

src/ns_vbucket_mover.erl

@@ -294,6 +294,12 @@ spawn_compaction_uninhibitor(Bucket, Node, MRef) ->
                               master_activity_events:note_compaction_uninhibit_done(Bucket, Node),
                               ok;
                           nack ->
+                              Msg = io_lib:format(
+                                      "failed to initiate compaction for "
+                                      "bucket ~p on node ~p",
+                                      [Bucket, Node]),
+                              master_activity_events:note_rebalance_stage_event(
+                                kv, Msg),
                               erlang:exit({failed_to_initiate_compaction, Bucket, Node, MRef})
                       end;
                   _ ->