diff --git a/helix-core/src/main/java/org/apache/helix/PropertyKey.java b/helix-core/src/main/java/org/apache/helix/PropertyKey.java index 1037c94e24..de3b735560 100644 --- a/helix-core/src/main/java/org/apache/helix/PropertyKey.java +++ b/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} diff --git a/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java b/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java index f0002bdbde..1701744c10 100644 --- a/helix-core/src/main/java/org/apache/helix/PropertyPathBuilder.java +++ b/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, diff --git a/helix-core/src/main/java/org/apache/helix/PropertyType.java b/helix-core/src/main/java/org/apache/helix/PropertyType.java index 474ea05a85..a7fd11b18a 100644 --- a/helix-core/src/main/java/org/apache/helix/PropertyType.java +++ b/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), diff --git a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java b/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java index 564a0391d2..4dc606a872 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/GenericHelixController.java +++ b/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> _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 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 attr : eventAttributes.entrySet()) { event.addAttribute(attr.getKey(), attr.getValue()); } diff --git a/helix-core/src/main/java/org/apache/helix/controller/pipeline/AsyncWorkerType.java b/helix-core/src/main/java/org/apache/helix/controller/pipeline/AsyncWorkerType.java index ecbe7eb0c4..2681a799f1 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/pipeline/AsyncWorkerType.java +++ b/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 } diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalanceStatus.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalanceStatus.java new file mode 100644 index 0000000000..6103fe0740 --- /dev/null +++ b/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; + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java b/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java index 03cb8d97d7..a93a58f7f7 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/WagedRebalancer.java +++ b/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 + _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 computeNewIdealStates(ResourceControllerDataProvi Map 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 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 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 computeBestPossibleStates( ResourceControllerDataProvider clusterData, Map resourceMap, diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java b/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java index 0db5252ee0..aafcb20342 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/stages/AttributeName.java +++ b/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, diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusCalculator.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusCalculator.java new file mode 100644 index 0000000000..198e13346d --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusCalculator.java @@ -0,0 +1,686 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.helix.HelixDefinedState; +import org.apache.helix.HelixManager; +import org.apache.helix.HelixRebalanceException; +import org.apache.helix.controller.common.PartitionStateMap; +import org.apache.helix.controller.dataproviders.BaseControllerDataProvider; +import org.apache.helix.controller.rebalancer.waged.WagedRebalanceStatus; +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.model.ConvergenceStatus.OptimizerStatus; +import org.apache.helix.model.ConvergenceStatus.PartitionDetail; +import org.apache.helix.model.ConvergenceStatus.Reason; +import org.apache.helix.model.ConvergenceStatus.Scope; +import org.apache.helix.model.ConvergenceStatus.Status; +import org.apache.helix.model.ConvergenceStatus.TargetFreshness; +import org.apache.helix.model.Message; +import org.apache.helix.model.Partition; +import org.apache.helix.model.Resource; +import org.apache.helix.task.TaskConstants; + +/** + * Computes a bounded convergence snapshot from one controller event. + */ +public class ConvergenceStatusCalculator { + private static final Comparator DETAIL_ORDER = + Comparator.comparingInt((PartitionEvaluation evaluation) -> + statusSeverity(evaluation._status)) + .reversed() + .thenComparing(evaluation -> evaluation._resourceName) + .thenComparing(evaluation -> evaluation._partitionName); + + public ConvergenceStatusSnapshot calculate(ClusterEvent event) { + BaseControllerDataProvider cache = + event.getAttribute(AttributeName.ControllerDataProvider.name()); + @SuppressWarnings("unchecked") + Map resources = + event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name()); + if (cache == null || resources == null) { + throw new IllegalArgumentException( + "Convergence calculation requires controller cache and resources"); + } + return calculate(event, ConvergenceStatusContext.from(event, cache, resources)); + } + + public ConvergenceStatusSnapshot calculate(ClusterEvent event, + ConvergenceStatusContext context) { + @SuppressWarnings("unchecked") + Map resources = + event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name()); + CurrentStateOutput currentStateOutput = + event.getAttribute(AttributeName.CURRENT_STATE.name()); + HelixManager manager = event.getAttribute(AttributeName.helixmanager.name()); + + if (context == null || resources == null || currentStateOutput == null || manager == null) { + throw new IllegalArgumentException( + "Convergence calculation requires context, resources, current state, and manager"); + } + + if (context.isMaintenanceModeEnabled()) { + return calculatePaused(event, resources, currentStateOutput, Reason.MAINTENANCE_MODE); + } + + BestPossibleStateOutput bestPossibleStateOutput = + event.getAttribute(AttributeName.BEST_POSSIBLE_STATE.name()); + MessageOutput allMessages = event.getAttribute(AttributeName.MESSAGES_ALL.name()); + MessageOutput selectedMessages = event.getAttribute(AttributeName.MESSAGES_SELECTED.name()); + MessageOutput throttledMessages = event.getAttribute(AttributeName.MESSAGES_THROTTLE.name()); + MessageDispatchResult dispatchResult = + event.getAttribute(AttributeName.MESSAGE_DISPATCH_RESULT.name()); + + DispatchIndex dispatchIndex = new DispatchIndex(dispatchResult); + WagedRebalanceStatus wagedStatus = context.getWagedStatus(); + Map resourceStatuses = new LinkedHashMap<>(); + List clusterDetails = new ArrayList<>(); + Aggregate clusterAggregate = new Aggregate(); + + List resourceNames = new ArrayList<>(resources.keySet()); + Collections.sort(resourceNames); + for (String resourceName : resourceNames) { + Resource resource = resources.get(resourceName); + if (TaskConstants.STATE_MODEL_NAME.equals(resource.getStateModelDefRef())) { + continue; + } + + ResourceEvaluation evaluation = + evaluateResource(resource, context, currentStateOutput, bestPossibleStateOutput, allMessages, + selectedMessages, throttledMessages, dispatchIndex); + TargetFreshness freshness = TargetFreshness.CURRENT; + OptimizerStatus optimizerStatus = OptimizerStatus.NOT_APPLICABLE; + if (context.isWagedResource(resourceName)) { + optimizerStatus = wagedStatus != null && wagedStatus.isBaselineComputationFailed() + ? OptimizerStatus.BASELINE_FAILED : OptimizerStatus.HEALTHY; + if (wagedStatus == null) { + freshness = TargetFreshness.UNKNOWN; + evaluation = overrideUnknown(evaluation, Reason.WAGED_INTERNAL_FAILURE); + optimizerStatus = OptimizerStatus.UNKNOWN; + } else if (wagedStatus.isLastKnownGoodFallback()) { + freshness = TargetFreshness.LAST_KNOWN_GOOD; + evaluation = overrideUnknown(evaluation, + wagedFailureReason(wagedStatus, Reason.WAGED_LAST_KNOWN_GOOD)); + } else if (wagedStatus.isServingComputationFailed()) { + freshness = TargetFreshness.UNKNOWN; + evaluation = overrideUnknown(evaluation, + wagedFailureReason(wagedStatus, Reason.WAGED_INTERNAL_FAILURE)); + } + } + ConvergenceStatus status = + buildResourceStatus(event, manager, resourceName, evaluation, freshness, optimizerStatus); + resourceStatuses.put(resourceName, status); + clusterAggregate.add(evaluation._aggregate); + evaluation._details.forEach(detail -> addBoundedDetail(clusterDetails, detail)); + } + + ConvergenceStatus clusterStatus = + buildClusterStatus(event, manager, clusterAggregate, resourceStatuses, clusterDetails); + return new ConvergenceStatusSnapshot(clusterStatus, resourceStatuses); + } + + public ConvergenceStatusSnapshot calculatePaused(ClusterEvent event, + Map resources, CurrentStateOutput currentStateOutput, Reason reason) { + return calculateOverride(event, resources, currentStateOutput, Status.PAUSED, reason); + } + + public ConvergenceStatusSnapshot calculateUnknown(ClusterEvent event, + Map resources, CurrentStateOutput currentStateOutput, Reason reason) { + return calculateOverride(event, resources, currentStateOutput, Status.UNKNOWN, reason); + } + + private ConvergenceStatusSnapshot calculateOverride(ClusterEvent event, + Map resources, CurrentStateOutput currentStateOutput, Status overrideStatus, + Reason reason) { + HelixManager manager = event.getAttribute(AttributeName.helixmanager.name()); + if (manager == null) { + throw new IllegalArgumentException("Convergence override calculation requires HelixManager"); + } + + Map resourceStatuses = new LinkedHashMap<>(); + Aggregate clusterAggregate = new Aggregate(); + List resourceNames = new ArrayList<>(resources.keySet()); + Collections.sort(resourceNames); + for (String resourceName : resourceNames) { + Resource resource = resources.get(resourceName); + if (TaskConstants.STATE_MODEL_NAME.equals(resource.getStateModelDefRef())) { + continue; + } + Set partitions = new HashSet<>(resource.getPartitions()); + partitions.addAll(currentStateOutput.getCurrentStateMap(resourceName).keySet()); + Aggregate aggregate = new Aggregate(); + for (int i = 0; i < partitions.size(); i++) { + aggregate.record(overrideStatus, reason); + } + ResourceEvaluation evaluation = + new ResourceEvaluation(overrideStatus, reason, aggregate, Collections.emptyList()); + resourceStatuses.put(resourceName, + buildResourceStatus(event, manager, resourceName, evaluation, TargetFreshness.UNKNOWN, + OptimizerStatus.UNKNOWN)); + clusterAggregate.add(aggregate); + } + + ConvergenceStatus clusterStatus = + buildClusterStatus(event, manager, clusterAggregate, resourceStatuses, + Collections.emptyList()); + clusterStatus.setStatus(overrideStatus); + clusterStatus.setPrimaryReason(reason); + clusterStatus.setTargetFreshness(TargetFreshness.UNKNOWN); + return new ConvergenceStatusSnapshot(clusterStatus, resourceStatuses); + } + + private ResourceEvaluation evaluateResource(Resource resource, + ConvergenceStatusContext context, CurrentStateOutput currentStateOutput, + BestPossibleStateOutput bestPossibleStateOutput, MessageOutput allMessages, + MessageOutput selectedMessages, MessageOutput throttledMessages, + DispatchIndex dispatchIndex) { + String resourceName = resource.getResourceName(); + Aggregate aggregate = new Aggregate(); + List details = new ArrayList<>(); + + Set partitions = new HashSet<>(); + partitions.addAll(resource.getPartitions()); + partitions.addAll(currentStateOutput.getCurrentStateMap(resourceName).keySet()); + + boolean resourceTargetAvailable = + bestPossibleStateOutput != null && bestPossibleStateOutput.resourceSet() + .contains(resourceName); + PartitionStateMap targetStateMap = resourceTargetAvailable + ? bestPossibleStateOutput.getPartitionStateMap(resourceName) : null; + if (targetStateMap != null) { + partitions.addAll(targetStateMap.partitionSet()); + } + + for (Partition partition : partitions) { + Map current = + copy(currentStateOutput.getCurrentStateMap(resourceName, partition)); + boolean partitionTargetAvailable = + resourceTargetAvailable && targetStateMap.partitionSet().contains(partition); + Map target = partitionTargetAvailable + ? copy(targetStateMap.getPartitionMap(partition)) : Collections.emptyMap(); + + PartitionEvaluation evaluation; + if (!context.hasStateModel(resourceName) || !partitionTargetAvailable) { + evaluation = new PartitionEvaluation(resourceName, partition.getPartitionName(), + Status.UNKNOWN, Reason.TARGET_ASSIGNMENT_MISSING, current, target); + } else { + evaluation = + evaluatePartition(resourceName, partition, current, target, + context.getInitialState(resourceName), context, currentStateOutput, allMessages, + selectedMessages, throttledMessages, + dispatchIndex); + } + aggregate.record(evaluation._status, evaluation._reason); + if (evaluation._status != Status.CONVERGED) { + addBoundedDetail(details, evaluation); + } + } + + Status status = aggregate.overallStatus(); + Reason reason = aggregate.primaryReason(status); + return new ResourceEvaluation(status, reason, aggregate, details); + } + + private PartitionEvaluation evaluatePartition(String resourceName, Partition partition, + Map current, Map target, + String initialState, ConvergenceStatusContext context, + CurrentStateOutput currentStateOutput, MessageOutput allMessages, + MessageOutput selectedMessages, MessageOutput throttledMessages, + DispatchIndex dispatchIndex) { + NormalizedAssignment normalizedCurrent = + normalize(current, initialState); + NormalizedAssignment normalizedTarget = + normalize(target, initialState); + + if (!normalizedCurrent._valid || !normalizedTarget._valid) { + return new PartitionEvaluation(resourceName, partition.getPartitionName(), Status.UNKNOWN, + Reason.INVALID_STATE, current, target); + } + + Map pending = + currentStateOutput.getPendingMessageMap(resourceName, partition); + Map cancellations = + currentStateOutput.getCancellationMessageMap(resourceName, partition); + Map relays = + currentStateOutput.getPendingRelayMessageMap(resourceName, partition); + boolean hasActiveMessage = + !pending.isEmpty() || !cancellations.isEmpty() || !relays.isEmpty(); + + if (normalizedCurrent._assignment.equals(normalizedTarget._assignment) + && !hasActiveMessage) { + return new PartitionEvaluation(resourceName, partition.getPartitionName(), Status.CONVERGED, + Reason.NONE, current, target); + } + + String partitionName = partition.getPartitionName(); + Reason progressReason = null; + if (!cancellations.isEmpty()) { + progressReason = Reason.CANCELLATION_PENDING; + } else if (!relays.isEmpty()) { + progressReason = Reason.RELAY_PENDING; + } else if (!pending.isEmpty()) { + progressReason = Reason.PENDING_TRANSITION; + } else if (dispatchIndex.wasSent(resourceName, partitionName)) { + progressReason = Reason.TRANSITION_DISPATCHED; + } else if (dispatchIndex.failed(resourceName, partitionName)) { + return new PartitionEvaluation(resourceName, partitionName, Status.BLOCKED, + Reason.MESSAGE_DISPATCH_FAILED, current, target); + } else { + boolean generated = hasMessages(allMessages, resourceName, partition); + boolean selected = hasMessages(selectedMessages, resourceName, partition); + boolean throttled = hasMessages(throttledMessages, resourceName, partition); + if (selected && !throttled) { + progressReason = Reason.MESSAGE_THROTTLED; + } else if (generated && !selected) { + progressReason = Reason.STATE_CONSTRAINT_WAIT; + } else if (throttled) { + progressReason = Reason.TRANSITION_DISPATCHED; + } + } + + if (progressReason != null) { + return new PartitionEvaluation(resourceName, partitionName, Status.IN_PROGRESS, + progressReason, current, target); + } + + if (containsState(current, HelixDefinedState.ERROR.name())) { + return new PartitionEvaluation(resourceName, partitionName, Status.BLOCKED, + Reason.ERROR_STATE, current, target); + } + + Set unavailableTargetInstances = new HashSet<>(normalizedTarget._assignment.keySet()); + unavailableTargetInstances.removeAll(context.getLiveInstances()); + if (!unavailableTargetInstances.isEmpty()) { + if (context.isDelayedResource(resourceName)) { + return new PartitionEvaluation(resourceName, partitionName, Status.IN_PROGRESS, + Reason.WAITING_FOR_DELAY, current, target); + } + return new PartitionEvaluation(resourceName, partitionName, Status.BLOCKED, + Reason.TARGET_INSTANCE_NOT_LIVE, current, target); + } + + return new PartitionEvaluation(resourceName, partitionName, Status.BLOCKED, + Reason.NO_PROGRESS_PATH, current, target); + } + + private ConvergenceStatus buildResourceStatus(ClusterEvent event, HelixManager manager, + String resourceName, ResourceEvaluation evaluation, TargetFreshness freshness, + OptimizerStatus optimizerStatus) { + ConvergenceStatus status = new ConvergenceStatus(resourceName); + populateCommon(status, event, manager); + status.setScope(Scope.RESOURCE); + status.setResourceName(resourceName); + status.setStatus(evaluation._status); + status.setPrimaryReason(evaluation._reason); + status.setTargetFreshness(freshness); + status.setOptimizerStatus(optimizerStatus); + populateAggregate(status, evaluation._aggregate); + List details = new ArrayList<>(); + evaluation._details.forEach(detail -> details.add(detail.toDetail())); + status.setPartitionDetails(details, ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS); + status.setTruncatedPartitionCount( + Math.max(0, status.getAffectedPartitionCount() - details.size())); + return status; + } + + private ConvergenceStatus buildClusterStatus(ClusterEvent event, HelixManager manager, + Aggregate aggregate, Map resourceStatuses, + List details) { + ConvergenceStatus status = new ConvergenceStatus(manager.getClusterName()); + populateCommon(status, event, manager); + status.setScope(Scope.CLUSTER); + status.setStatus(overallResourceStatus(resourceStatuses.values())); + status.setPrimaryReason(primaryResourceReason(status.getStatus(), resourceStatuses.values())); + status.setTargetFreshness(clusterFreshness(resourceStatuses.values())); + status.setOptimizerStatus(clusterOptimizerStatus(resourceStatuses.values())); + status.setTotalResourceCount(resourceStatuses.size()); + populateAggregate(status, aggregate); + List partitionDetails = new ArrayList<>(); + details.forEach(detail -> partitionDetails.add(detail.toDetail())); + status.setPartitionDetails(partitionDetails, ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS); + status.setTruncatedPartitionCount( + Math.max(0, status.getAffectedPartitionCount() - partitionDetails.size())); + return status; + } + + private void populateCommon(ConvergenceStatus status, ClusterEvent event, HelixManager manager) { + status.setGeneratedAt(System.currentTimeMillis()); + status.setControllerSessionId(manager.getSessionId()); + status.setSourceEventId(event.getEventId()); + status.setComplete(true); + } + + private void populateAggregate(ConvergenceStatus status, Aggregate aggregate) { + status.setTotalPartitionCount(aggregate._total); + status.setConvergedPartitionCount(aggregate.count(Status.CONVERGED)); + status.setInProgressPartitionCount(aggregate.count(Status.IN_PROGRESS)); + status.setBlockedPartitionCount(aggregate.count(Status.BLOCKED)); + status.setUnknownPartitionCount(aggregate.count(Status.UNKNOWN)); + status.setAffectedPartitionCount( + aggregate._total - aggregate.count(Status.CONVERGED) - aggregate.count(Status.PAUSED)); + status.setStatusCounts(aggregate._statusCounts); + status.setReasonCounts(aggregate._reasonCounts); + } + + private static Status overallResourceStatus(Collection statuses) { + Status result = Status.CONVERGED; + for (ConvergenceStatus status : statuses) { + if (statusSeverity(status.getStatus()) > statusSeverity(result)) { + result = status.getStatus(); + } + } + return result; + } + + private static Reason primaryResourceReason(Status status, + Collection statuses) { + return statuses.stream().filter(value -> value.getStatus() == status) + .map(ConvergenceStatus::getPrimaryReason) + .min(Comparator.comparingInt(ConvergenceStatusCalculator::reasonPriority)) + .orElse(Reason.NONE); + } + + private static TargetFreshness clusterFreshness(Collection statuses) { + if (statuses.stream() + .anyMatch(status -> status.getTargetFreshness() == TargetFreshness.LAST_KNOWN_GOOD)) { + return TargetFreshness.LAST_KNOWN_GOOD; + } + if (statuses.stream() + .anyMatch(status -> status.getTargetFreshness() == TargetFreshness.UNKNOWN)) { + return TargetFreshness.UNKNOWN; + } + return TargetFreshness.CURRENT; + } + + private static OptimizerStatus clusterOptimizerStatus( + Collection statuses) { + if (statuses.stream() + .anyMatch(status -> status.getOptimizerStatus() == OptimizerStatus.BASELINE_FAILED)) { + return OptimizerStatus.BASELINE_FAILED; + } + if (statuses.stream() + .anyMatch(status -> status.getOptimizerStatus() == OptimizerStatus.UNKNOWN)) { + return OptimizerStatus.UNKNOWN; + } + if (statuses.stream() + .anyMatch(status -> status.getOptimizerStatus() == OptimizerStatus.HEALTHY)) { + return OptimizerStatus.HEALTHY; + } + return OptimizerStatus.NOT_APPLICABLE; + } + + private static ResourceEvaluation overrideUnknown(ResourceEvaluation evaluation, Reason reason) { + Aggregate aggregate = new Aggregate(); + for (int i = 0; i < evaluation._aggregate._total; i++) { + aggregate.record(Status.UNKNOWN, reason); + } + List details = new ArrayList<>(); + evaluation._details.forEach(detail -> details.add( + new PartitionEvaluation(detail._resourceName, detail._partitionName, Status.UNKNOWN, + reason, detail._current, detail._target))); + return new ResourceEvaluation(Status.UNKNOWN, reason, aggregate, details); + } + + private static Reason wagedFailureReason(WagedRebalanceStatus status, Reason defaultReason) { + HelixRebalanceException.FailureCategory category = status.getServingFailureCategory(); + if (category == null) { + return defaultReason; + } + switch (category) { + case CAPACITY_DEFICIT: + return Reason.WAGED_CAPACITY_DEFICIT; + case NO_CANDIDATE_NODE: + return Reason.WAGED_NO_CANDIDATE_NODE; + case INVALID_RESOURCE_CONFIG: + case INVALID_CLUSTER_CONFIG: + return Reason.WAGED_INVALID_CONFIGURATION; + case METADATA_STORE_IO: + case ALGORITHM_INTERNAL: + case ASYNC_EXECUTION: + case UNKNOWN: + default: + return Reason.WAGED_INTERNAL_FAILURE; + } + } + + private static boolean hasMessages(MessageOutput output, String resourceName, + Partition partition) { + return output != null && !output.getMessages(resourceName, partition).isEmpty(); + } + + private static NormalizedAssignment normalize(Map assignment, + String initialState) { + Map result = new HashMap<>(); + for (Map.Entry entry : assignment.entrySet()) { + String instance = entry.getKey(); + String state = entry.getValue(); + if (instance == null || instance.isEmpty() || state == null || state.isEmpty()) { + return new NormalizedAssignment(Collections.emptyMap(), false); + } + if (state.equalsIgnoreCase(HelixDefinedState.DROPPED.name()) + || initialState != null && state.equalsIgnoreCase(initialState)) { + continue; + } + result.put(instance, state); + } + return new NormalizedAssignment(Collections.unmodifiableMap(result), true); + } + + private static boolean containsState(Map assignment, String state) { + return assignment.values().stream() + .anyMatch(value -> value != null && value.equalsIgnoreCase(state)); + } + + private static Map copy(Map input) { + return input == null || input.isEmpty() ? Collections.emptyMap() : new HashMap<>(input); + } + + private static void addBoundedDetail(List details, + PartitionEvaluation evaluation) { + details.add(evaluation); + details.sort(DETAIL_ORDER); + if (details.size() > ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS) { + details.remove(details.size() - 1); + } + } + + private static int statusSeverity(Status status) { + switch (status) { + case PAUSED: + return 5; + case BLOCKED: + return 4; + case UNKNOWN: + return 3; + case IN_PROGRESS: + return 2; + case CONVERGED: + default: + return 1; + } + } + + private static int reasonPriority(Reason reason) { + switch (reason) { + case MANAGEMENT_MODE: + case MAINTENANCE_MODE: + return 1; + case WAGED_LAST_KNOWN_GOOD: + case WAGED_CAPACITY_DEFICIT: + case WAGED_NO_CANDIDATE_NODE: + case WAGED_INVALID_CONFIGURATION: + case WAGED_INTERNAL_FAILURE: + return 2; + case TARGET_ASSIGNMENT_MISSING: + case INVALID_STATE: + return 3; + case ERROR_STATE: + case MESSAGE_DISPATCH_FAILED: + case NO_PROGRESS_PATH: + return 4; + case TARGET_INSTANCE_NOT_LIVE: + return 5; + case MESSAGE_THROTTLED: + case STATE_CONSTRAINT_WAIT: + return 6; + case PENDING_TRANSITION: + case CANCELLATION_PENDING: + case RELAY_PENDING: + case TRANSITION_DISPATCHED: + default: + return 7; + } + } + + private static final class NormalizedAssignment { + private final Map _assignment; + private final boolean _valid; + + private NormalizedAssignment(Map assignment, boolean valid) { + _assignment = assignment; + _valid = valid; + } + } + + private static final class PartitionEvaluation { + private final String _resourceName; + private final String _partitionName; + private final Status _status; + private final Reason _reason; + private final Map _current; + private final Map _target; + + private PartitionEvaluation(String resourceName, String partitionName, Status status, + Reason reason, Map current, Map target) { + _resourceName = resourceName; + _partitionName = partitionName; + _status = status; + _reason = reason; + _current = current; + _target = target; + } + + private PartitionDetail toDetail() { + return new PartitionDetail(_resourceName, _partitionName, _status, _reason, _current, + _target); + } + } + + private static final class ResourceEvaluation { + private final Status _status; + private final Reason _reason; + private final Aggregate _aggregate; + private final List _details; + + private ResourceEvaluation(Status status, Reason reason, Aggregate aggregate, + List details) { + _status = status; + _reason = reason; + _aggregate = aggregate; + _details = details; + } + } + + private static final class Aggregate { + private final Map _statusCounts = new EnumMap<>(Status.class); + private final Map _reasonCounts = new EnumMap<>(Reason.class); + private int _total; + + private void record(Status status, Reason reason) { + _total++; + _statusCounts.merge(status, 1, Integer::sum); + if (reason != Reason.NONE) { + _reasonCounts.merge(reason, 1, Integer::sum); + } + } + + private void add(Aggregate other) { + _total += other._total; + other._statusCounts.forEach((key, value) -> _statusCounts.merge(key, value, Integer::sum)); + other._reasonCounts.forEach((key, value) -> _reasonCounts.merge(key, value, Integer::sum)); + } + + private int count(Status status) { + return _statusCounts.getOrDefault(status, 0); + } + + private Status overallStatus() { + return _statusCounts.keySet().stream() + .max(Comparator.comparingInt(ConvergenceStatusCalculator::statusSeverity)) + .orElse(Status.CONVERGED); + } + + private Reason primaryReason(Status status) { + if (status == Status.CONVERGED) { + return Reason.NONE; + } + return _reasonCounts.keySet().stream() + .min(Comparator.comparingInt(ConvergenceStatusCalculator::reasonPriority)) + .orElse(Reason.NONE); + } + } + + private static final class DispatchIndex { + private final Set _sent = new HashSet<>(); + private final Set _failed = new HashSet<>(); + + private DispatchIndex(MessageDispatchResult result) { + if (result != null) { + index(result.getSentMessages(), _sent); + index(result.getFailedMessages(), _failed); + } + } + + private boolean wasSent(String resourceName, String partitionName) { + return _sent.contains(key(resourceName, partitionName)); + } + + private boolean failed(String resourceName, String partitionName) { + return _failed.contains(key(resourceName, partitionName)); + } + + private static void index(List messages, Set output) { + for (Message message : messages) { + List partitionNames = message.getPartitionNames(); + if (partitionNames.isEmpty() && message.getPartitionName() != null) { + output.add(key(message.getResourceName(), message.getPartitionName())); + } else { + partitionNames.forEach( + partitionName -> output.add(key(message.getResourceName(), partitionName))); + } + } + } + + private static String key(String resourceName, String partitionName) { + return resourceName + '\u0000' + partitionName; + } + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusContext.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusContext.java new file mode 100644 index 0000000000..4dbc206f5d --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusContext.java @@ -0,0 +1,117 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import org.apache.helix.controller.dataproviders.BaseControllerDataProvider; +import org.apache.helix.controller.rebalancer.util.DelayedRebalanceUtil; +import org.apache.helix.controller.rebalancer.util.WagedValidationUtil; +import org.apache.helix.controller.rebalancer.waged.WagedRebalanceStatus; +import org.apache.helix.controller.rebalancer.waged.WagedRebalancer; +import org.apache.helix.model.IdealState; +import org.apache.helix.model.Resource; +import org.apache.helix.model.StateModelDefinition; + +/** + * Immutable metadata needed to calculate convergence away from the controller pipeline thread. + */ +public final class ConvergenceStatusContext { + private final boolean _maintenanceModeEnabled; + private final Map _initialStateByResource; + private final Set _liveInstances; + private final Set _wagedResources; + private final Set _delayedResources; + private final WagedRebalanceStatus _wagedStatus; + + private ConvergenceStatusContext(boolean maintenanceModeEnabled, + Map initialStateByResource, Set liveInstances, + Set wagedResources, Set delayedResources, + WagedRebalanceStatus wagedStatus) { + _maintenanceModeEnabled = maintenanceModeEnabled; + _initialStateByResource = + Collections.unmodifiableMap(new HashMap<>(initialStateByResource)); + _liveInstances = Collections.unmodifiableSet(new HashSet<>(liveInstances)); + _wagedResources = Collections.unmodifiableSet(new HashSet<>(wagedResources)); + _delayedResources = Collections.unmodifiableSet(new HashSet<>(delayedResources)); + _wagedStatus = wagedStatus; + } + + public static ConvergenceStatusContext from(ClusterEvent event, + BaseControllerDataProvider cache, Map resources) { + Map initialStates = new HashMap<>(); + Set wagedResources = new HashSet<>(); + Set delayedResources = new HashSet<>(); + for (Map.Entry entry : resources.entrySet()) { + String resourceName = entry.getKey(); + StateModelDefinition stateModelDefinition = + cache.getStateModelDef(entry.getValue().getStateModelDefRef()); + if (stateModelDefinition != null) { + initialStates.put(resourceName, stateModelDefinition.getInitialState()); + } + IdealState idealState = cache.getIdealState(resourceName); + if (idealState != null && WagedValidationUtil.isWagedEnabled(idealState)) { + wagedResources.add(resourceName); + } + if (idealState != null && cache.getClusterConfig() != null && DelayedRebalanceUtil + .isDelayRebalanceEnabled(idealState, cache.getClusterConfig())) { + delayedResources.add(resourceName); + } + } + + Object rebalancer = event.getAttribute(AttributeName.STATEFUL_REBALANCER.name()); + WagedRebalanceStatus wagedStatus = rebalancer instanceof WagedRebalancer + ? ((WagedRebalancer) rebalancer).getConvergenceStatus() : null; + return new ConvergenceStatusContext(cache.isMaintenanceModeEnabled(), initialStates, + cache.getLiveInstances().keySet(), wagedResources, delayedResources, wagedStatus); + } + + public boolean isMaintenanceModeEnabled() { + return _maintenanceModeEnabled; + } + + public String getInitialState(String resourceName) { + return _initialStateByResource.get(resourceName); + } + + public boolean hasStateModel(String resourceName) { + return _initialStateByResource.containsKey(resourceName); + } + + public Set getLiveInstances() { + return _liveInstances; + } + + public boolean isWagedResource(String resourceName) { + return _wagedResources.contains(resourceName); + } + + public boolean isDelayedResource(String resourceName) { + return _delayedResources.contains(resourceName); + } + + public WagedRebalanceStatus getWagedStatus() { + return _wagedStatus; + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusPersistStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusPersistStage.java new file mode 100644 index 0000000000..e48b582e5f --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusPersistStage.java @@ -0,0 +1,266 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import org.apache.helix.HelixDataAccessor; +import org.apache.helix.HelixManager; +import org.apache.helix.PropertyKey; +import org.apache.helix.controller.dataproviders.BaseControllerDataProvider; +import org.apache.helix.controller.pipeline.AbstractAsyncBaseStage; +import org.apache.helix.controller.pipeline.AsyncWorkerType; +import org.apache.helix.controller.pipeline.Pipeline; +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.model.ConvergenceStatus.Reason; +import org.apache.helix.model.ConvergenceStatus.Status; +import org.apache.helix.model.Resource; +import org.apache.helix.zookeeper.datamodel.ZNRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Persists convergence snapshots without blocking the reconciliation pipeline. + */ +public class ConvergenceStatusPersistStage extends AbstractAsyncBaseStage { + private static final Logger LOG = LoggerFactory.getLogger(ConvergenceStatusPersistStage.class); + private static final long MINIMUM_PERSIST_INTERVAL_MS = 1_000L; + private final ConvergenceStatusCalculator _calculator = new ConvergenceStatusCalculator(); + private final ConvergenceStatusPersistenceCache _localPersistenceCache = + new ConvergenceStatusPersistenceCache(); + + @Override + public void process(ClusterEvent event) throws Exception { + BaseControllerDataProvider cache = + event.getAttribute(AttributeName.ControllerDataProvider.name()); + @SuppressWarnings("unchecked") + Map resources = + event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name()); + Boolean monitoringEnabled = + event.getAttribute(AttributeName.CONVERGENCE_MONITORING_ENABLED.name()); + if (monitoringEnabled == null) { + monitoringEnabled = cache != null && cache.getClusterConfig() != null + && cache.getClusterConfig().isConvergenceMonitoringEnabled(); + } + if (cache == null || resources == null || !monitoringEnabled) { + return; + } + event.addAttribute(AttributeName.CONVERGENCE_STATUS_CONTEXT.name(), + ConvergenceStatusContext.from(event, cache, resources)); + super.process(event); + } + + @Override + public AsyncWorkerType getAsyncWorkerType() { + return AsyncWorkerType.ConvergenceStatusPersistWorker; + } + + @Override + public void execute(ClusterEvent event) { + ConvergenceStatusSnapshot snapshot = + event.getAttribute(AttributeName.CONVERGENCE_STATUS.name()); + HelixManager manager = event.getAttribute(AttributeName.helixmanager.name()); + if (manager == null || !isCurrentLeader(event, manager)) { + return; + } + HelixDataAccessor accessor = manager.getHelixDataAccessor(); + PropertyKey.Builder keyBuilder = accessor.keyBuilder(); + ConvergenceStatusPersistenceCache persistenceCache = + event.getAttribute(AttributeName.CONVERGENCE_STATUS_PERSISTENCE_CACHE.name()); + if (persistenceCache == null) { + persistenceCache = _localPersistenceCache; + } + initializePersistedState(accessor, keyBuilder, manager.getSessionId(), persistenceCache); + + if (snapshot == null) { + long delay = persistenceCache.getRemainingPersistDelay(System.currentTimeMillis(), + MINIMUM_PERSIST_INTERVAL_MS); + if (delay > 0) { + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + if (!isCurrentLeader(event, manager)) { + return; + } + snapshot = calculateSnapshot(event); + if (snapshot == null) { + return; + } + } + + ConvergenceStatus oldClusterStatus = persistenceCache.getClusterStatus(); + Map oldResourceStatuses = + persistenceCache.getResourceStatuses(); + + long now = System.currentTimeMillis(); + Map newResourceStatuses = + new LinkedHashMap<>(snapshot.getResourceStatuses()); + List changedKeys = new ArrayList<>(); + List changedStatuses = new ArrayList<>(); + long oldestUnconvergedSince = 0L; + + for (Map.Entry entry : newResourceStatuses.entrySet()) { + String resourceName = entry.getKey(); + ConvergenceStatus status = + new ConvergenceStatus(new ZNRecord(entry.getValue().getRecord())); + ConvergenceStatus oldStatus = oldResourceStatuses.get(resourceName); + updateUnconvergedSince(status, oldStatus, now); + if (status.getUnconvergedSince() > 0 + && (oldestUnconvergedSince == 0 + || status.getUnconvergedSince() < oldestUnconvergedSince)) { + oldestUnconvergedSince = status.getUnconvergedSince(); + } + newResourceStatuses.put(resourceName, status); + if (oldStatus == null || !status.semanticallyEquals(oldStatus)) { + changedKeys.add(keyBuilder.convergenceStatus(resourceName)); + changedStatuses.add(status); + } + } + + ConvergenceStatus newClusterStatus = snapshot.getClusterStatus(); + newClusterStatus.setUnconvergedSince(oldestUnconvergedSince); + List removedResources = new ArrayList<>(); + for (String resourceName : oldResourceStatuses.keySet()) { + if (!newResourceStatuses.containsKey(resourceName)) { + removedResources.add(resourceName); + } + } + boolean hasChildMutations = !changedKeys.isEmpty() || !removedResources.isEmpty(); + + if (!isCurrentLeader(event, manager)) { + return; + } + + if (hasChildMutations) { + ConvergenceStatus incompleteStatus = + new ConvergenceStatus(new ZNRecord(newClusterStatus.getRecord())); + incompleteStatus.setComplete(false); + if (!accessor.setProperty(keyBuilder.convergenceStatus(), incompleteStatus)) { + LOG.warn("Failed to mark convergence status update incomplete for cluster {}", + manager.getClusterName()); + return; + } + } + + boolean childUpdatesSucceeded = true; + if (!changedKeys.isEmpty()) { + boolean[] results = accessor.setChildren(changedKeys, changedStatuses); + for (int i = 0; i < results.length; i++) { + if (!results[i]) { + childUpdatesSucceeded = false; + LOG.warn("Failed to persist convergence status at {}", changedKeys.get(i).getPath()); + } + } + } + + for (String resourceName : removedResources) { + if (!accessor.removeProperty(keyBuilder.convergenceStatus(resourceName))) { + childUpdatesSucceeded = false; + LOG.warn("Failed to remove convergence status for dropped resource {}", resourceName); + } + } + + if (!childUpdatesSucceeded) { + return; + } + + boolean rootWriteRequired = hasChildMutations || oldClusterStatus == null + || !newClusterStatus.semanticallyEquals(oldClusterStatus); + if (rootWriteRequired) { + if (!accessor.setProperty(keyBuilder.convergenceStatus(), newClusterStatus)) { + LOG.warn("Failed to persist convergence status root for cluster {}", + manager.getClusterName()); + return; + } + persistenceCache.update(manager.getSessionId(), newClusterStatus, newResourceStatuses); + persistenceCache.markPersisted(System.currentTimeMillis()); + } + } + + private ConvergenceStatusSnapshot calculateSnapshot(ClusterEvent event) { + @SuppressWarnings("unchecked") + Map resources = + event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name()); + CurrentStateOutput currentStateOutput = + event.getAttribute(AttributeName.CURRENT_STATE.name()); + ConvergenceStatusContext context = + event.getAttribute(AttributeName.CONVERGENCE_STATUS_CONTEXT.name()); + if (resources == null || currentStateOutput == null || context == null) { + return null; + } + try { + String pipelineType = event.getAttribute(AttributeName.PipelineType.name()); + if (Pipeline.Type.MANAGEMENT_MODE.name().equals(pipelineType)) { + return _calculator.calculatePaused(event, resources, currentStateOutput, + Reason.MANAGEMENT_MODE); + } + return _calculator.calculate(event, context); + } catch (Exception e) { + LOG.error("Failed to calculate convergence for event {}", event.getEventId(), e); + return _calculator.calculateUnknown(event, resources, currentStateOutput, + Reason.TARGET_ASSIGNMENT_MISSING); + } + } + + private void initializePersistedState(HelixDataAccessor accessor, PropertyKey.Builder keyBuilder, + String sessionId, ConvergenceStatusPersistenceCache persistenceCache) { + if (persistenceCache.isInitializedFor(sessionId)) { + return; + } + ConvergenceStatus clusterStatus = accessor.getProperty(keyBuilder.convergenceStatus()); + Map resourceStatuses = + clusterStatus == null ? Collections.emptyMap() + : accessor.getChildValuesMap(keyBuilder.convergenceStatus(), false); + persistenceCache.update(sessionId, clusterStatus, resourceStatuses); + } + + private static void updateUnconvergedSince(ConvergenceStatus status, + ConvergenceStatus oldStatus, long now) { + if (status.getStatus() == Status.CONVERGED) { + status.setUnconvergedSince(0L); + return; + } + if (oldStatus != null && oldStatus.getStatus() != Status.CONVERGED + && oldStatus.getUnconvergedSince() > 0) { + status.setUnconvergedSince(oldStatus.getUnconvergedSince()); + } else { + status.setUnconvergedSince(now); + } + } + + private static boolean isCurrentLeader(ClusterEvent event, HelixManager manager) { + if (!manager.isLeader()) { + return false; + } + Optional expectedSession = + event.getAttribute(AttributeName.EVENT_SESSION.name()); + return expectedSession == null || !expectedSession.isPresent() + || expectedSession.get().equals(manager.getSessionId()); + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusPersistenceCache.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusPersistenceCache.java new file mode 100644 index 0000000000..21fa6e8e54 --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusPersistenceCache.java @@ -0,0 +1,81 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.zookeeper.datamodel.ZNRecord; + +/** + * Controller-scoped cache shared by default and management pipelines. + */ +public final class ConvergenceStatusPersistenceCache { + private String _sessionId; + private ConvergenceStatus _clusterStatus; + private Map _resourceStatuses = Collections.emptyMap(); + private long _lastPersistTimestamp; + + public boolean isInitializedFor(String sessionId) { + return sessionId != null && sessionId.equals(_sessionId); + } + + public ConvergenceStatus getClusterStatus() { + return copy(_clusterStatus); + } + + public Map getResourceStatuses() { + return copy(_resourceStatuses); + } + + public void update(String sessionId, ConvergenceStatus clusterStatus, + Map resourceStatuses) { + if (sessionId != null && !sessionId.equals(_sessionId)) { + _lastPersistTimestamp = 0L; + } + _sessionId = sessionId; + _clusterStatus = copy(clusterStatus); + _resourceStatuses = copy(resourceStatuses); + } + + public long getRemainingPersistDelay(long now, long minimumIntervalMs) { + return Math.max(0L, _lastPersistTimestamp + minimumIntervalMs - now); + } + + public void markPersisted(long timestamp) { + _lastPersistTimestamp = timestamp; + } + + private static ConvergenceStatus copy(ConvergenceStatus status) { + return status == null ? null : new ConvergenceStatus(new ZNRecord(status.getRecord())); + } + + private static Map copy( + Map statuses) { + if (statuses == null || statuses.isEmpty()) { + return Collections.emptyMap(); + } + Map result = new LinkedHashMap<>(); + statuses.forEach((name, status) -> result.put(name, copy(status))); + return Collections.unmodifiableMap(result); + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusSnapshot.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusSnapshot.java new file mode 100644 index 0000000000..59b0163585 --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ConvergenceStatusSnapshot.java @@ -0,0 +1,57 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.zookeeper.datamodel.ZNRecord; + +/** + * Immutable convergence records calculated from one controller event. + */ +public class ConvergenceStatusSnapshot { + private final ConvergenceStatus _clusterStatus; + private final Map _resourceStatuses; + + public ConvergenceStatusSnapshot(ConvergenceStatus clusterStatus, + Map resourceStatuses) { + _clusterStatus = new ConvergenceStatus(new ZNRecord(clusterStatus.getRecord())); + Map copy = new LinkedHashMap<>(); + resourceStatuses.forEach( + (name, status) -> copy.put(name, + new ConvergenceStatus(new ZNRecord(status.getRecord())))); + _resourceStatuses = Collections.unmodifiableMap(copy); + } + + public ConvergenceStatus getClusterStatus() { + return new ConvergenceStatus(new ZNRecord(_clusterStatus.getRecord())); + } + + public Map getResourceStatuses() { + Map copy = new LinkedHashMap<>(); + _resourceStatuses.forEach( + (name, status) -> copy.put(name, + new ConvergenceStatus(new ZNRecord(status.getRecord())))); + return Collections.unmodifiableMap(copy); + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java index 0a85dc7b77..6bb576a2b8 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CurrentStateOutput.java @@ -260,6 +260,17 @@ public Message getCancellationMessage(String resourceName, Partition partition, return getStateMessage(resourceName, partition, instanceName, _cancellationMessageMap); } + public Map getCancellationMessageMap(String resourceName, + Partition partition) { + if (_cancellationMessageMap.containsKey(resourceName)) { + Map> map = _cancellationMessageMap.get(resourceName); + if (map.containsKey(partition)) { + return map.get(partition); + } + } + return Collections.emptyMap(); + } + private Message getStateMessage(String resourceName, Partition partition, String instanceName, Map>> stateMessageMap) { Map> map = stateMessageMap.get(resourceName); diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchResult.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchResult.java new file mode 100644 index 0000000000..8b8e74c133 --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchResult.java @@ -0,0 +1,51 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.apache.helix.model.Message; + +/** + * Immutable result of one physical message dispatch attempt. + */ +public class MessageDispatchResult { + private final List _sentMessages; + private final List _failedMessages; + + public MessageDispatchResult(List sentMessages, List failedMessages) { + _sentMessages = + Collections.unmodifiableList(new ArrayList<>(sentMessages == null + ? Collections.emptyList() : sentMessages)); + _failedMessages = + Collections.unmodifiableList(new ArrayList<>(failedMessages == null + ? Collections.emptyList() : failedMessages)); + } + + public List getSentMessages() { + return _sentMessages; + } + + public List getFailedMessages() { + return _failedMessages; + } +} diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchStage.java index fab07e4a8d..5d1ef90b3a 100644 --- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchStage.java +++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageDispatchStage.java @@ -22,10 +22,12 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import org.apache.helix.HelixDataAccessor; import org.apache.helix.HelixManager; @@ -94,6 +96,13 @@ protected void processEvent(ClusterEvent event, MessageOutput messageOutput) thr } List messagesSent = sendMessages(dataAccessor, outputMessages); + Set sentMessageIds = new HashSet<>(); + messagesSent.forEach(message -> sentMessageIds.add(message.getMsgId())); + List messagesFailed = new ArrayList<>(); + outputMessages.stream().filter(message -> !sentMessageIds.contains(message.getMsgId())) + .forEach(messagesFailed::add); + event.addAttribute(AttributeName.MESSAGE_DISPATCH_RESULT.name(), + new MessageDispatchResult(messagesSent, messagesFailed)); // TODO: Need also count messages from task rebalancer if (!(cache instanceof WorkflowControllerDataProvider)) { diff --git a/helix-core/src/main/java/org/apache/helix/model/ClusterConfig.java b/helix-core/src/main/java/org/apache/helix/model/ClusterConfig.java index 845d454df7..40d2325907 100644 --- a/helix-core/src/main/java/org/apache/helix/model/ClusterConfig.java +++ b/helix-core/src/main/java/org/apache/helix/model/ClusterConfig.java @@ -200,6 +200,9 @@ public enum ClusterConfigProperty { // for message ordering and extracted throttle logic. When disabled (default), the original // IntermediateStateCalcStage (V1) is used for backward compatibility. INTERMEDIATE_STATE_CALC_STAGE_V2_ENABLED, + + // Enable asynchronous customer-facing assignment convergence reporting. + CONVERGENCE_MONITORING_ENABLED, } public enum GlobalRebalancePreferenceKey { @@ -867,6 +870,24 @@ public boolean isTargetExternalViewEnabled() { return _record.getBooleanField(ClusterConfigProperty.TARGET_EXTERNALVIEW_ENABLED.name(), false); } + /** + * Enable or disable asynchronous convergence monitoring. Disabled by default so clusters that do + * not consume convergence reports pay only a constant-time feature flag check. + * @param enabled true to enable convergence monitoring + */ + public void setConvergenceMonitoringEnabled(boolean enabled) { + _record.setBooleanField(ClusterConfigProperty.CONVERGENCE_MONITORING_ENABLED.name(), enabled); + } + + /** + * Determine whether convergence monitoring is enabled. + * @return true when convergence monitoring is enabled + */ + public boolean isConvergenceMonitoringEnabled() { + return _record + .getBooleanField(ClusterConfigProperty.CONVERGENCE_MONITORING_ENABLED.name(), false); + } + /** * Get maximum allowed running task count on all instances in this cluster. * @return the maximum task count diff --git a/helix-core/src/main/java/org/apache/helix/model/ConvergenceStatus.java b/helix-core/src/main/java/org/apache/helix/model/ConvergenceStatus.java new file mode 100644 index 0000000000..02439b9142 --- /dev/null +++ b/helix-core/src/main/java/org/apache/helix/model/ConvergenceStatus.java @@ -0,0 +1,480 @@ +package org.apache.helix.model; + +/* + * 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 java.util.ArrayList; +import java.util.Collections; +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; + +import org.apache.helix.HelixProperty; +import org.apache.helix.zookeeper.datamodel.ZNRecord; + +/** + * A bounded, customer-facing snapshot of assignment convergence. + */ +public class ConvergenceStatus extends HelixProperty { + public static final int SCHEMA_VERSION = 1; + public static final int DEFAULT_MAX_PARTITION_DETAILS = 20; + public static final int DEFAULT_MAX_ASSIGNMENT_ENTRIES = 64; + + private static final String STATUS_COUNTS = "STATUS_COUNTS"; + private static final String REASON_COUNTS = "REASON_COUNTS"; + private static final String DETAIL_PREFIX = "DETAIL_"; + private static final String CURRENT_PREFIX = "CURRENT_"; + private static final String TARGET_PREFIX = "TARGET_"; + + public enum Scope { + CLUSTER, + RESOURCE + } + + public enum Status { + CONVERGED, + IN_PROGRESS, + BLOCKED, + UNKNOWN, + PAUSED + } + + public enum Reason { + NONE, + PENDING_TRANSITION, + CANCELLATION_PENDING, + RELAY_PENDING, + TRANSITION_DISPATCHED, + MESSAGE_THROTTLED, + STATE_CONSTRAINT_WAIT, + MESSAGE_DISPATCH_FAILED, + ERROR_STATE, + TARGET_INSTANCE_NOT_LIVE, + NO_PROGRESS_PATH, + TARGET_ASSIGNMENT_MISSING, + INVALID_STATE, + MAINTENANCE_MODE, + MANAGEMENT_MODE, + WAITING_FOR_DELAY, + WAGED_LAST_KNOWN_GOOD, + WAGED_CAPACITY_DEFICIT, + WAGED_NO_CANDIDATE_NODE, + WAGED_INVALID_CONFIGURATION, + WAGED_INTERNAL_FAILURE + } + + public enum TargetFreshness { + CURRENT, + LAST_KNOWN_GOOD, + UNKNOWN + } + + public enum OptimizerStatus { + NOT_APPLICABLE, + HEALTHY, + BASELINE_FAILED, + UNKNOWN + } + + public enum Field { + SCHEMA_VERSION, + SCOPE, + STATUS, + PRIMARY_REASON, + RESOURCE_NAME, + GENERATED_AT, + UNCONVERGED_SINCE, + CONTROLLER_SESSION_ID, + SOURCE_EVENT_ID, + TOTAL_RESOURCE_COUNT, + TOTAL_PARTITION_COUNT, + CONVERGED_PARTITION_COUNT, + IN_PROGRESS_PARTITION_COUNT, + BLOCKED_PARTITION_COUNT, + UNKNOWN_PARTITION_COUNT, + AFFECTED_PARTITION_COUNT, + TRUNCATED_PARTITION_COUNT, + TARGET_FRESHNESS, + OPTIMIZER_STATUS, + COMPLETE + } + + public static final class PartitionDetail { + private final String _resourceName; + private final String _partitionName; + private final Status _status; + private final Reason _reason; + private final Map _currentAssignment; + private final Map _targetAssignment; + + public PartitionDetail(String resourceName, String partitionName, Status status, Reason reason, + Map currentAssignment, Map targetAssignment) { + _resourceName = resourceName; + _partitionName = partitionName; + _status = status; + _reason = reason; + _currentAssignment = immutableSortedCopy(currentAssignment); + _targetAssignment = immutableSortedCopy(targetAssignment); + } + + public String getResourceName() { + return _resourceName; + } + + public String getPartitionName() { + return _partitionName; + } + + public Status getStatus() { + return _status; + } + + public Reason getReason() { + return _reason; + } + + public Map getCurrentAssignment() { + return _currentAssignment; + } + + public Map getTargetAssignment() { + return _targetAssignment; + } + } + + public ConvergenceStatus(String id) { + super(new ZNRecord(id)); + setSchemaVersion(SCHEMA_VERSION); + setComplete(true); + } + + public ConvergenceStatus(ZNRecord record) { + super(record); + } + + public void setSchemaVersion(int version) { + _record.setIntField(Field.SCHEMA_VERSION.name(), version); + } + + public int getSchemaVersion() { + return _record.getIntField(Field.SCHEMA_VERSION.name(), 0); + } + + public void setScope(Scope scope) { + _record.setEnumField(Field.SCOPE.name(), scope); + } + + public Scope getScope() { + return _record.getEnumField(Field.SCOPE.name(), Scope.class, null); + } + + public void setStatus(Status status) { + _record.setEnumField(Field.STATUS.name(), status); + } + + public Status getStatus() { + return _record.getEnumField(Field.STATUS.name(), Status.class, Status.UNKNOWN); + } + + public void setPrimaryReason(Reason reason) { + _record.setEnumField(Field.PRIMARY_REASON.name(), reason); + } + + public Reason getPrimaryReason() { + return _record.getEnumField(Field.PRIMARY_REASON.name(), Reason.class, Reason.NONE); + } + + public void setResourceName(String resourceName) { + _record.setSimpleField(Field.RESOURCE_NAME.name(), resourceName); + } + + public String getResourceName() { + return _record.getSimpleField(Field.RESOURCE_NAME.name()); + } + + public void setGeneratedAt(long timestamp) { + _record.setLongField(Field.GENERATED_AT.name(), timestamp); + } + + public long getGeneratedAt() { + return _record.getLongField(Field.GENERATED_AT.name(), 0L); + } + + public void setUnconvergedSince(long timestamp) { + _record.setLongField(Field.UNCONVERGED_SINCE.name(), timestamp); + } + + public long getUnconvergedSince() { + return _record.getLongField(Field.UNCONVERGED_SINCE.name(), 0L); + } + + public void setControllerSessionId(String sessionId) { + _record.setSimpleField(Field.CONTROLLER_SESSION_ID.name(), sessionId); + } + + public String getControllerSessionId() { + return _record.getSimpleField(Field.CONTROLLER_SESSION_ID.name()); + } + + public void setSourceEventId(String eventId) { + _record.setSimpleField(Field.SOURCE_EVENT_ID.name(), eventId); + } + + public String getSourceEventId() { + return _record.getSimpleField(Field.SOURCE_EVENT_ID.name()); + } + + public void setTargetFreshness(TargetFreshness freshness) { + _record.setEnumField(Field.TARGET_FRESHNESS.name(), freshness); + } + + public TargetFreshness getTargetFreshness() { + return _record.getEnumField(Field.TARGET_FRESHNESS.name(), TargetFreshness.class, + TargetFreshness.UNKNOWN); + } + + public void setOptimizerStatus(OptimizerStatus status) { + _record.setEnumField(Field.OPTIMIZER_STATUS.name(), status); + } + + public OptimizerStatus getOptimizerStatus() { + return _record.getEnumField(Field.OPTIMIZER_STATUS.name(), OptimizerStatus.class, + OptimizerStatus.UNKNOWN); + } + + public void setComplete(boolean complete) { + _record.setBooleanField(Field.COMPLETE.name(), complete); + } + + public boolean isComplete() { + return _record.getBooleanField(Field.COMPLETE.name(), false); + } + + public void setTotalResourceCount(int value) { + setCount(Field.TOTAL_RESOURCE_COUNT, value); + } + + public int getTotalResourceCount() { + return getCount(Field.TOTAL_RESOURCE_COUNT); + } + + public void setTotalPartitionCount(int value) { + setCount(Field.TOTAL_PARTITION_COUNT, value); + } + + public int getTotalPartitionCount() { + return getCount(Field.TOTAL_PARTITION_COUNT); + } + + public void setConvergedPartitionCount(int value) { + setCount(Field.CONVERGED_PARTITION_COUNT, value); + } + + public int getConvergedPartitionCount() { + return getCount(Field.CONVERGED_PARTITION_COUNT); + } + + public void setInProgressPartitionCount(int value) { + setCount(Field.IN_PROGRESS_PARTITION_COUNT, value); + } + + public int getInProgressPartitionCount() { + return getCount(Field.IN_PROGRESS_PARTITION_COUNT); + } + + public void setBlockedPartitionCount(int value) { + setCount(Field.BLOCKED_PARTITION_COUNT, value); + } + + public int getBlockedPartitionCount() { + return getCount(Field.BLOCKED_PARTITION_COUNT); + } + + public void setUnknownPartitionCount(int value) { + setCount(Field.UNKNOWN_PARTITION_COUNT, value); + } + + public int getUnknownPartitionCount() { + return getCount(Field.UNKNOWN_PARTITION_COUNT); + } + + public void setAffectedPartitionCount(int value) { + setCount(Field.AFFECTED_PARTITION_COUNT, value); + } + + public int getAffectedPartitionCount() { + return getCount(Field.AFFECTED_PARTITION_COUNT); + } + + public void setTruncatedPartitionCount(int value) { + setCount(Field.TRUNCATED_PARTITION_COUNT, value); + } + + public int getTruncatedPartitionCount() { + return getCount(Field.TRUNCATED_PARTITION_COUNT); + } + + public void setStatusCounts(Map counts) { + _record.setMapField(STATUS_COUNTS, enumCountMap(counts)); + } + + public Map getStatusCounts() { + return parseEnumCountMap(_record.getMapField(STATUS_COUNTS), Status.class); + } + + public void setReasonCounts(Map counts) { + _record.setMapField(REASON_COUNTS, enumCountMap(counts)); + } + + public Map getReasonCounts() { + return parseEnumCountMap(_record.getMapField(REASON_COUNTS), Reason.class); + } + + public void setPartitionDetails(List details, int maxDetails) { + clearPartitionDetails(); + int retained = Math.min(details.size(), Math.max(0, maxDetails)); + for (int i = 0; i < retained; i++) { + PartitionDetail detail = details.get(i); + String suffix = detailSuffix(i); + Map metadata = new TreeMap<>(); + metadata.put(Field.RESOURCE_NAME.name(), detail.getResourceName()); + metadata.put("PARTITION_NAME", detail.getPartitionName()); + metadata.put(Field.STATUS.name(), detail.getStatus().name()); + metadata.put(Field.PRIMARY_REASON.name(), detail.getReason().name()); + metadata.put("CURRENT_ASSIGNMENT_COUNT", + String.valueOf(detail.getCurrentAssignment().size())); + metadata.put("TARGET_ASSIGNMENT_COUNT", String.valueOf(detail.getTargetAssignment().size())); + _record.setMapField(DETAIL_PREFIX + suffix, metadata); + _record.setMapField(CURRENT_PREFIX + suffix, + boundedMap(detail.getCurrentAssignment(), DEFAULT_MAX_ASSIGNMENT_ENTRIES)); + _record.setMapField(TARGET_PREFIX + suffix, + boundedMap(detail.getTargetAssignment(), DEFAULT_MAX_ASSIGNMENT_ENTRIES)); + } + setTruncatedPartitionCount(Math.max(0, details.size() - retained)); + } + + public List getPartitionDetails() { + List details = new ArrayList<>(); + for (int i = 0; i < DEFAULT_MAX_PARTITION_DETAILS; i++) { + String suffix = detailSuffix(i); + Map metadata = _record.getMapField(DETAIL_PREFIX + suffix); + if (metadata == null) { + continue; + } + String resourceName = metadata.get(Field.RESOURCE_NAME.name()); + String partitionName = metadata.get("PARTITION_NAME"); + Status status = parseEnum(Status.class, metadata.get(Field.STATUS.name()), Status.UNKNOWN); + Reason reason = + parseEnum(Reason.class, metadata.get(Field.PRIMARY_REASON.name()), Reason.NONE); + details.add(new PartitionDetail(resourceName, partitionName, status, reason, + _record.getMapField(CURRENT_PREFIX + suffix), + _record.getMapField(TARGET_PREFIX + suffix))); + } + return Collections.unmodifiableList(details); + } + + public boolean semanticallyEquals(ConvergenceStatus other) { + if (other == null) { + return false; + } + ZNRecord left = new ZNRecord(_record); + ZNRecord right = new ZNRecord(other.getRecord()); + removeVolatileFields(left); + removeVolatileFields(right); + return left.equals(right); + } + + @Override + public boolean isValid() { + return getSchemaVersion() == SCHEMA_VERSION && getScope() != null && getStatus() != null; + } + + private void clearPartitionDetails() { + _record.getMapFields().keySet().removeIf( + key -> key.startsWith(DETAIL_PREFIX) || key.startsWith(CURRENT_PREFIX) + || key.startsWith(TARGET_PREFIX)); + } + + private void setCount(Field field, int value) { + _record.setIntField(field.name(), Math.max(0, value)); + } + + private int getCount(Field field) { + return _record.getIntField(field.name(), 0); + } + + private static Map immutableSortedCopy(Map input) { + if (input == null || input.isEmpty()) { + return Collections.emptyMap(); + } + return Collections.unmodifiableMap(new TreeMap<>(input)); + } + + private static Map boundedMap(Map input, int maxEntries) { + Map result = new TreeMap<>(); + input.entrySet().stream().limit(maxEntries) + .forEach(entry -> result.put(entry.getKey(), entry.getValue())); + return result; + } + + private static > Map enumCountMap(Map counts) { + Map result = new TreeMap<>(); + if (counts != null) { + counts.forEach((key, value) -> result.put(key.name(), String.valueOf(value))); + } + return result; + } + + private static > Map parseEnumCountMap(Map input, + Class enumType) { + Map result = new EnumMap<>(enumType); + if (input != null) { + input.forEach((key, value) -> { + try { + result.put(Enum.valueOf(enumType, key), Integer.parseInt(value)); + } catch (IllegalArgumentException ignored) { + // Ignore fields written by a newer schema. + } + }); + } + return Collections.unmodifiableMap(result); + } + + private static > E parseEnum(Class enumType, String value, E defaultValue) { + if (value == null) { + return defaultValue; + } + try { + return Enum.valueOf(enumType, value); + } catch (IllegalArgumentException ignored) { + return defaultValue; + } + } + + private static String detailSuffix(int index) { + return String.format("%03d", index); + } + + private static void removeVolatileFields(ZNRecord record) { + record.getSimpleFields().remove(Field.GENERATED_AT.name()); + record.getSimpleFields().remove(Field.SOURCE_EVENT_ID.name()); + record.getSimpleFields().remove(Field.UNCONVERGED_SINCE.name()); + } +} diff --git a/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java b/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java index 9212568356..1b1eedf047 100644 --- a/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java +++ b/helix-core/src/test/java/org/apache/helix/TestPropertyPathBuilder.java @@ -59,5 +59,13 @@ public void testGetPath() { actual = PropertyPathBuilder.clusterStatus("test_cluster"); Assert.assertEquals(actual, "/test_cluster/STATUS/CLUSTER/test_cluster"); + + actual = PropertyPathBuilder.getPath(PropertyType.CONVERGENCESTATUS, "test_cluster"); + Assert.assertEquals(actual, + "/test_cluster/PROPERTYSTORE/HELIX_CONVERGENCE_STATUS"); + actual = + PropertyPathBuilder.getPath(PropertyType.CONVERGENCESTATUS, "test_cluster", "resource"); + Assert.assertEquals(actual, + "/test_cluster/PROPERTYSTORE/HELIX_CONVERGENCE_STATUS/resource"); } } diff --git a/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/TestWagedRebalanceStatus.java b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/TestWagedRebalanceStatus.java new file mode 100644 index 0000000000..f9cb5ef51f --- /dev/null +++ b/helix-core/src/test/java/org/apache/helix/controller/rebalancer/waged/TestWagedRebalanceStatus.java @@ -0,0 +1,71 @@ +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 java.util.Optional; + +import org.apache.helix.HelixRebalanceException; +import org.apache.helix.controller.rebalancer.waged.constraints.MockRebalanceAlgorithm; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestWagedRebalanceStatus { + @Test + public void testServingFailureAndRecovery_areExposedToConvergenceTracker() { + WagedRebalancer rebalancer = + new WagedRebalancer(null, new MockRebalanceAlgorithm(), Optional.empty()); + try { + HelixRebalanceException failure = + new HelixRebalanceException("capacity exhausted", + HelixRebalanceException.Type.FAILED_TO_CALCULATE, + HelixRebalanceException.FailureCategory.CAPACITY_DEFICIT); + + rebalancer.reportAsyncFailure(failure); + WagedRebalanceStatus failed = rebalancer.getConvergenceStatus(); + + Assert.assertTrue(failed.isServingComputationFailed()); + Assert.assertEquals(failed.getServingFailureCategory(), + HelixRebalanceException.FailureCategory.CAPACITY_DEFICIT); + + rebalancer.reportPartialRebalanceSuccess(); + WagedRebalanceStatus recovered = rebalancer.getConvergenceStatus(); + + Assert.assertFalse(recovered.isServingComputationFailed()); + Assert.assertNull(recovered.getServingFailureCategory()); + } finally { + rebalancer.close(); + } + } + + @Test + public void testBaselineFailureAndRecovery_areExposedAsOptimizerHealth() { + WagedRebalancer rebalancer = + new WagedRebalancer(null, new MockRebalanceAlgorithm(), Optional.empty()); + try { + rebalancer.reportBaselineComputeStatus(false); + Assert.assertTrue(rebalancer.getConvergenceStatus().isBaselineComputationFailed()); + + rebalancer.reportBaselineComputeStatus(true); + Assert.assertFalse(rebalancer.getConvergenceStatus().isBaselineComputationFailed()); + } finally { + rebalancer.close(); + } + } +} diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestConvergenceStatusCalculator.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestConvergenceStatusCalculator.java new file mode 100644 index 0000000000..f010c1e186 --- /dev/null +++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestConvergenceStatusCalculator.java @@ -0,0 +1,236 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import org.apache.helix.MockAccessor; +import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider; +import org.apache.helix.model.ClusterConfig; +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.model.ConvergenceStatus.Reason; +import org.apache.helix.model.ConvergenceStatus.Status; +import org.apache.helix.model.IdealState; +import org.apache.helix.model.IdealState.RebalanceMode; +import org.apache.helix.model.LiveInstance; +import org.apache.helix.model.Message; +import org.apache.helix.model.Message.MessageType; +import org.apache.helix.model.Partition; +import org.apache.helix.model.Resource; +import org.apache.helix.model.StateModelDefinition; +import org.apache.helix.tools.StateModelConfigGenerator; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class TestConvergenceStatusCalculator { + private static final String CLUSTER = "TestConvergenceCluster"; + private static final String RESOURCE = "TestDB"; + private static final String PARTITION_NAME = "TestDB_0"; + private static final String INSTANCE = "localhost_12918"; + private static final String STATE_MODEL = "MasterSlave"; + + private ClusterEvent _event; + private Partition _partition; + private CurrentStateOutput _currentStateOutput; + private BestPossibleStateOutput _bestPossibleStateOutput; + + @BeforeMethod + public void setUp() { + MockAccessor accessor = new MockAccessor(); + accessor.setProperty(accessor.keyBuilder().clusterConfig(), new ClusterConfig(CLUSTER)); + + StateModelDefinition stateModelDefinition = + new StateModelDefinition(StateModelConfigGenerator.generateConfigForMasterSlave()); + accessor.setProperty(accessor.keyBuilder().stateModelDef(STATE_MODEL), stateModelDefinition); + + IdealState idealState = new IdealState(RESOURCE); + idealState.setStateModelDefRef(STATE_MODEL); + idealState.setRebalanceMode(RebalanceMode.CUSTOMIZED); + idealState.setNumPartitions(1); + idealState.setReplicas("1"); + accessor.setProperty(accessor.keyBuilder().idealStates(RESOURCE), idealState); + + LiveInstance liveInstance = new LiveInstance(INSTANCE); + liveInstance.setSessionId("session_0"); + accessor.setProperty(accessor.keyBuilder().liveInstance(INSTANCE), liveInstance); + + ResourceControllerDataProvider cache = new ResourceControllerDataProvider(CLUSTER); + cache.refresh(accessor); + + Resource resource = new Resource(RESOURCE); + resource.setStateModelDefRef(STATE_MODEL); + resource.addPartition(PARTITION_NAME); + _partition = resource.getPartition(PARTITION_NAME); + + _currentStateOutput = new CurrentStateOutput(); + _bestPossibleStateOutput = new BestPossibleStateOutput(); + _event = new ClusterEvent(CLUSTER, ClusterEventType.CurrentStateChange, "event_0"); + _event.addAttribute(AttributeName.ControllerDataProvider.name(), cache); + _event.addAttribute(AttributeName.RESOURCES_TO_REBALANCE.name(), + Collections.singletonMap(RESOURCE, resource)); + _event.addAttribute(AttributeName.CURRENT_STATE.name(), _currentStateOutput); + _event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), _bestPossibleStateOutput); + _event.addAttribute(AttributeName.helixmanager.name(), + new DummyClusterManager(CLUSTER, accessor, "controller_session")); + _event.addAttribute(AttributeName.MESSAGES_ALL.name(), new MessageOutput()); + _event.addAttribute(AttributeName.MESSAGES_SELECTED.name(), new MessageOutput()); + _event.addAttribute(AttributeName.MESSAGES_THROTTLE.name(), new MessageOutput()); + _event.addAttribute(AttributeName.MESSAGE_DISPATCH_RESULT.name(), + new MessageDispatchResult(Collections.emptyList(), Collections.emptyList())); + } + + @Test + public void testCalculate_matchingAssignment_reportsConverged() { + setCurrentAndTarget("MASTER", "MASTER"); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.CONVERGED); + Assert.assertEquals(status.getConvergedPartitionCount(), 1); + Assert.assertEquals(status.getAffectedPartitionCount(), 0); + Assert.assertTrue(status.getPartitionDetails().isEmpty()); + } + + @Test + public void testCalculate_pendingTransition_reportsInProgress() { + setCurrentAndTarget("SLAVE", "MASTER"); + _currentStateOutput.setPendingMessage(RESOURCE, _partition, INSTANCE, + transitionMessage("pending", "SLAVE", "MASTER")); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.IN_PROGRESS); + Assert.assertEquals(status.getPrimaryReason(), Reason.PENDING_TRANSITION); + Assert.assertEquals(status.getInProgressPartitionCount(), 1); + Assert.assertEquals(status.getPartitionDetails().get(0).getCurrentAssignment().get(INSTANCE), + "SLAVE"); + Assert.assertEquals(status.getPartitionDetails().get(0).getTargetAssignment().get(INSTANCE), + "MASTER"); + } + + @Test + public void testCalculate_generatedButUnselected_reportsConstraintWait() { + setCurrentAndTarget("SLAVE", "MASTER"); + MessageOutput generated = new MessageOutput(); + generated.addMessage(RESOURCE, _partition, transitionMessage("generated", "SLAVE", "MASTER")); + _event.addAttribute(AttributeName.MESSAGES_ALL.name(), generated); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.IN_PROGRESS); + Assert.assertEquals(status.getPrimaryReason(), Reason.STATE_CONSTRAINT_WAIT); + } + + @Test + public void testCalculate_failedDispatch_reportsBlocked() { + setCurrentAndTarget("SLAVE", "MASTER"); + Message failed = transitionMessage("failed", "SLAVE", "MASTER"); + _event.addAttribute(AttributeName.MESSAGE_DISPATCH_RESULT.name(), + new MessageDispatchResult(Collections.emptyList(), Collections.singletonList(failed))); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.BLOCKED); + Assert.assertEquals(status.getPrimaryReason(), Reason.MESSAGE_DISPATCH_FAILED); + Assert.assertEquals(status.getBlockedPartitionCount(), 1); + } + + @Test + public void testCalculate_noProgressPath_reportsBlocked() { + setCurrentAndTarget("SLAVE", "MASTER"); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.BLOCKED); + Assert.assertEquals(status.getPrimaryReason(), Reason.NO_PROGRESS_PATH); + } + + @Test + public void testCalculate_missingTarget_reportsUnknown() { + _currentStateOutput.setCurrentState(RESOURCE, _partition, INSTANCE, "SLAVE"); + _event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), new BestPossibleStateOutput()); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.UNKNOWN); + Assert.assertEquals(status.getPrimaryReason(), Reason.TARGET_ASSIGNMENT_MISSING); + Assert.assertEquals(status.getUnknownPartitionCount(), 1); + } + + @Test + public void testCalculate_initialAndDroppedStates_areTreatedAsAbsent() { + Map target = new HashMap<>(); + target.put(INSTANCE, "DROPPED"); + _currentStateOutput.setCurrentState(RESOURCE, _partition, INSTANCE, "OFFLINE"); + _bestPossibleStateOutput.setState(RESOURCE, _partition, target); + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getStatus(), Status.CONVERGED); + } + + @Test + public void testCalculate_largeAffectedResource_keepsDiagnosticDetailsBounded() { + @SuppressWarnings("unchecked") + Map resources = + _event.getAttribute(AttributeName.RESOURCES_TO_REBALANCE.name()); + Resource resource = resources.get(RESOURCE); + int partitionCount = 10_000; + for (int i = 0; i < partitionCount; i++) { + String partitionName = RESOURCE + "_" + i; + resource.addPartition(partitionName); + Partition partition = resource.getPartition(partitionName); + _currentStateOutput.setCurrentState(RESOURCE, partition, INSTANCE, "SLAVE"); + _bestPossibleStateOutput.setState(RESOURCE, partition, INSTANCE, "MASTER"); + } + + ConvergenceStatus status = calculateResourceStatus(); + + Assert.assertEquals(status.getTotalPartitionCount(), partitionCount); + Assert.assertEquals(status.getAffectedPartitionCount(), partitionCount); + Assert.assertEquals(status.getPartitionDetails().size(), + ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS); + Assert.assertEquals(status.getTruncatedPartitionCount(), + partitionCount - ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS); + } + + private void setCurrentAndTarget(String currentState, String targetState) { + _currentStateOutput.setCurrentState(RESOURCE, _partition, INSTANCE, currentState); + _bestPossibleStateOutput.setState(RESOURCE, _partition, INSTANCE, targetState); + } + + private ConvergenceStatus calculateResourceStatus() { + ConvergenceStatusSnapshot snapshot = new ConvergenceStatusCalculator().calculate(_event); + return snapshot.getResourceStatuses().get(RESOURCE); + } + + private Message transitionMessage(String id, String fromState, String toState) { + Message message = new Message(MessageType.STATE_TRANSITION, id); + message.setResourceName(RESOURCE); + message.setPartitionName(PARTITION_NAME); + message.setTgtName(INSTANCE); + message.setFromState(fromState); + message.setToState(toState); + return message; + } +} diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestConvergenceStatusPersistStage.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestConvergenceStatusPersistStage.java new file mode 100644 index 0000000000..4fc6388b9a --- /dev/null +++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestConvergenceStatusPersistStage.java @@ -0,0 +1,162 @@ +package org.apache.helix.controller.stages; + +/* + * 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 java.util.Collections; +import java.util.Map; +import java.util.Optional; + +import org.apache.helix.MockAccessor; +import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider; +import org.apache.helix.model.ClusterConfig; +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.model.ConvergenceStatus.OptimizerStatus; +import org.apache.helix.model.ConvergenceStatus.Reason; +import org.apache.helix.model.ConvergenceStatus.Scope; +import org.apache.helix.model.ConvergenceStatus.Status; +import org.apache.helix.model.ConvergenceStatus.TargetFreshness; +import org.apache.helix.model.Resource; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class TestConvergenceStatusPersistStage { + private static final String CLUSTER = "TestConvergencePersistence"; + private static final String RESOURCE = "TestDB"; + private static final String SESSION = "controller_session"; + + private MockAccessor _accessor; + private DummyClusterManager _manager; + private ClusterEvent _event; + private ConvergenceStatusPersistStage _stage; + + @BeforeMethod + public void setUp() { + _accessor = new MockAccessor(); + _manager = new DummyClusterManager(CLUSTER, _accessor, SESSION) { + @Override + public boolean isLeader() { + return true; + } + }; + _event = new ClusterEvent(CLUSTER, ClusterEventType.CurrentStateChange, "event_0"); + _event.addAttribute(AttributeName.helixmanager.name(), _manager); + _event.addAttribute(AttributeName.EVENT_SESSION.name(), Optional.of(SESSION)); + _stage = new ConvergenceStatusPersistStage(); + } + + @Test + public void testExecute_continuousMismatch_preservesFirstObservedTime() { + persistSnapshot(Status.BLOCKED, Reason.NO_PROGRESS_PATH); + ConvergenceStatus first = + _accessor.getProperty(_accessor.keyBuilder().convergenceStatus(RESOURCE)); + long firstObserved = first.getUnconvergedSince(); + + persistSnapshot(Status.IN_PROGRESS, Reason.PENDING_TRANSITION); + ConvergenceStatus second = + _accessor.getProperty(_accessor.keyBuilder().convergenceStatus(RESOURCE)); + + Assert.assertTrue(firstObserved > 0); + Assert.assertEquals(second.getUnconvergedSince(), firstObserved); + Assert.assertEquals(second.getStatus(), Status.IN_PROGRESS); + } + + @Test + public void testExecute_convergedResource_clearsFirstObservedTime() { + persistSnapshot(Status.BLOCKED, Reason.NO_PROGRESS_PATH); + + persistSnapshot(Status.CONVERGED, Reason.NONE); + ConvergenceStatus status = + _accessor.getProperty(_accessor.keyBuilder().convergenceStatus(RESOURCE)); + + Assert.assertEquals(status.getStatus(), Status.CONVERGED); + Assert.assertEquals(status.getUnconvergedSince(), 0L); + } + + @Test + public void testExecute_removedResource_deletesPersistedStatus() { + persistSnapshot(Status.BLOCKED, Reason.NO_PROGRESS_PATH); + ConvergenceStatus emptyCluster = createStatus(CLUSTER, Scope.CLUSTER, Status.CONVERGED, + Reason.NONE, 0); + _event.addAttribute(AttributeName.CONVERGENCE_STATUS.name(), + new ConvergenceStatusSnapshot(emptyCluster, Collections.emptyMap())); + + _stage.execute(_event); + + Assert.assertNull(_accessor.getProperty(_accessor.keyBuilder().convergenceStatus(RESOURCE))); + ConvergenceStatus cluster = + _accessor.getProperty(_accessor.keyBuilder().convergenceStatus()); + Assert.assertEquals(cluster.getTotalResourceCount(), 0); + } + + @Test + public void testProcess_monitoringDisabled_skipsAsyncWork() throws Exception { + ClusterConfig clusterConfig = new ClusterConfig(CLUSTER); + _accessor.setProperty(_accessor.keyBuilder().clusterConfig(), clusterConfig); + ResourceControllerDataProvider cache = new ResourceControllerDataProvider(CLUSTER); + cache.refresh(_accessor); + Resource resource = new Resource(RESOURCE); + Map resources = Collections.singletonMap(RESOURCE, resource); + _event.addAttribute(AttributeName.ControllerDataProvider.name(), cache); + _event.addAttribute(AttributeName.RESOURCES_TO_REBALANCE.name(), resources); + + _stage.process(_event); + + Assert.assertFalse(_event.containsAttribute( + AttributeName.CONVERGENCE_STATUS_CONTEXT.name())); + Assert.assertFalse(clusterConfig.isConvergenceMonitoringEnabled()); + } + + private void persistSnapshot(Status status, Reason reason) { + ConvergenceStatus resourceStatus = + createStatus(RESOURCE, Scope.RESOURCE, status, reason, 1); + resourceStatus.setResourceName(RESOURCE); + ConvergenceStatus clusterStatus = + createStatus(CLUSTER, Scope.CLUSTER, status, reason, 1); + clusterStatus.setTotalResourceCount(1); + _event.addAttribute(AttributeName.CONVERGENCE_STATUS.name(), + new ConvergenceStatusSnapshot(clusterStatus, + Collections.singletonMap(RESOURCE, resourceStatus))); + + _stage.execute(_event); + } + + private ConvergenceStatus createStatus(String id, Scope scope, Status status, Reason reason, + int partitionCount) { + ConvergenceStatus convergenceStatus = new ConvergenceStatus(id); + convergenceStatus.setScope(scope); + convergenceStatus.setStatus(status); + convergenceStatus.setPrimaryReason(reason); + convergenceStatus.setTargetFreshness(TargetFreshness.CURRENT); + convergenceStatus.setOptimizerStatus(OptimizerStatus.NOT_APPLICABLE); + convergenceStatus.setControllerSessionId(SESSION); + convergenceStatus.setSourceEventId(_event.getEventId()); + convergenceStatus.setGeneratedAt(System.currentTimeMillis()); + convergenceStatus.setTotalPartitionCount(partitionCount); + convergenceStatus.setConvergedPartitionCount(status == Status.CONVERGED ? partitionCount : 0); + convergenceStatus.setInProgressPartitionCount( + status == Status.IN_PROGRESS ? partitionCount : 0); + convergenceStatus.setBlockedPartitionCount(status == Status.BLOCKED ? partitionCount : 0); + convergenceStatus.setUnknownPartitionCount(status == Status.UNKNOWN ? partitionCount : 0); + convergenceStatus.setAffectedPartitionCount( + status == Status.CONVERGED ? 0 : partitionCount); + return convergenceStatus; + } +} diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java index 456ab8c9c2..f98618c87d 100644 --- a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java +++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ClusterAccessor.java @@ -58,6 +58,7 @@ import org.apache.helix.model.CloudConfig; import org.apache.helix.model.ClusterConfig; import org.apache.helix.model.ClusterTopologyConfig; +import org.apache.helix.model.ConvergenceStatus; import org.apache.helix.model.ControllerHistory; import org.apache.helix.model.CustomizedStateConfig; import org.apache.helix.model.HelixConfigScope; @@ -162,6 +163,24 @@ public Response getClusterInfo(@PathParam("clusterId") String clusterId) { return JSONRepresentation(clusterInfo); } + @ClusterAuth + @ResponseMetered(name = HttpConstants.READ_REQUEST) + @Timed(name = HttpConstants.READ_REQUEST) + @GET + @Path("{clusterId}/convergence") + public Response getClusterConvergence(@PathParam("clusterId") String clusterId) { + HelixDataAccessor dataAccessor = getDataAccssor(clusterId); + PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder(); + ConvergenceStatus report = dataAccessor.getProperty(keyBuilder.convergenceStatus()); + LiveInstance leader = dataAccessor.getProperty(keyBuilder.controllerLeader()); + ClusterConfig clusterConfig = dataAccessor.getProperty(keyBuilder.clusterConfig()); + boolean monitoringEnabled = + clusterConfig != null && clusterConfig.isConvergenceMonitoringEnabled(); + return JSONRepresentation( + ConvergenceStatusResponseMapper + .mapCluster(clusterId, report, leader, monitoringEnabled)); + } + @NamespaceAuth @ResponseMetered(name = HttpConstants.WRITE_REQUEST) @Timed(name = HttpConstants.WRITE_REQUEST) diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ConvergenceStatusResponseMapper.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ConvergenceStatusResponseMapper.java new file mode 100644 index 0000000000..523d4edae3 --- /dev/null +++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ConvergenceStatusResponseMapper.java @@ -0,0 +1,217 @@ +package org.apache.helix.rest.server.resources.helix; + +/* + * 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 java.util.Map; + +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.model.ConvergenceStatus.PartitionDetail; +import org.apache.helix.model.ConvergenceStatus.Reason; +import org.apache.helix.model.ConvergenceStatus.Scope; +import org.apache.helix.model.ConvergenceStatus.Status; +import org.apache.helix.model.LiveInstance; + +/** + * Maps the bounded convergence model to the customer-facing REST representation. + */ +final class ConvergenceStatusResponseMapper { + private enum StaleReason { + REPORT_MISSING, + SCHEMA_UNSUPPORTED, + REPORT_INCOMPLETE, + REPORT_INVALID, + CONTROLLER_LEADER_MISSING, + CONTROLLER_SESSION_MISMATCH, + CLUSTER_REPORT_STALE, + MONITORING_DISABLED + } + + private ConvergenceStatusResponseMapper() { + } + + static ObjectNode mapCluster(String clusterId, ConvergenceStatus report, LiveInstance leader) { + return mapCluster(clusterId, report, leader, true); + } + + static ObjectNode mapCluster(String clusterId, ConvergenceStatus report, LiveInstance leader, + boolean monitoringEnabled) { + return map(clusterId, null, report, null, leader, Scope.CLUSTER, monitoringEnabled); + } + + static ObjectNode mapResource(String clusterId, String resourceName, ConvergenceStatus report, + LiveInstance leader) { + return map(clusterId, resourceName, report, null, leader, Scope.RESOURCE, true); + } + + static ObjectNode mapResource(String clusterId, String resourceName, ConvergenceStatus report, + ConvergenceStatus clusterReport, LiveInstance leader) { + return mapResource(clusterId, resourceName, report, clusterReport, leader, true); + } + + static ObjectNode mapResource(String clusterId, String resourceName, ConvergenceStatus report, + ConvergenceStatus clusterReport, LiveInstance leader, boolean monitoringEnabled) { + return map(clusterId, resourceName, report, clusterReport, leader, Scope.RESOURCE, + monitoringEnabled); + } + + static boolean isFreshClusterReport(ConvergenceStatus report, LiveInstance leader) { + return isFreshClusterReport(report, leader, true); + } + + static boolean isFreshClusterReport(ConvergenceStatus report, LiveInstance leader, + boolean monitoringEnabled) { + return monitoringEnabled && staleReason(report, leader, Scope.CLUSTER, null) == null; + } + + private static ObjectNode map(String clusterId, String resourceName, ConvergenceStatus report, + ConvergenceStatus clusterReport, LiveInstance leader, Scope expectedScope, + boolean monitoringEnabled) { + ObjectNode response = JsonNodeFactory.instance.objectNode(); + response.put("clusterId", clusterId); + if (resourceName != null) { + response.put("resourceName", resourceName); + } + + StaleReason staleReason = monitoringEnabled ? staleReason(report, leader, expectedScope, + resourceName) : StaleReason.MONITORING_DISABLED; + if (staleReason == null && clusterReport != null + && staleReason(clusterReport, leader, Scope.CLUSTER, null) != null) { + staleReason = StaleReason.CLUSTER_REPORT_STALE; + } + Status reportedStatus = report == null ? Status.UNKNOWN : report.getStatus(); + Status effectiveStatus = staleReason == null ? reportedStatus : Status.UNKNOWN; + response.put("status", effectiveStatus.name()); + response.put("effectiveStatus", effectiveStatus.name()); + response.put("reportedStatus", reportedStatus.name()); + response.put("primaryReason", + (report == null ? Reason.NONE : report.getPrimaryReason()).name()); + + ObjectNode partitionSummary = response.putObject("partitionSummary"); + if (expectedScope == Scope.CLUSTER) { + response.put("resourceCount", report == null ? 0 : report.getTotalResourceCount()); + } + partitionSummary.put("total", report == null ? 0 : report.getTotalPartitionCount()); + partitionSummary.put("converged", report == null ? 0 : report.getConvergedPartitionCount()); + partitionSummary.put("inProgress", + report == null ? 0 : report.getInProgressPartitionCount()); + partitionSummary.put("blocked", report == null ? 0 : report.getBlockedPartitionCount()); + partitionSummary.put("unknown", report == null ? 0 : report.getUnknownPartitionCount()); + partitionSummary.put("affected", report == null ? 0 : report.getAffectedPartitionCount()); + + ObjectNode reasonCounts = response.putObject("reasonCounts"); + if (report != null) { + report.getReasonCounts() + .forEach((reason, count) -> reasonCounts.put(reason.name(), count)); + } + ObjectNode statusCounts = response.putObject("statusCounts"); + if (report != null) { + report.getStatusCounts() + .forEach((status, count) -> statusCounts.put(status.name(), count)); + } + + response.put("targetFreshness", + (report == null ? ConvergenceStatus.TargetFreshness.UNKNOWN + : report.getTargetFreshness()).name()); + response.put("optimizerStatus", + (report == null ? ConvergenceStatus.OptimizerStatus.UNKNOWN + : report.getOptimizerStatus()).name()); + response.put("oldestUnconvergedMs", oldestUnconvergedMs(report)); + response.put("generatedAt", report == null ? 0L : report.getGeneratedAt()); + putNullable(response, "sourceEventId", report == null ? null : report.getSourceEventId()); + putNullable(response, "controllerSessionId", + report == null ? null : report.getControllerSessionId()); + response.put("stale", staleReason != null); + putNullable(response, "staleReason", staleReason == null ? null : staleReason.name()); + response.put("truncatedPartitionCount", + report == null ? 0 : report.getTruncatedPartitionCount()); + + ArrayNode affectedPartitions = response.putArray("affectedPartitions"); + if (report != null) { + int detailCount = 0; + for (PartitionDetail detail : report.getPartitionDetails()) { + if (detailCount++ >= ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS) { + break; + } + ObjectNode detailNode = affectedPartitions.addObject(); + putNullable(detailNode, "resourceName", detail.getResourceName()); + putNullable(detailNode, "partitionName", detail.getPartitionName()); + detailNode.put("status", detail.getStatus().name()); + detailNode.put("reason", detail.getReason().name()); + putAssignment(detailNode.putObject("currentAssignment"), detail.getCurrentAssignment()); + putAssignment(detailNode.putObject("expectedAssignment"), detail.getTargetAssignment()); + } + } + return response; + } + + private static StaleReason staleReason(ConvergenceStatus report, LiveInstance leader, + Scope expectedScope, String expectedResourceName) { + if (report == null) { + return StaleReason.REPORT_MISSING; + } + if (report.getSchemaVersion() != ConvergenceStatus.SCHEMA_VERSION) { + return StaleReason.SCHEMA_UNSUPPORTED; + } + if (!report.isComplete()) { + return StaleReason.REPORT_INCOMPLETE; + } + if (!report.isValid() || report.getScope() != expectedScope + || expectedResourceName != null + && !expectedResourceName.equals(report.getResourceName())) { + return StaleReason.REPORT_INVALID; + } + if (leader == null) { + return StaleReason.CONTROLLER_LEADER_MISSING; + } + String leaderSession = leader.getEphemeralOwner(); + if (leaderSession == null || !leaderSession.equals(report.getControllerSessionId())) { + return StaleReason.CONTROLLER_SESSION_MISMATCH; + } + return null; + } + + private static long oldestUnconvergedMs(ConvergenceStatus report) { + if (report == null || report.getUnconvergedSince() <= 0L) { + return 0L; + } + return Math.max(0L, System.currentTimeMillis() - report.getUnconvergedSince()); + } + + private static void putAssignment(ObjectNode target, Map assignment) { + int entryCount = 0; + for (Map.Entry entry : assignment.entrySet()) { + if (entryCount++ >= ConvergenceStatus.DEFAULT_MAX_ASSIGNMENT_ENTRIES) { + break; + } + putNullable(target, entry.getKey(), entry.getValue()); + } + } + + private static void putNullable(ObjectNode target, String fieldName, String value) { + if (value == null) { + target.putNull(fieldName); + } else { + target.put(fieldName, value); + } + } +} diff --git a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ResourceAccessor.java b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ResourceAccessor.java index cc4988569f..c39dae6ed3 100644 --- a/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ResourceAccessor.java +++ b/helix-rest/src/main/java/org/apache/helix/rest/server/resources/helix/ResourceAccessor.java @@ -45,12 +45,17 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.helix.ConfigAccessor; import org.apache.helix.HelixAdmin; +import org.apache.helix.HelixDataAccessor; import org.apache.helix.HelixException; +import org.apache.helix.PropertyKey; import org.apache.helix.PropertyPathBuilder; +import org.apache.helix.model.ClusterConfig; +import org.apache.helix.model.ConvergenceStatus; import org.apache.helix.model.CustomizedView; import org.apache.helix.model.ExternalView; import org.apache.helix.model.HelixConfigScope; import org.apache.helix.model.IdealState; +import org.apache.helix.model.LiveInstance; import org.apache.helix.model.ResourceConfig; import org.apache.helix.model.StateModelDefinition; import org.apache.helix.model.builder.HelixConfigScopeBuilder; @@ -171,6 +176,33 @@ public Response getPartitionHealth(@PathParam("clusterId") String clusterId, return JSONRepresentation(computePartitionHealth(clusterId, resourceName)); } + @ResponseMetered(name = HttpConstants.READ_REQUEST) + @Timed(name = HttpConstants.READ_REQUEST) + @GET + @Path("{resourceName}/convergence") + public Response getResourceConvergence(@PathParam("clusterId") String clusterId, + @PathParam("resourceName") String resourceName) { + HelixDataAccessor dataAccessor = getDataAccssor(clusterId); + PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder(); + ConvergenceStatus report = + dataAccessor.getProperty(keyBuilder.convergenceStatus(resourceName)); + ConvergenceStatus clusterReport = + dataAccessor.getProperty(keyBuilder.convergenceStatus()); + LiveInstance leader = dataAccessor.getProperty(keyBuilder.controllerLeader()); + ClusterConfig clusterConfig = dataAccessor.getProperty(keyBuilder.clusterConfig()); + boolean monitoringEnabled = + clusterConfig != null && clusterConfig.isConvergenceMonitoringEnabled(); + if (report == null + && ConvergenceStatusResponseMapper + .isFreshClusterReport(clusterReport, leader, monitoringEnabled)) { + return notFound(); + } + return JSONRepresentation( + ConvergenceStatusResponseMapper + .mapResource(clusterId, resourceName, report, clusterReport, leader, + monitoringEnabled)); + } + @ResponseMetered(name = HttpConstants.READ_REQUEST) @Timed(name = HttpConstants.READ_REQUEST) @GET diff --git a/helix-rest/src/test/java/org/apache/helix/rest/server/TestConvergenceStatusAccessor.java b/helix-rest/src/test/java/org/apache/helix/rest/server/TestConvergenceStatusAccessor.java new file mode 100644 index 0000000000..59f28eba85 --- /dev/null +++ b/helix-rest/src/test/java/org/apache/helix/rest/server/TestConvergenceStatusAccessor.java @@ -0,0 +1,148 @@ +package org.apache.helix.rest.server; + +/* + * 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 java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import javax.ws.rs.core.Response; + +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.helix.NotificationContext; +import org.apache.helix.TestHelper; +import org.apache.helix.integration.manager.MockParticipantManager; +import org.apache.helix.mock.participant.MockTransition; +import org.apache.helix.model.ClusterConfig; +import org.apache.helix.model.Message; +import org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestConvergenceStatusAccessor extends AbstractTestClass { + private static final String CLUSTER = "TestCluster_0"; + private static final String RESOURCE = "ConvergenceStatusTestDB"; + private static final long TEST_TIMEOUT_MS = 30_000L; + + @Test + public void testConvergenceEndpoint_transitionInFlight_reportsProgressThenConverges() + throws Exception { + List participants = _mockParticipantManagers.stream() + .filter(manager -> CLUSTER.equals(manager.getClusterName())).collect(Collectors.toList()); + Assert.assertFalse(participants.isEmpty()); + BlockingTransition transition = new BlockingTransition(); + participants.forEach(participant -> participant.setTransition(transition)); + ClusterConfig clusterConfig = _configAccessor.getClusterConfig(CLUSTER); + boolean monitoringPreviouslyEnabled = clusterConfig.isConvergenceMonitoringEnabled(); + clusterConfig.setConvergenceMonitoringEnabled(true); + _configAccessor.setClusterConfig(CLUSTER, clusterConfig); + + BestPossibleExternalViewVerifier verifier = null; + try { + addResource(CLUSTER, RESOURCE, 2, "MasterSlave", 1, 2); + verifier = new BestPossibleExternalViewVerifier.Builder(CLUSTER).setZkAddr(ZK_ADDR) + .setResources(java.util.Collections.singleton(RESOURCE)).build(); + + Assert.assertTrue(transition.awaitStarted(TEST_TIMEOUT_MS)); + Assert.assertTrue(TestHelper.verify( + () -> "IN_PROGRESS".equals(readStatus(resourcePath())), TEST_TIMEOUT_MS)); + Assert.assertTrue(TestHelper.verify( + () -> "IN_PROGRESS".equals(readStatus(clusterPath())), TEST_TIMEOUT_MS)); + String inProgressResponse = readResponse(resourcePath()); + Assert.assertTrue( + OBJECT_MAPPER.readTree(inProgressResponse).path("partitionSummary").path("affected") + .asInt() > 0); + System.out.println("CONVERGENCE_IN_PROGRESS=" + inProgressResponse); + + transition.release(); + Assert.assertTrue(verifier.verifyByPolling()); + Assert.assertTrue(TestHelper.verify( + () -> "CONVERGED".equals(readStatus(resourcePath())), TEST_TIMEOUT_MS)); + Assert.assertTrue(TestHelper.verify( + () -> "CONVERGED".equals(readStatus(clusterPath())), TEST_TIMEOUT_MS)); + String convergedResponse = readResponse(resourcePath()); + Assert.assertEquals( + OBJECT_MAPPER.readTree(convergedResponse).path("partitionSummary").path("affected") + .asInt(), 0); + System.out.println("CONVERGENCE_CONVERGED=" + convergedResponse); + + _gSetupTool.getClusterManagementTool().enableCluster(CLUSTER, false); + Assert.assertTrue(TestHelper.verify( + () -> "PAUSED".equals(readStatus(clusterPath())), TEST_TIMEOUT_MS)); + _gSetupTool.getClusterManagementTool().enableCluster(CLUSTER, true); + Assert.assertTrue(TestHelper.verify( + () -> "CONVERGED".equals(readStatus(clusterPath())), TEST_TIMEOUT_MS)); + } finally { + transition.release(); + participants.forEach(participant -> participant.setTransition(new MockTransition())); + if (verifier != null) { + verifier.close(); + } + if (_resourcesMap.get(CLUSTER).contains(RESOURCE)) { + _gSetupTool.dropResourceFromCluster(CLUSTER, RESOURCE); + _resourcesMap.get(CLUSTER).remove(RESOURCE); + } + clusterConfig.setConvergenceMonitoringEnabled(monitoringPreviouslyEnabled); + _configAccessor.setClusterConfig(CLUSTER, clusterConfig); + } + } + + private String readStatus(String path) { + String body = readResponse(path); + try { + JsonNode response = OBJECT_MAPPER.readTree(body); + return response.path("status").asText(); + } catch (Exception e) { + return ""; + } + } + + private String readResponse(String path) { + return get(path, null, Response.Status.OK.getStatusCode(), true); + } + + private String clusterPath() { + return "clusters/" + CLUSTER + "/convergence"; + } + + private String resourcePath() { + return "clusters/" + CLUSTER + "/resources/" + RESOURCE + "/convergence"; + } + + private static final class BlockingTransition extends MockTransition { + private final CountDownLatch _started = new CountDownLatch(1); + private final CountDownLatch _release = new CountDownLatch(1); + + @Override + public void doTransition(Message message, NotificationContext context) + throws InterruptedException { + _started.countDown(); + _release.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } + + private boolean awaitStarted(long timeoutMs) throws InterruptedException { + return _started.await(timeoutMs, TimeUnit.MILLISECONDS); + } + + private void release() { + _release.countDown(); + } + } +} diff --git a/helix-rest/src/test/java/org/apache/helix/rest/server/resources/helix/TestConvergenceStatusResponseMapper.java b/helix-rest/src/test/java/org/apache/helix/rest/server/resources/helix/TestConvergenceStatusResponseMapper.java new file mode 100644 index 0000000000..aabb1ebd79 --- /dev/null +++ b/helix-rest/src/test/java/org/apache/helix/rest/server/resources/helix/TestConvergenceStatusResponseMapper.java @@ -0,0 +1,117 @@ +package org.apache.helix.rest.server.resources.helix; + +/* + * 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 java.util.Collections; + +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.helix.model.ConvergenceStatus; +import org.apache.helix.model.ConvergenceStatus.OptimizerStatus; +import org.apache.helix.model.ConvergenceStatus.PartitionDetail; +import org.apache.helix.model.ConvergenceStatus.Reason; +import org.apache.helix.model.ConvergenceStatus.Scope; +import org.apache.helix.model.ConvergenceStatus.Status; +import org.apache.helix.model.ConvergenceStatus.TargetFreshness; +import org.apache.helix.model.LiveInstance; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestConvergenceStatusResponseMapper { + private static final String CLUSTER = "TestCluster"; + private static final String RESOURCE = "TestDB"; + private static final String SESSION = "abc123"; + + @Test + public void testMapResource_freshReport_returnsCustomerFields() { + ConvergenceStatus status = createReport(SESSION); + status.setPartitionDetails(Collections.singletonList( + new PartitionDetail(RESOURCE, "TestDB_0", Status.IN_PROGRESS, + Reason.PENDING_TRANSITION, Collections.singletonMap("host_0", "SLAVE"), + Collections.singletonMap("host_0", "MASTER"))), + ConvergenceStatus.DEFAULT_MAX_PARTITION_DETAILS); + LiveInstance leader = createLeader(SESSION); + + ObjectNode response = + ConvergenceStatusResponseMapper.mapResource(CLUSTER, RESOURCE, status, leader); + + Assert.assertEquals(response.path("status").asText(), Status.IN_PROGRESS.name()); + Assert.assertFalse(response.path("stale").asBoolean()); + Assert.assertEquals(response.path("targetFreshness").asText(), + TargetFreshness.CURRENT.name()); + Assert.assertEquals(response.path("optimizerStatus").asText(), + OptimizerStatus.HEALTHY.name()); + Assert.assertEquals( + response.path("affectedPartitions").get(0).path("currentAssignment").path("host_0") + .asText(), "SLAVE"); + Assert.assertEquals( + response.path("affectedPartitions").get(0).path("expectedAssignment").path("host_0") + .asText(), "MASTER"); + } + + @Test + public void testMapResource_controllerSessionChanged_returnsStaleUnknown() { + ConvergenceStatus status = createReport(SESSION); + LiveInstance leader = createLeader("different_session"); + + ObjectNode response = + ConvergenceStatusResponseMapper.mapResource(CLUSTER, RESOURCE, status, leader); + + Assert.assertEquals(response.path("status").asText(), Status.UNKNOWN.name()); + Assert.assertEquals(response.path("reportedStatus").asText(), Status.IN_PROGRESS.name()); + Assert.assertTrue(response.path("stale").asBoolean()); + Assert.assertEquals(response.path("staleReason").asText(), + "CONTROLLER_SESSION_MISMATCH"); + } + + @Test + public void testMapCluster_monitoringDisabled_returnsExplicitUnknown() { + ConvergenceStatus status = createReport(SESSION); + status.setScope(Scope.CLUSTER); + + ObjectNode response = ConvergenceStatusResponseMapper + .mapCluster(CLUSTER, status, createLeader(SESSION), false); + + Assert.assertEquals(response.path("status").asText(), Status.UNKNOWN.name()); + Assert.assertTrue(response.path("stale").asBoolean()); + Assert.assertEquals(response.path("staleReason").asText(), "MONITORING_DISABLED"); + } + + private ConvergenceStatus createReport(String session) { + ConvergenceStatus status = new ConvergenceStatus(RESOURCE); + status.setScope(Scope.RESOURCE); + status.setResourceName(RESOURCE); + status.setStatus(Status.IN_PROGRESS); + status.setPrimaryReason(Reason.PENDING_TRANSITION); + status.setTargetFreshness(TargetFreshness.CURRENT); + status.setOptimizerStatus(OptimizerStatus.HEALTHY); + status.setControllerSessionId(session); + status.setGeneratedAt(System.currentTimeMillis()); + status.setTotalPartitionCount(1); + status.setInProgressPartitionCount(1); + status.setAffectedPartitionCount(1); + return status; + } + + private LiveInstance createLeader(String session) { + LiveInstance leader = new LiveInstance("controller"); + leader.setSessionId(session); + return leader; + } +}