Add convergence visibility

helix-core/src/main/java/org/apache/helix/PropertyKey.java

@@ -26,6 +26,7 @@
 import org.apache.helix.model.ClusterConfig;
 import org.apache.helix.model.ClusterConstraints;
 import org.apache.helix.model.ClusterStatus;
+import org.apache.helix.model.ConvergenceStatus;
 import org.apache.helix.model.ControllerHistory;
 import org.apache.helix.model.CurrentState;
 import org.apache.helix.model.CustomizedState;
@@ -54,6 +55,7 @@
 import org.slf4j.LoggerFactory;
 
 import static org.apache.helix.PropertyType.CONFIGS;
+import static org.apache.helix.PropertyType.CONVERGENCESTATUS;
 import static org.apache.helix.PropertyType.CONTROLLER;
 import static org.apache.helix.PropertyType.CURRENTSTATES;
 import static org.apache.helix.PropertyType.CUSTOMIZEDSTATES;
@@ -755,6 +757,23 @@ public PropertyKey targetExternalView(String resourceName) {
       return new PropertyKey(TARGETEXTERNALVIEW, ExternalView.class, _clusterName, resourceName);
     }
 
+    /**
+     * Get the cluster convergence status root.
+     * @return {@link PropertyKey}
+     */
+    public PropertyKey convergenceStatus() {
+      return new PropertyKey(CONVERGENCESTATUS, ConvergenceStatus.class, _clusterName);
+    }
+
+    /**
+     * Get the convergence status of a resource.
+     * @param resourceName resource name
+     * @return {@link PropertyKey}
+     */
+    public PropertyKey convergenceStatus(String resourceName) {
+      return new PropertyKey(CONVERGENCESTATUS, ConvergenceStatus.class, _clusterName, resourceName);
+    }
+
     /**
      * Get a property key associated with a controller
      * @return {@link PropertyKey}

helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java

@@ -26,6 +26,7 @@
 import java.util.regex.Pattern;
 
 import org.apache.helix.model.ClusterStatus;
+import org.apache.helix.model.ConvergenceStatus;
 import org.apache.helix.model.ControllerHistory;
 import org.apache.helix.model.CurrentState;
 import org.apache.helix.model.CustomizedView;
@@ -68,6 +69,7 @@ public class PropertyPathBuilder {
     typeToClassMapping.put(PropertyType.PAUSE, PauseSignal.class);
     typeToClassMapping.put(PropertyType.MAINTENANCE, MaintenanceSignal.class);
     typeToClassMapping.put(PropertyType.STATUS, ClusterStatus.class);
+    typeToClassMapping.put(PropertyType.CONVERGENCESTATUS, ConvergenceStatus.class);
     // TODO: Below must handle the case for future versions of Task Framework with a different path
     // structure
     typeToClassMapping.put(PropertyType.WORKFLOWCONTEXT, WorkflowContext.class);
@@ -94,6 +96,10 @@ public class PropertyPathBuilder {
     addEntry(PropertyType.TARGETEXTERNALVIEW, 1, "/{clusterName}/TARGETEXTERNALVIEW");
     addEntry(PropertyType.TARGETEXTERNALVIEW, 2,
         "/{clusterName}/TARGETEXTERNALVIEW/{resourceName}");
+    addEntry(PropertyType.CONVERGENCESTATUS, 1,
+        "/{clusterName}/PROPERTYSTORE/HELIX_CONVERGENCE_STATUS");
+    addEntry(PropertyType.CONVERGENCESTATUS, 2,
+        "/{clusterName}/PROPERTYSTORE/HELIX_CONVERGENCE_STATUS/{resourceName}");
     addEntry(PropertyType.CUSTOMIZEDVIEW, 1, "/{clusterName}/CUSTOMIZEDVIEW");
     addEntry(PropertyType.CUSTOMIZEDVIEW, 2, "/{clusterName}/CUSTOMIZEDVIEW/{resourceName}");
     addEntry(PropertyType.CUSTOMIZEDVIEW, 3,

helix-core/src/main/java/org/apache/helix/PropertyType.java

@@ -45,6 +45,7 @@ public enum PropertyType {
   EXTERNALVIEW(Type.CLUSTER, true, false),
   CUSTOMIZEDVIEW(Type.CLUSTER, true, false),
   TARGETEXTERNALVIEW(Type.CLUSTER, true, false),
+  CONVERGENCESTATUS(Type.CLUSTER, true, false),
   STATEMODELDEFS(Type.CLUSTER, true, false, false, false, true),
   CONTROLLER(Type.CLUSTER, true, false),
   PROPERTYSTORE(Type.CLUSTER, true, false),

helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java

@@ -76,6 +76,8 @@
 import org.apache.helix.controller.stages.ClusterEvent;
 import org.apache.helix.controller.stages.ClusterEventType;
 import org.apache.helix.controller.stages.CompatibilityCheckStage;
+import org.apache.helix.controller.stages.ConvergenceStatusPersistenceCache;
+import org.apache.helix.controller.stages.ConvergenceStatusPersistStage;
 import org.apache.helix.controller.stages.CurrentStateComputationStage;
 import org.apache.helix.controller.stages.CustomizedStateComputationStage;
 import org.apache.helix.controller.stages.CustomizedViewAggregationStage;
@@ -173,6 +175,8 @@ public class GenericHelixController implements IdealStateChangeListener, LiveIns
   private final ClusterEventProcessor _managementModeEventThread;
 
   private final Map<AsyncWorkerType, DedupEventProcessor<String, Runnable>> _asyncFIFOWorkerPool;
+  private final ConvergenceStatusPersistenceCache _convergenceStatusPersistenceCache =
+      new ConvergenceStatusPersistenceCache();
 
   private long _continuousRebalanceFailureCount = 0;
   private long _continuousResourceRebalanceFailureCount = 0;
@@ -540,6 +544,7 @@ private static PipelineRegistry createDefaultRegistry(String pipelineName) {
       rebalancePipeline.addStage(new PersistAssignmentStage());
       rebalancePipeline.addStage(new TargetExteralViewCalcStage());
       rebalancePipeline.addStage(new ParticipantDeregistrationStage());
+      rebalancePipeline.addStage(new ConvergenceStatusPersistStage());
 
       // external view generation
       Pipeline externalViewPipeline = new Pipeline(pipelineName);
@@ -666,6 +671,7 @@ private static PipelineRegistry createManagementModeRegistry(String pipelineName
       managementMode.addStage(new ManagementModeStage());
       managementMode.addStage(new ManagementMessageGenerationPhase());
       managementMode.addStage(new ManagementMessageDispatchStage());
+      managementMode.addStage(new ConvergenceStatusPersistStage());
 
       PipelineRegistry registry = new PipelineRegistry();
       Arrays.asList(
@@ -813,6 +819,11 @@ private void handleEvent(ClusterEvent event, BaseControllerDataProvider dataProv
     // regains leadership.
     event.addAttribute(AttributeName.STATEFUL_REBALANCER.name(),
         _rebalancerRef.getRebalancer(manager));
+    ClusterConfig convergenceClusterConfig =
+        _resourceControlDataProvider == null ? null : _resourceControlDataProvider.getClusterConfig();
+    event.addAttribute(AttributeName.CONVERGENCE_MONITORING_ENABLED.name(),
+        convergenceClusterConfig != null
+            && convergenceClusterConfig.isConvergenceMonitoringEnabled());
 
     Optional<String> eventSessionId = Optional.empty();
     // We should expect only events in tests don't have it.
@@ -1321,6 +1332,8 @@ private void pushToEventQueues(ClusterEventType eventType, NotificationContext c
     event.addAttribute(AttributeName.helixmanager.name(), changeContext.getManager());
     event.addAttribute(AttributeName.changeContext.name(), changeContext);
     event.addAttribute(AttributeName.AsyncFIFOWorkerPool.name(), _asyncFIFOWorkerPool);
+    event.addAttribute(AttributeName.CONVERGENCE_STATUS_PERSISTENCE_CACHE.name(),
+        _convergenceStatusPersistenceCache);
     for (Map.Entry<String, Object> attr : eventAttributes.entrySet()) {
       event.addAttribute(attr.getKey(), attr.getValue());
     }

helix-core/src/main/java/org/apache/helix/controller/pipeline/AsyncWorkerType.java

@@ -34,5 +34,6 @@ public enum AsyncWorkerType {
   MaintenanceRecoveryWorker,
   TaskJobPurgeWorker,
   CustomizedStateViewComputeWorker,
-  ParticipantDeregistrationWorker
+  ParticipantDeregistrationWorker,
+  ConvergenceStatusPersistWorker
 }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalanceStatus.java

@@ -0,0 +1,57 @@
+package org.apache.helix.controller.rebalancer.waged;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * 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.
+ */
+
+import org.apache.helix.HelixRebalanceException;
+
+/**
+ * Immutable health metadata for the WAGED assignment currently served by the controller.
+ */
+public final class WagedRebalanceStatus {
+  private final boolean _lastKnownGoodFallback;
+  private final boolean _servingComputationFailed;
+  private final boolean _baselineComputationFailed;
+  private final HelixRebalanceException.FailureCategory _servingFailureCategory;
+
+  public WagedRebalanceStatus(boolean lastKnownGoodFallback, boolean servingComputationFailed,
+      boolean baselineComputationFailed,
+      HelixRebalanceException.FailureCategory servingFailureCategory) {
+    _lastKnownGoodFallback = lastKnownGoodFallback;
+    _servingComputationFailed = servingComputationFailed;
+    _baselineComputationFailed = baselineComputationFailed;
+    _servingFailureCategory = servingFailureCategory;
+  }
+
+  public boolean isLastKnownGoodFallback() {
+    return _lastKnownGoodFallback;
+  }
+
+  public boolean isServingComputationFailed() {
+    return _servingComputationFailed;
+  }
+
+  public boolean isBaselineComputationFailed() {
+    return _baselineComputationFailed;
+  }
+
+  public HelixRebalanceException.FailureCategory getServingFailureCategory() {
+    return _servingFailureCategory;
+  }
+}

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java

@@ -30,6 +30,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 import com.google.common.collect.ImmutableMap;
@@ -60,6 +61,7 @@
 import org.apache.helix.monitoring.metrics.WagedRebalancerMetricCollector;
 import org.apache.helix.monitoring.metrics.model.CountMetric;
 import org.apache.helix.monitoring.metrics.model.LatencyMetric;
+import org.apache.helix.util.RebalanceUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -113,6 +115,12 @@ public class WagedRebalancer implements StatefulRebalancer<ResourceControllerDat
   // dashboards see the same signal. May be null when WagedRebalancer is used outside the
   // pipeline (e.g. ReadOnlyWagedRebalancer for the REST partitionAssignment API).
   private volatile ClusterStatusMonitor _clusterStatusMonitor;
+  private final AtomicBoolean _servingComputationFailed = new AtomicBoolean(false);
+  private final AtomicBoolean _baselineComputationFailed = new AtomicBoolean(false);
+  private final AtomicReference<HelixRebalanceException.FailureCategory>
+      _servingFailureCategory = new AtomicReference<>();
+  private volatile boolean _lastRunUsedFallback;
+  private volatile HelixRebalanceException.FailureCategory _lastRunFailureCategory;
 
   private static AssignmentMetadataStore constructAssignmentStore(String metadataStoreAddrs,
       String clusterName) {
@@ -366,6 +374,10 @@ void reportHardConstraintBlockingSnapshot(ClusterModel.RebalanceScopeType scope,
   void reportAsyncFailure(HelixRebalanceException ex) {
     _rebalanceFailureCount.increment(1L);
     reportFailureCategory(ex);
+    _servingFailureCategory.set(ex.getFailureCategory());
+    if (_servingComputationFailed.compareAndSet(false, true)) {
+      scheduleConvergenceStatusRefresh();
+    }
   }
 
   /**
@@ -392,9 +404,14 @@ void reportBaselineAsyncFailure(HelixRebalanceException ex) {
    * GLOBAL_BASELINE phase, so it is reversible regardless of async mode. Null-tolerant.
    */
   void reportBaselineComputeStatus(boolean clean) {
+    boolean failed = !clean;
+    boolean statusChanged = _baselineComputationFailed.getAndSet(failed) != failed;
     ClusterStatusMonitor monitor = _clusterStatusMonitor;
     if (monitor != null) {
-      monitor.updateWagedBaselineComputeFailing(!clean);
+      monitor.updateWagedBaselineComputeFailing(failed);
+    }
+    if (statusChanged) {
+      scheduleConvergenceStatusRefresh();
     }
   }
 
@@ -421,10 +438,15 @@ void reportOverwriteComputeStatus(boolean clean) {
    * Null-tolerant.
    */
   void reportPartialRebalanceSuccess() {
+    boolean recovered = _servingComputationFailed.compareAndSet(true, false);
+    _servingFailureCategory.set(null);
     ClusterStatusMonitor monitor = _clusterStatusMonitor;
     if (monitor != null) {
       monitor.resetWagedFailureRollupGauges();
     }
+    if (recovered) {
+      scheduleConvergenceStatusRefresh();
+    }
   }
 
   // Update the global rebalance mode to be asynchronous or synchronous
@@ -487,11 +509,13 @@ public Map<String, IdealState> computeNewIdealStates(ResourceControllerDataProvi
 
     Map<String, IdealState> newIdealStates;
     boolean usedFallback = false;
+    _lastRunFailureCategory = null;
     try {
       // Calculate the target assignment based on the current cluster status.
       newIdealStates = computeBestPossibleStates(clusterData, resourceMap, currentStateOutput,
           _rebalanceAlgorithm);
     } catch (HelixRebalanceException ex) {
+      _lastRunFailureCategory = ex.getFailureCategory();
       LOG.error("Failed to calculate the new assignments. category={} customerActionable={}",
           ex.getFailureCategory(), ex.isCustomerActionable(), ex);
       // Record the failure in metrics.
@@ -531,6 +555,7 @@ public Map<String, IdealState> computeNewIdealStates(ResourceControllerDataProvi
     if (monitor != null) {
       monitor.setWagedFallbackInUseGauge(usedFallback);
     }
+    _lastRunUsedFallback = usedFallback;
 
     // Construct the new best possible states according to the current state and target assignment.
     // Note that the new ideal state might be an intermediate state between the current state and
@@ -557,6 +582,19 @@ public Map<String, IdealState> computeNewIdealStates(ResourceControllerDataProvi
     return newIdealStates;
   }
 
+  public WagedRebalanceStatus getConvergenceStatus() {
+    HelixRebalanceException.FailureCategory category =
+        _lastRunUsedFallback ? _lastRunFailureCategory : _servingFailureCategory.get();
+    return new WagedRebalanceStatus(_lastRunUsedFallback, _servingComputationFailed.get(),
+        _baselineComputationFailed.get(), category);
+  }
+
+  private void scheduleConvergenceStatusRefresh() {
+    if (_manager != null) {
+      RebalanceUtil.scheduleOnDemandPipeline(_manager.getClusterName(), 0L, false);
+    }
+  }
+
   // Coordinate global rebalance and partial rebalance according to the cluster changes.
   private Map<String, IdealState> computeBestPossibleStates(
       ResourceControllerDataProvider clusterData, Map<String, Resource> resourceMap,

helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java

@@ -30,6 +30,11 @@ public enum AttributeName {
   MESSAGES_ALL,
   MESSAGES_SELECTED,
   MESSAGES_THROTTLE,
+  MESSAGE_DISPATCH_RESULT,
+  CONVERGENCE_STATUS,
+  CONVERGENCE_STATUS_CONTEXT,
+  CONVERGENCE_STATUS_PERSISTENCE_CACHE,
+  CONVERGENCE_MONITORING_ENABLED,
   LOCAL_STATE,
   EVENT_CREATE_TIME,
   helixmanager,