| package org.apache.helix.controller.rebalancer.waged.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.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.stream.Collectors; |
| |
| import org.apache.helix.HelixConstants; |
| import org.apache.helix.HelixException; |
| import org.apache.helix.controller.dataproviders.ResourceControllerDataProvider; |
| import org.apache.helix.controller.rebalancer.util.DelayedRebalanceUtil; |
| import org.apache.helix.model.ClusterConfig; |
| import org.apache.helix.model.ClusterTopologyConfig; |
| import org.apache.helix.model.IdealState; |
| import org.apache.helix.model.InstanceConfig; |
| import org.apache.helix.model.Partition; |
| import org.apache.helix.model.Resource; |
| import org.apache.helix.model.ResourceAssignment; |
| import org.apache.helix.model.ResourceConfig; |
| import org.apache.helix.model.StateModelDefinition; |
| |
| /** |
| * This util class generates Cluster Model object based on the controller's data cache. |
| */ |
| public class ClusterModelProvider { |
| |
| private enum RebalanceScopeType { |
| // Set the rebalance scope to cover the difference between the current assignment and the |
| // Baseline assignment only. |
| PARTIAL, |
| // Set the rebalance scope to cover all replicas that need relocation based on the cluster |
| // changes. |
| GLOBAL_BASELINE, |
| // Set the rebalance scope to cover only replicas that are assigned to downed instances. |
| EMERGENCY, |
| // A temporary overwrites for partition replicas on downed instance but still within the delayed window but missing |
| // minActiveReplicas |
| DELAYED_REBALANCE_OVERWRITES |
| } |
| |
| /** |
| * TODO: On integration with WAGED, have to integrate with counter and latency metrics -- qqu |
| * Compute a new Cluster Model with scope limited to partitions with best possible assignment missing minActiveReplicas |
| * because of delayed rebalance setting. |
| * @param dataProvider The controller's data cache |
| * @param resourceMap The full map of the resource by name |
| * @param activeInstances The active instances that will be used in the calculation. |
| * @param resourceAssignment The resource assignment state to compute on. This should be the current state assignment; |
| * if it's run right after another rebalance calculation, the best possible assignment from |
| * previous result can be used. |
| * @return the ClusterModel |
| */ |
| public static ClusterModel generateClusterModelForDelayedRebalanceOverwrites( |
| ResourceControllerDataProvider dataProvider, |
| Map<String, Resource> resourceMap, |
| Set<String> activeInstances, |
| Map<String, ResourceAssignment> resourceAssignment) { |
| return generateClusterModel(dataProvider, resourceMap, activeInstances, Collections.emptyMap(), |
| Collections.emptyMap(), resourceAssignment, |
| RebalanceScopeType.DELAYED_REBALANCE_OVERWRITES); |
| } |
| |
| /** |
| * Generate a new Cluster Model object according to the current cluster status for emergency |
| * rebalance. The rebalance scope is configured for recovering replicas that are on permanently |
| * downed nodes |
| * @param dataProvider The controller's data cache. |
| * @param resourceMap The full list of the resources to be rebalanced. Note that any |
| * resources that are not in this list will be removed from the |
| * final assignment. |
| * @param activeInstances The active instances that will be used in the calculation. |
| * Note this list can be different from the real active node list |
| * according to the rebalancer logic. |
| * @param bestPossibleAssignment The persisted Best Possible assignment that was generated in the |
| * previous rebalance. |
| * @return the new cluster model |
| */ |
| public static ClusterModel generateClusterModelForEmergencyRebalance(ResourceControllerDataProvider dataProvider, |
| Map<String, Resource> resourceMap, Set<String> activeInstances, |
| Map<String, ResourceAssignment> bestPossibleAssignment) { |
| return generateClusterModel(dataProvider, resourceMap, activeInstances, Collections.emptyMap(), |
| Collections.emptyMap(), bestPossibleAssignment, RebalanceScopeType.EMERGENCY); |
| } |
| |
| /** |
| * Generate a new Cluster Model object according to the current cluster status for partial |
| * rebalance. The rebalance scope is configured for recovering the missing replicas that are in |
| * the Baseline assignment but not in the current Best possible assignment only. |
| * @param dataProvider The controller's data cache. |
| * @param resourceMap The full list of the resources to be rebalanced. Note that any |
| * resources that are not in this list will be removed from the |
| * final assignment. |
| * @param activeInstances The active instances that will be used in the calculation. |
| * Note this list can be different from the real active node list |
| * according to the rebalancer logic. |
| * @param baselineAssignment The persisted Baseline assignment. |
| * @param bestPossibleAssignment The persisted Best Possible assignment that was generated in the |
| * previous rebalance. |
| * @return the new cluster model |
| */ |
| public static ClusterModel generateClusterModelForPartialRebalance( |
| ResourceControllerDataProvider dataProvider, Map<String, Resource> resourceMap, |
| Set<String> activeInstances, Map<String, ResourceAssignment> baselineAssignment, |
| Map<String, ResourceAssignment> bestPossibleAssignment) { |
| return generateClusterModel(dataProvider, resourceMap, activeInstances, Collections.emptyMap(), |
| baselineAssignment, bestPossibleAssignment, RebalanceScopeType.PARTIAL); |
| } |
| |
| /** |
| * Generate a new Cluster Model object according to the current cluster status for the Baseline |
| * calculation. The rebalance scope is determined according to the cluster changes. |
| * @param dataProvider The controller's data cache. |
| * @param resourceMap The full list of the resources to be rebalanced. Note that any |
| * resources that are not in this list will be removed from the |
| * final assignment. |
| * @param allInstances All the instances that will be used in the calculation. |
| * @param clusterChanges All the cluster changes that happened after the previous rebalance. |
| * @param baselineAssignment The previous Baseline assignment. |
| * @return the new cluster model |
| */ |
| public static ClusterModel generateClusterModelForBaseline( |
| ResourceControllerDataProvider dataProvider, Map<String, Resource> resourceMap, |
| Set<String> allInstances, Map<HelixConstants.ChangeType, Set<String>> clusterChanges, |
| Map<String, ResourceAssignment> baselineAssignment) { |
| return generateClusterModel(dataProvider, resourceMap, allInstances, clusterChanges, |
| Collections.emptyMap(), baselineAssignment, RebalanceScopeType.GLOBAL_BASELINE); |
| } |
| |
| /** |
| * Generate a cluster model based on the current state output and data cache. The rebalance scope |
| * is configured for recovering the missing replicas only. |
| * @param dataProvider The controller's data cache. |
| * @param resourceMap The full list of the resources to be rebalanced. Note that any |
| * resources that are not in this list will be removed from the |
| * final assignment. |
| * @param currentStateAssignment The resource assignment built from current state output. |
| * @return the new cluster model |
| */ |
| public static ClusterModel generateClusterModelFromExistingAssignment( |
| ResourceControllerDataProvider dataProvider, Map<String, Resource> resourceMap, |
| Map<String, ResourceAssignment> currentStateAssignment) { |
| return generateClusterModel(dataProvider, resourceMap, |
| dataProvider.getEnabledLiveInstances(), Collections.emptyMap(), |
| Collections.emptyMap(), currentStateAssignment, |
| RebalanceScopeType.GLOBAL_BASELINE); |
| } |
| |
| /** |
| * Generate a new Cluster Model object according to the current cluster status. |
| * @param dataProvider The controller's data cache. |
| * @param resourceMap The full list of the resources to be rebalanced. Note that any |
| * resources that are not in this list will be removed from the |
| * final assignment. |
| * @param activeInstances The active instances that will be used in the calculation. |
| * Note this list can be different from the real active node list |
| * according to the rebalancer logic. |
| * @param clusterChanges All the cluster changes that happened after the previous rebalance. |
| * @param idealAssignment The ideal assignment. |
| * @param currentAssignment The current assignment that was generated in the previous rebalance. |
| * @param scopeType Specify how to determine the rebalance scope. |
| * @return the new cluster model |
| */ |
| private static ClusterModel generateClusterModel(ResourceControllerDataProvider dataProvider, |
| Map<String, Resource> resourceMap, Set<String> activeInstances, |
| Map<HelixConstants.ChangeType, Set<String>> clusterChanges, |
| Map<String, ResourceAssignment> idealAssignment, |
| Map<String, ResourceAssignment> currentAssignment, RebalanceScopeType scopeType) { |
| Map<String, InstanceConfig> assignableInstanceConfigMap = dataProvider.getAssignableInstanceConfigMap(); |
| // Construct all the assignable nodes and initialize with the allocated replicas. |
| Set<AssignableNode> assignableNodes = |
| getAllAssignableNodes(dataProvider.getClusterConfig(), assignableInstanceConfigMap, |
| activeInstances); |
| |
| // Generate the logical view of the ideal assignment and the current assignment. |
| ClusterTopologyConfig clusterTopologyConfig = |
| ClusterTopologyConfig.createFromClusterConfig(dataProvider.getClusterConfig()); |
| Map<String, ResourceAssignment> logicalIdIdealAssignment = |
| idealAssignment.isEmpty() ? idealAssignment |
| : generateResourceAssignmentMapLogicalIdView(idealAssignment, clusterTopologyConfig, |
| dataProvider); |
| Map<String, ResourceAssignment> logicalIdCurrentAssignment = |
| currentAssignment.isEmpty() ? currentAssignment |
| : generateResourceAssignmentMapLogicalIdView(currentAssignment, clusterTopologyConfig, |
| dataProvider); |
| |
| // Get the set of active logical ids. |
| Set<String> activeLogicalIds = activeInstances.stream().map( |
| instanceName -> assignableInstanceConfigMap.getOrDefault(instanceName, |
| new InstanceConfig(instanceName)) |
| .getLogicalId(clusterTopologyConfig.getEndNodeType())).collect(Collectors.toSet()); |
| |
| // TODO: Figure out why streaming the keySet directly in rare cases causes ConcurrentModificationException |
| // In theory, this should not be happening since cache refresh is at beginning of the pipeline, so could be some other reason. |
| // For now, we just copy the keySet to a new HashSet to avoid the exception. |
| Set<String> assignableLiveInstanceNames = new HashSet<>(dataProvider.getAssignableLiveInstances().keySet()); |
| Set<String> assignableLiveInstanceLogicalIds = |
| assignableLiveInstanceNames.stream().map( |
| instanceName -> assignableInstanceConfigMap.getOrDefault(instanceName, |
| new InstanceConfig(instanceName)) |
| .getLogicalId(clusterTopologyConfig.getEndNodeType())).collect(Collectors.toSet()); |
| |
| // Generate replica objects for all the resource partitions. |
| // <resource, replica set> |
| Map<String, Set<AssignableReplica>> replicaMap = |
| getAllAssignableReplicas(dataProvider, resourceMap, assignableNodes); |
| |
| // Check if the replicas need to be reassigned. |
| Map<String, Set<AssignableReplica>> allocatedReplicas = |
| new HashMap<>(); // <instanceName, replica set> |
| Set<AssignableReplica> toBeAssignedReplicas; |
| switch (scopeType) { |
| case GLOBAL_BASELINE: |
| toBeAssignedReplicas = |
| findToBeAssignedReplicasByClusterChanges(replicaMap, activeLogicalIds, |
| assignableLiveInstanceLogicalIds, clusterChanges, logicalIdCurrentAssignment, |
| allocatedReplicas); |
| break; |
| case PARTIAL: |
| // Filter to remove the replicas that do not exist in the ideal assignment given but exist |
| // in the replicaMap. This is because such replicas are new additions that do not need to be |
| // rebalanced right away. |
| retainExistingReplicas(replicaMap, logicalIdIdealAssignment); |
| toBeAssignedReplicas = |
| findToBeAssignedReplicasByComparingWithIdealAssignment(replicaMap, activeLogicalIds, |
| logicalIdIdealAssignment, logicalIdCurrentAssignment, allocatedReplicas); |
| break; |
| case EMERGENCY: |
| toBeAssignedReplicas = findToBeAssignedReplicasOnDownInstances(replicaMap, activeLogicalIds, |
| logicalIdCurrentAssignment, allocatedReplicas); |
| break; |
| case DELAYED_REBALANCE_OVERWRITES: |
| toBeAssignedReplicas = |
| DelayedRebalanceUtil.findToBeAssignedReplicasForMinActiveReplica(dataProvider, replicaMap.keySet(), |
| activeLogicalIds, logicalIdCurrentAssignment, allocatedReplicas); |
| break; |
| default: |
| throw new HelixException("Unknown rebalance scope type: " + scopeType); |
| } |
| |
| // Update the allocated replicas to the assignable nodes. |
| assignableNodes.parallelStream().forEach(node -> node.assignInitBatch( |
| allocatedReplicas.getOrDefault(node.getLogicalId(), Collections.emptySet()))); |
| |
| // Construct and initialize cluster context. |
| ClusterContext context = new ClusterContext( |
| replicaMap.values().stream().flatMap(Set::stream).collect(Collectors.toSet()), |
| assignableNodes, logicalIdIdealAssignment, logicalIdCurrentAssignment, dataProvider.getClusterConfig()); |
| |
| // Initial the cluster context with the allocated assignments. |
| context.setAssignmentForFaultZoneMap(mapAssignmentToFaultZone(assignableNodes)); |
| |
| return new ClusterModel(context, toBeAssignedReplicas, assignableNodes); |
| } |
| |
| private static Map<String, ResourceAssignment> generateResourceAssignmentMapLogicalIdView( |
| Map<String, ResourceAssignment> resourceAssignmentMap, |
| ClusterTopologyConfig clusterTopologyConfig, ResourceControllerDataProvider dataProvider) { |
| |
| Map<String, InstanceConfig> allInstanceConfigMap = dataProvider.getInstanceConfigMap(); |
| |
| return resourceAssignmentMap.entrySet().parallelStream() |
| .collect(Collectors.toMap(Map.Entry::getKey, entry -> { |
| String resourceName = entry.getKey(); |
| ResourceAssignment instanceNameResourceAssignment = entry.getValue(); |
| ResourceAssignment logicalIdResourceAssignment = new ResourceAssignment(resourceName); |
| |
| StateModelDefinition stateModelDefinition = dataProvider.getStateModelDef( |
| dataProvider.getIdealState(resourceName).getStateModelDefRef()); |
| |
| instanceNameResourceAssignment.getMappedPartitions().forEach(partition -> { |
| Map<String, String> logicalIdStateMap = new HashMap<>(); |
| |
| instanceNameResourceAssignment.getReplicaMap(partition) |
| .forEach((instanceName, state) -> { |
| if (allInstanceConfigMap.containsKey(instanceName)) { |
| String logicalId = allInstanceConfigMap.get(instanceName) |
| .getLogicalId(clusterTopologyConfig.getEndNodeType()); |
| if (!logicalIdStateMap.containsKey(logicalId) || state.equals( |
| stateModelDefinition.getTopState())) { |
| logicalIdStateMap.put(logicalId, state); |
| } |
| } |
| }); |
| |
| logicalIdResourceAssignment.addReplicaMap(partition, logicalIdStateMap); |
| }); |
| |
| return logicalIdResourceAssignment; |
| })); |
| } |
| |
| // Filter the replicas map so only the replicas that have been allocated in the existing |
| // assignmentMap remain in the map. |
| private static void retainExistingReplicas(Map<String, Set<AssignableReplica>> replicaMap, |
| Map<String, ResourceAssignment> assignmentMap) { |
| replicaMap.entrySet().parallelStream().forEach(replicaSetEntry -> { |
| // <partition, <state, instances set>> |
| Map<String, Map<String, Set<String>>> stateInstanceMap = |
| getStateInstanceMap(assignmentMap.get(replicaSetEntry.getKey())); |
| // Iterate the replicas of the resource to find the ones that require reallocating. |
| Iterator<AssignableReplica> replicaIter = replicaSetEntry.getValue().iterator(); |
| while (replicaIter.hasNext()) { |
| AssignableReplica replica = replicaIter.next(); |
| Set<String> validInstances = |
| stateInstanceMap.getOrDefault(replica.getPartitionName(), Collections.emptyMap()) |
| .getOrDefault(replica.getReplicaState(), Collections.emptySet()); |
| if (validInstances.isEmpty()) { |
| // Removing the replica if it is not known in the assignment map. |
| replicaIter.remove(); |
| } else { |
| // Remove the instance from the state map record after processing so it won't be |
| // double-processed as we loop through all replica |
| validInstances.remove(validInstances.iterator().next()); |
| } |
| } |
| }); |
| } |
| |
| /** |
| * Find the minimum set of replicas that need to be reassigned by comparing the current assignment |
| * with the ideal assignment. |
| * A replica needs to be reassigned or newly assigned if either of the following conditions is true: |
| * 1. The partition allocation (the instance the replica is placed on) in the ideal assignment and |
| * the current assignment are different. And the allocation in the ideal assignment is valid. |
| * So it is worthwhile to move it. |
| * 2. The partition allocation is in neither the ideal assignment nor the current assignment. Or |
| * those allocations are not valid due to offline or disabled instances. |
| * Otherwise, the rebalancer just keeps the current assignment allocation. |
| * |
| * @param replicaMap A map contains all the replicas grouped by resource name. |
| * @param activeInstances All the instances that are live and enabled according to the delay rebalance configuration. |
| * @param idealAssignment The ideal assignment. |
| * @param currentAssignment The current assignment that was generated in the previous rebalance. |
| * @param allocatedReplicas A map of <Instance -> replicas> to return the allocated replicas grouped by the target instance name. |
| * @return The replicas that need to be reassigned. |
| */ |
| private static Set<AssignableReplica> findToBeAssignedReplicasByComparingWithIdealAssignment( |
| Map<String, Set<AssignableReplica>> replicaMap, Set<String> activeInstances, |
| Map<String, ResourceAssignment> idealAssignment, |
| Map<String, ResourceAssignment> currentAssignment, |
| Map<String, Set<AssignableReplica>> allocatedReplicas) { |
| Set<AssignableReplica> toBeAssignedReplicas = new HashSet<>(); |
| // check each resource to identify the allocated replicas and to-be-assigned replicas. |
| for (String resourceName : replicaMap.keySet()) { |
| // <partition, <state, instances set>> |
| Map<String, Map<String, Set<String>>> idealPartitionStateMap = |
| getValidStateInstanceMap(idealAssignment.get(resourceName), activeInstances); |
| Map<String, Map<String, Set<String>>> currentPartitionStateMap = |
| getValidStateInstanceMap(currentAssignment.get(resourceName), activeInstances); |
| // Iterate the replicas of the resource to find the ones that require reallocating. |
| for (AssignableReplica replica : replicaMap.get(resourceName)) { |
| String partitionName = replica.getPartitionName(); |
| String replicaState = replica.getReplicaState(); |
| Set<String> idealAllocations = |
| idealPartitionStateMap.getOrDefault(partitionName, Collections.emptyMap()) |
| .getOrDefault(replicaState, Collections.emptySet()); |
| Set<String> currentAllocations = |
| currentPartitionStateMap.getOrDefault(partitionName, Collections.emptyMap()) |
| .getOrDefault(replicaState, Collections.emptySet()); |
| |
| // Compare the current assignments with the ideal assignment for the common part. |
| List<String> commonAllocations = new ArrayList<>(currentAllocations); |
| commonAllocations.retainAll(idealAllocations); |
| if (!commonAllocations.isEmpty()) { |
| // 1. If the partition is allocated at the same location in both ideal and current |
| // assignments, there is no need to reassign it. |
| String allocatedInstance = commonAllocations.get(0); |
| allocatedReplicas.computeIfAbsent(allocatedInstance, key -> new HashSet<>()).add(replica); |
| // Remove the instance from the record to prevent this instance from being processed twice. |
| idealAllocations.remove(allocatedInstance); |
| currentAllocations.remove(allocatedInstance); |
| } else if (!idealAllocations.isEmpty()) { |
| // 2. If the partition is allocated at an active instance in the ideal assignment but the |
| // same allocation does not exist in the current assignment, try to rebalance the replica |
| // or assign it if the replica has not been assigned. |
| // There are two possible conditions, |
| // * This replica has been newly added and has not been assigned yet, so it appears in |
| // the ideal assignment and does not appear in the current assignment. |
| // * The allocation of this replica in the ideal assignment has been updated due to a |
| // cluster change. For example, new instance is added. So the old allocation in the |
| // current assignment might be sub-optimal. |
| // In either condition, we add it to toBeAssignedReplicas so that it will get assigned. |
| toBeAssignedReplicas.add(replica); |
| // Remove the pending allocation from the idealAllocations after processing so that the |
| // instance won't be double-processed as we loop through all replicas |
| String pendingAllocation = idealAllocations.iterator().next(); |
| idealAllocations.remove(pendingAllocation); |
| } else if (!currentAllocations.isEmpty()) { |
| // 3. This replica exists in the current assignment but does not appear or has a valid |
| // allocation in the ideal assignment. |
| // This means either 1) that the ideal assignment actually has this replica allocated on |
| // this instance, but it does not show up because the instance is temporarily offline or |
| // disabled (note that all such instances have been filtered out in earlier part of the |
| // logic) or that the most recent version of the ideal assignment was not fetched |
| // correctly from the assignment metadata store. |
| // In either case, the solution is to keep the current assignment. So put this replica |
| // with the allocated instance into the allocatedReplicas map. |
| String allocatedInstance = currentAllocations.iterator().next(); |
| allocatedReplicas.computeIfAbsent(allocatedInstance, key -> new HashSet<>()).add(replica); |
| // Remove the instance from the record to prevent the same location being processed again. |
| currentAllocations.remove(allocatedInstance); |
| } else { |
| // 4. This replica is not found in either the ideal assignment or the current assignment |
| // with a valid allocation. This implies that the replica was newly added but was never |
| // assigned in reality or was added so recently that it hasn't shown up in the ideal |
| // assignment (because it's calculation takes longer and is asynchronously calculated). |
| // In that case, we add it to toBeAssignedReplicas so that it will get assigned as a |
| // result of partialRebalance. |
| toBeAssignedReplicas.add(replica); |
| } |
| } |
| } |
| return toBeAssignedReplicas; |
| } |
| |
| /** |
| * Find the minimum set of replicas that need to be reassigned according to the cluster change. |
| * A replica needs to be reassigned if one of the following condition is true: |
| * 1. Cluster topology (the cluster config / any instance config) has been updated. |
| * 2. The resource config has been updated. |
| * 3. If the current assignment does not contain the partition's valid assignment. |
| * |
| * @param replicaMap A map contains all the replicas grouped by resource name. |
| * @param activeInstances All the instances that are live and enabled according to the delay rebalance configuration. |
| * @param liveInstances All the instances that are live. |
| * @param clusterChanges A map that contains all the important metadata updates that happened after the previous rebalance. |
| * @param currentAssignment The current replica assignment. |
| * @param allocatedReplicas Return the allocated replicas grouped by the target instance name. |
| * @return The replicas that need to be reassigned. |
| */ |
| private static Set<AssignableReplica> findToBeAssignedReplicasByClusterChanges( |
| Map<String, Set<AssignableReplica>> replicaMap, Set<String> activeInstances, |
| Set<String> liveInstances, Map<HelixConstants.ChangeType, Set<String>> clusterChanges, |
| Map<String, ResourceAssignment> currentAssignment, |
| Map<String, Set<AssignableReplica>> allocatedReplicas) { |
| Set<AssignableReplica> toBeAssignedReplicas = new HashSet<>(); |
| |
| // A newly connected node = A new LiveInstance znode (or session Id updated) & the |
| // corresponding instance is live. |
| // TODO: The assumption here is that if the LiveInstance znode is created or it's session Id is |
| // TODO: updated, we need to call algorithm for moving some partitions to this new node. |
| // TODO: However, if the liveInstance znode is changed because of some other reason, it will be |
| // TODO: treated as a newly connected nodes. We need to find a better way to identify which one |
| // TODO: is the real newly connected nodes. |
| Set<String> newlyConnectedNodes = clusterChanges |
| .getOrDefault(HelixConstants.ChangeType.LIVE_INSTANCE, Collections.emptySet()); |
| newlyConnectedNodes.retainAll(liveInstances); |
| |
| if (clusterChanges.containsKey(HelixConstants.ChangeType.CLUSTER_CONFIG) |
| || clusterChanges.containsKey(HelixConstants.ChangeType.INSTANCE_CONFIG) |
| || !newlyConnectedNodes.isEmpty()) { |
| |
| // 1. If the cluster topology has been modified, need to reassign all replicas. |
| // 2. If any node was newly connected, need to rebalance all replicas for the evenness of |
| // distribution. |
| toBeAssignedReplicas |
| .addAll(replicaMap.values().stream().flatMap(Set::stream).collect(Collectors.toSet())); |
| } else { |
| // check each resource to identify the allocated replicas and to-be-assigned replicas. |
| for (Map.Entry<String, Set<AssignableReplica>> replicaMapEntry : replicaMap.entrySet()) { |
| String resourceName = replicaMapEntry.getKey(); |
| Set<AssignableReplica> replicas = replicaMapEntry.getValue(); |
| // 1. if the resource config/idealstate is changed, need to reassign. |
| // 2. if the resource does not appear in the current assignment, need to reassign. |
| if (clusterChanges |
| .getOrDefault(HelixConstants.ChangeType.RESOURCE_CONFIG, Collections.emptySet()) |
| .contains(resourceName) || clusterChanges |
| .getOrDefault(HelixConstants.ChangeType.IDEAL_STATE, Collections.emptySet()) |
| .contains(resourceName) || !currentAssignment.containsKey(resourceName)) { |
| toBeAssignedReplicas.addAll(replicas); |
| // go to check next resource |
| } else { |
| // check for every replica assignment to identify if the related replicas need to be reassigned. |
| // <partition, <state, instances list>> |
| Map<String, Map<String, Set<String>>> stateMap = |
| getValidStateInstanceMap(currentAssignment.get(resourceName), activeInstances); |
| for (AssignableReplica replica : replicas) { |
| // Find any ACTIVE instance allocation that has the same state with the replica |
| Set<String> validInstances = |
| stateMap.getOrDefault(replica.getPartitionName(), Collections.emptyMap()) |
| .getOrDefault(replica.getReplicaState(), Collections.emptySet()); |
| if (validInstances.isEmpty()) { |
| // 3. if no such an instance in the current assignment, need to reassign the replica |
| toBeAssignedReplicas.add(replica); |
| } else { |
| Iterator<String> iter = validInstances.iterator(); |
| // Remove the instance from the current allocation record after processing so that it |
| // won't be double-processed as we loop through all replicas |
| String logicalId = iter.next(); |
| iter.remove(); |
| // the current assignment for this replica is valid, |
| // add to the allocated replica list. |
| allocatedReplicas.computeIfAbsent(logicalId, key -> new HashSet<>()).add(replica); |
| } |
| } |
| } |
| } |
| } |
| return toBeAssignedReplicas; |
| } |
| |
| /** |
| * Find replicas that were assigned to non-active nodes in the current assignment. |
| * |
| * @param replicaMap A map contains all the replicas grouped by resource name. |
| * @param activeInstances All the instances that are live and enabled according to the delay rebalance configuration. |
| * @param currentAssignment The current assignment that was generated in the previous rebalance. |
| * @param allocatedReplicas A map of <Instance -> replicas> to return the allocated replicas grouped by the target instance name. |
| * @return The replicas that need to be reassigned. |
| */ |
| private static Set<AssignableReplica> findToBeAssignedReplicasOnDownInstances( |
| Map<String, Set<AssignableReplica>> replicaMap, Set<String> activeInstances, |
| Map<String, ResourceAssignment> currentAssignment, |
| Map<String, Set<AssignableReplica>> allocatedReplicas) { |
| // For any replica that are assigned to non-active instances (down instances), add them. |
| Set<AssignableReplica> toBeAssignedReplicas = new HashSet<>(); |
| for (String resourceName : replicaMap.keySet()) { |
| Map<String, Map<String, Set<String>>> stateInstanceMap = getStateInstanceMap(currentAssignment.get(resourceName)); |
| |
| for (AssignableReplica replica : replicaMap.get(resourceName)) { |
| String partitionName = replica.getPartitionName(); |
| String replicaState = replica.getReplicaState(); |
| Set<String> currentAllocations = |
| stateInstanceMap.getOrDefault(partitionName, Collections.emptyMap()) |
| .getOrDefault(replicaState, Collections.emptySet()); |
| if (!currentAllocations.isEmpty()) { |
| String allocatedInstance = currentAllocations.iterator().next(); |
| if (activeInstances.contains(allocatedInstance)) { |
| allocatedReplicas.computeIfAbsent(allocatedInstance, key -> new HashSet<>()).add(replica); |
| } |
| else { |
| toBeAssignedReplicas.add(replica); |
| } |
| currentAllocations.remove(allocatedInstance); |
| } |
| } |
| } |
| |
| return toBeAssignedReplicas; |
| } |
| |
| /** |
| * Filter to remove all invalid allocations that are not on the active instances. |
| * @param assignment |
| * @param activeInstances |
| * @return A map of <partition, <state, instances set>> contains the valid state to instance map. |
| */ |
| private static Map<String, Map<String, Set<String>>> getValidStateInstanceMap( |
| ResourceAssignment assignment, Set<String> activeInstances) { |
| Map<String, Map<String, Set<String>>> stateInstanceMap = getStateInstanceMap(assignment); |
| stateInstanceMap.values().forEach(stateMap -> |
| stateMap.values().forEach(instanceSet -> instanceSet.retainAll(activeInstances))); |
| return stateInstanceMap; |
| } |
| |
| // <partition, <state, instances set>> |
| public static Map<String, Map<String, Set<String>>> getStateInstanceMap( |
| ResourceAssignment assignment) { |
| if (assignment == null) { |
| return Collections.emptyMap(); |
| } |
| return assignment.getMappedPartitions().stream() |
| .collect(Collectors.toMap(Partition::getPartitionName, partition -> { |
| Map<String, Set<String>> stateInstanceMap = new HashMap<>(); |
| assignment.getReplicaMap(partition) |
| .forEach((key1, value) -> stateInstanceMap.computeIfAbsent(value, key -> new HashSet<>()).add(key1)); |
| return stateInstanceMap; |
| })); |
| } |
| |
| /** |
| * Get all the nodes that can be assigned replicas based on the configurations. |
| * |
| * @param clusterConfig The cluster configuration. |
| * @param instanceConfigMap A map of all the instance configuration. |
| * If any active instance has no configuration, it will be ignored. |
| * @param activeInstances All the instances that are online and enabled. |
| * @return A map of assignable node set, <InstanceName, node set>. |
| */ |
| private static Set<AssignableNode> getAllAssignableNodes(ClusterConfig clusterConfig, |
| Map<String, InstanceConfig> instanceConfigMap, Set<String> activeInstances) { |
| ClusterTopologyConfig clusterTopologyConfig = |
| ClusterTopologyConfig.createFromClusterConfig(clusterConfig); |
| return activeInstances.parallelStream() |
| .filter(instanceConfigMap::containsKey).map( |
| instanceName -> new AssignableNode(clusterConfig, clusterTopologyConfig, |
| instanceConfigMap.get(instanceName), |
| instanceName)).collect(Collectors.toSet()); |
| } |
| |
| /** |
| * Get all the replicas that need to be reallocated from the cluster data cache. |
| * |
| * @param dataProvider The cluster status cache that contains the current cluster status. |
| * @param resourceMap All the valid resources that are managed by the rebalancer. |
| * @param assignableNodes All the active assignable nodes. |
| * @return A map of assignable replica set, <ResourceName, replica set>. |
| */ |
| private static Map<String, Set<AssignableReplica>> getAllAssignableReplicas( |
| ResourceControllerDataProvider dataProvider, Map<String, Resource> resourceMap, |
| Set<AssignableNode> assignableNodes) { |
| ClusterConfig clusterConfig = dataProvider.getClusterConfig(); |
| int activeFaultZoneCount = assignableNodes.stream().map(AssignableNode::getFaultZone) |
| .collect(Collectors.toSet()).size(); |
| return resourceMap.keySet().parallelStream().map(resourceName -> { |
| ResourceConfig resourceConfig = dataProvider.getResourceConfig(resourceName); |
| if (resourceConfig == null) { |
| resourceConfig = new ResourceConfig(resourceName); |
| } |
| IdealState is = dataProvider.getIdealState(resourceName); |
| if (is == null) { |
| throw new HelixException( |
| "Cannot find the resource ideal state for resource: " + resourceName); |
| } |
| String defName = is.getStateModelDefRef(); |
| StateModelDefinition def = dataProvider.getStateModelDef(defName); |
| if (def == null) { |
| throw new IllegalArgumentException(String |
| .format("Cannot find state model definition %s for resource %s.", |
| is.getStateModelDefRef(), resourceName)); |
| } |
| Map<String, Integer> stateCountMap = |
| def.getStateCountMap(activeFaultZoneCount, is.getReplicaCount(assignableNodes.size())); |
| ResourceConfig mergedResourceConfig = |
| ResourceConfig.mergeIdealStateWithResourceConfig(resourceConfig, is); |
| Set<AssignableReplica> replicas = new HashSet<>(); |
| for (String partition : is.getPartitionSet()) { |
| for (Map.Entry<String, Integer> entry : stateCountMap.entrySet()) { |
| String state = entry.getKey(); |
| for (int i = 0; i < entry.getValue(); i++) { |
| replicas.add(new AssignableReplica(clusterConfig, mergedResourceConfig, partition, state, |
| def.getStatePriorityMap().get(state))); |
| } |
| } |
| } |
| return new HashMap.SimpleEntry<>(resourceName, replicas); |
| }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); |
| } |
| |
| /** |
| * @return A map containing the assignments for each fault zone. <fault zone, <resource, set of partitions>> |
| */ |
| private static Map<String, Map<String, Set<String>>> mapAssignmentToFaultZone( |
| Set<AssignableNode> assignableNodes) { |
| Map<String, Map<String, Set<String>>> faultZoneAssignmentMap = new HashMap<>(); |
| assignableNodes.forEach(node -> { |
| for (Map.Entry<String, Set<String>> resourceMap : node.getAssignedPartitionsMap() |
| .entrySet()) { |
| faultZoneAssignmentMap.computeIfAbsent(node.getFaultZone(), k -> new HashMap<>()) |
| .computeIfAbsent(resourceMap.getKey(), k -> new HashSet<>()) |
| .addAll(resourceMap.getValue()); |
| } |
| }); |
| return faultZoneAssignmentMap; |
| } |
| } |
