| /** |
| * 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. |
| */ |
| package org.apache.hadoop.hdfs.server.blockmanagement; |
| |
| import org.apache.hadoop.thirdparty.com.google.common.collect.Iterables; |
| import org.apache.hadoop.hdfs.DFSConfigKeys; |
| import org.apache.hadoop.hdfs.server.namenode.INode; |
| import org.apache.hadoop.hdfs.server.namenode.INodeFile; |
| import org.apache.hadoop.hdfs.server.namenode.INodeId; |
| import org.apache.hadoop.hdfs.util.LightWeightHashSet; |
| import org.apache.hadoop.hdfs.util.LightWeightLinkedSet; |
| import org.apache.hadoop.classification.VisibleForTesting; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import java.util.HashMap; |
| import java.util.ArrayList; |
| import java.util.Map; |
| import java.util.List; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.stream.Collectors; |
| |
| /** |
| * This class implements the logic to track decommissioning and entering |
| * maintenance nodes, ensure all their blocks are adequately replicated |
| * before they are moved to the decommissioned or maintenance state. |
| * |
| * This monitor avoids flooding the replication queue with all pending blocks |
| * and instead feeds them to the queue as the prior set complete replication. |
| * |
| * HDFS-14854 contains details about the overall design of this class. |
| * |
| */ |
| public class DatanodeAdminBackoffMonitor extends DatanodeAdminMonitorBase |
| implements DatanodeAdminMonitorInterface { |
| /** |
| * Map containing the DECOMMISSION_INPROGRESS or ENTERING_MAINTENANCE |
| * datanodes that are being tracked so they can be be marked as |
| * DECOMMISSIONED or IN_MAINTENANCE. Even after the node is marked as |
| * IN_MAINTENANCE, the node remains in the map until |
| * maintenance expires checked during a monitor tick. |
| * <p/> |
| * This holds a set of references to the under-replicated blocks on the DN |
| * at the time the DN is added to the map, i.e. the blocks that are |
| * preventing the node from being marked as decommissioned. During a monitor |
| * tick, this list is pruned as blocks becomes replicated. |
| * <p/> |
| * Note also that the reference to the list of under-replicated blocks |
| * will be null on initial add |
| * <p/> |
| * However, this map can become out-of-date since it is not updated by block |
| * reports or other events. Before being finally marking as decommissioned, |
| * another check is done with the actual block map. |
| */ |
| private HashMap<DatanodeDescriptor, HashMap<BlockInfo, Integer>> |
| outOfServiceNodeBlocks = new HashMap<>(); |
| |
| /** |
| * The number of blocks to process when moving blocks to pendingReplication |
| * before releasing and reclaiming the namenode lock. |
| */ |
| private volatile int blocksPerLock; |
| |
| /** |
| * The number of blocks that have been checked on this tick. |
| */ |
| private int numBlocksChecked = 0; |
| /** |
| * The maximum number of blocks to hold in PendingRep at any time. |
| */ |
| private volatile int pendingRepLimit; |
| |
| /** |
| * The list of blocks which have been placed onto the replication queue |
| * and are waiting to be sufficiently replicated. |
| */ |
| private final Map<DatanodeDescriptor, List<BlockInfo>> |
| pendingRep = new HashMap<>(); |
| |
| private static final Logger LOG = |
| LoggerFactory.getLogger(DatanodeAdminBackoffMonitor.class); |
| |
| DatanodeAdminBackoffMonitor() { |
| } |
| |
| |
| @Override |
| protected void processConf() { |
| this.pendingRepLimit = conf.getInt( |
| DFSConfigKeys.DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT, |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT); |
| if (this.pendingRepLimit < 1) { |
| LOG.error("{} is set to an invalid value, it must be greater than "+ |
| "zero. Defaulting to {}", |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT, |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT |
| ); |
| this.pendingRepLimit = DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_LIMIT_DEFAULT; |
| } |
| this.blocksPerLock = conf.getInt( |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK, |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT |
| ); |
| if (blocksPerLock <= 0) { |
| LOG.error("{} is set to an invalid value, it must be greater than "+ |
| "zero. Defaulting to {}", |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK, |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT); |
| blocksPerLock = |
| DFSConfigKeys. |
| DFS_NAMENODE_DECOMMISSION_BACKOFF_MONITOR_PENDING_BLOCKS_PER_LOCK_DEFAULT; |
| } |
| LOG.info("Initialized the Backoff Decommission and Maintenance Monitor"); |
| } |
| |
| /** |
| * Queue a node to be removed from tracking. This method must be called |
| * under the namenode write lock. |
| * @param dn The datanode to stop tracking for decommission. |
| */ |
| @Override |
| public void stopTrackingNode(DatanodeDescriptor dn) { |
| getPendingNodes().remove(dn); |
| getCancelledNodes().add(dn); |
| } |
| |
| @Override |
| public int getTrackedNodeCount() { |
| return outOfServiceNodeBlocks.size(); |
| } |
| |
| @Override |
| public int getNumNodesChecked() { |
| // We always check all nodes on each tick |
| return outOfServiceNodeBlocks.size(); |
| } |
| |
| @Override |
| public void run() { |
| LOG.debug("DatanodeAdminMonitorV2 is running."); |
| if (!namesystem.isRunning()) { |
| LOG.info("Namesystem is not running, skipping " + |
| "decommissioning/maintenance checks."); |
| return; |
| } |
| // Reset the checked count at beginning of each iteration |
| numBlocksChecked = 0; |
| // Check decommission or maintenance progress. |
| try { |
| namesystem.writeLock(); |
| try { |
| /** |
| * Other threads can modify the pendingNode list and the cancelled |
| * node list, so we must process them under the NN write lock to |
| * prevent any concurrent modifications. |
| * |
| * Always process the cancelled list before the pending list, as |
| * it is possible for a node to be cancelled, and then quickly added |
| * back again. If we process these the other way around, the added |
| * node will be removed from tracking by the pending cancel. |
| */ |
| processCancelledNodes(); |
| |
| // Having more nodes decommissioning than can be tracked will impact decommissioning |
| // performance due to queueing delay |
| int numTrackedNodes = outOfServiceNodeBlocks.size(); |
| int numQueuedNodes = getPendingNodes().size(); |
| int numDecommissioningNodes = numTrackedNodes + numQueuedNodes; |
| if (numDecommissioningNodes > maxConcurrentTrackedNodes) { |
| LOG.warn( |
| "{} nodes are decommissioning but only {} nodes will be tracked at a time. " |
| + "{} nodes are currently queued waiting to be decommissioned.", |
| numDecommissioningNodes, maxConcurrentTrackedNodes, numQueuedNodes); |
| |
| // Re-queue unhealthy nodes to make space for decommissioning healthy nodes |
| final List<DatanodeDescriptor> unhealthyDns = outOfServiceNodeBlocks.keySet().stream() |
| .filter(dn -> !blockManager.isNodeHealthyForDecommissionOrMaintenance(dn)) |
| .collect(Collectors.toList()); |
| getUnhealthyNodesToRequeue(unhealthyDns, numDecommissioningNodes).forEach(dn -> { |
| getPendingNodes().add(dn); |
| outOfServiceNodeBlocks.remove(dn); |
| pendingRep.remove(dn); |
| }); |
| } |
| |
| processPendingNodes(); |
| } finally { |
| namesystem.writeUnlock("DatanodeAdminMonitorV2Thread"); |
| } |
| // After processing the above, various parts of the check() method will |
| // take and drop the read / write lock as needed. Aside from the |
| // cancelled and pending lists, nothing outside of the monitor thread |
| // modifies anything inside this class, so many things can be done |
| // without any lock. |
| check(); |
| } catch (Exception e) { |
| LOG.warn("DatanodeAdminMonitor caught exception when processing node.", |
| e); |
| } |
| if (numBlocksChecked + outOfServiceNodeBlocks.size() > 0) { |
| LOG.info("Checked {} blocks this tick. {} nodes are now " + |
| "in maintenance or transitioning state. {} nodes pending. {} " + |
| "nodes waiting to be cancelled.", |
| numBlocksChecked, outOfServiceNodeBlocks.size(), getPendingNodes().size(), |
| getCancelledNodes().size()); |
| } |
| } |
| |
| /** |
| * Move any pending nodes into outOfServiceNodeBlocks to initiate the |
| * decommission or maintenance mode process. |
| * |
| * This method must be executed under the namenode write lock to prevent |
| * the pendingNodes list from being modified externally. |
| */ |
| private void processPendingNodes() { |
| while (!getPendingNodes().isEmpty() && |
| (maxConcurrentTrackedNodes == 0 || |
| outOfServiceNodeBlocks.size() < maxConcurrentTrackedNodes)) { |
| outOfServiceNodeBlocks.put(getPendingNodes().poll(), null); |
| } |
| } |
| |
| /** |
| * Process any nodes which have had their decommission or maintenance mode |
| * cancelled by an administrator. |
| * |
| * This method must be executed under the |
| * write lock to prevent the cancelledNodes list being modified externally. |
| */ |
| private void processCancelledNodes() { |
| while(!getCancelledNodes().isEmpty()) { |
| DatanodeDescriptor dn = getCancelledNodes().poll(); |
| outOfServiceNodeBlocks.remove(dn); |
| pendingRep.remove(dn); |
| } |
| } |
| |
| /** |
| * This method performs each of the steps to track a node from |
| * decommissioning or entering maintenance to the end state. |
| * |
| * First, any newly added nodes are scanned. |
| * |
| * Then any expired maintenance nodes are handled. |
| * |
| * Next the pendingRep map is scanned and all blocks which are now |
| * sufficiently replicated are removed |
| * |
| * Then new blocks are moved to pendingRep |
| * |
| * Finally we check if any nodes have completed the replication process and |
| * if so move them to their final states. |
| * |
| * This methods which this method calls will take and release the namenode |
| * read and write lock several times. |
| * |
| */ |
| private void check() { |
| final List<DatanodeDescriptor> toRemove = new ArrayList<>(); |
| |
| if (outOfServiceNodeBlocks.size() == 0) { |
| // No nodes currently being tracked so simply return |
| return; |
| } |
| |
| // Check if there are any pending nodes to process, ie those where the |
| // storage has not been scanned yet. For all which are pending, scan |
| // the storage and load the under-replicated block list into |
| // outOfServiceNodeBlocks. As this does not modify any external structures |
| // it can be done under the namenode *read* lock, and the lock can be |
| // dropped between each storage on each node. |
| // |
| // TODO - This is an expensive call, depending on how many nodes are |
| // to be processed, but it requires only the read lock and it will |
| // be dropped and re-taken frequently. We may want to throttle this |
| // to process only a few nodes per iteration. |
| outOfServiceNodeBlocks.keySet() |
| .stream() |
| .filter(n -> outOfServiceNodeBlocks.get(n) == null) |
| .forEach(n -> scanDatanodeStorage(n, true)); |
| |
| processMaintenanceNodes(); |
| // First check the pending replication list and remove any blocks |
| // which are now replicated OK. This list is constrained in size so this |
| // call should not be overly expensive. |
| processPendingReplication(); |
| |
| // Now move a limited number of blocks to pending |
| moveBlocksToPending(); |
| |
| // Check if any nodes have reached zero blocks and also update the stats |
| // exposed via JMX for all nodes still being processed. |
| checkForCompletedNodes(toRemove); |
| |
| // Finally move the nodes to their final state if they are ready. |
| processCompletedNodes(toRemove); |
| } |
| |
| /** |
| * Checks for any nodes which are in maintenance and if maintenance has |
| * expired, the node will be moved back to in_service (or dead) as required. |
| */ |
| private void processMaintenanceNodes() { |
| // Check for any maintenance state nodes which need to be expired |
| namesystem.writeLock(); |
| try { |
| for (DatanodeDescriptor dn : outOfServiceNodeBlocks.keySet()) { |
| if (dn.isMaintenance() && dn.maintenanceExpired()) { |
| // If maintenance expires, stop tracking it. This can be an |
| // expensive call, as it may need to invalidate blocks. Therefore |
| // we can yield and retake the write lock after each node |
| // |
| // The call to stopMaintenance makes a call to stopTrackingNode() |
| // which added the node to the cancelled list. Therefore expired |
| // maintenance nodes do not need to be added to the toRemove list. |
| dnAdmin.stopMaintenance(dn); |
| namesystem.writeUnlock("processMaintenanceNodes"); |
| namesystem.writeLock(); |
| } |
| } |
| } finally { |
| namesystem.writeUnlock("processMaintenanceNodes"); |
| } |
| } |
| |
| /** |
| * Loop over all nodes in the passed toRemove list and move the node to |
| * the required end state. This will also remove any entries from |
| * outOfServiceNodeBlocks and pendingRep for the node if required. |
| * |
| * @param toRemove The list of nodes to process for completion. |
| */ |
| private void processCompletedNodes(List<DatanodeDescriptor> toRemove) { |
| if (toRemove.size() == 0) { |
| // If there are no nodes to process simply return and avoid |
| // taking the write lock at all. |
| return; |
| } |
| namesystem.writeLock(); |
| try { |
| for (DatanodeDescriptor dn : toRemove) { |
| final boolean isHealthy = |
| blockManager.isNodeHealthyForDecommissionOrMaintenance(dn); |
| if (isHealthy) { |
| if (dn.isDecommissionInProgress()) { |
| dnAdmin.setDecommissioned(dn); |
| outOfServiceNodeBlocks.remove(dn); |
| pendingRep.remove(dn); |
| } else if (dn.isEnteringMaintenance()) { |
| // IN_MAINTENANCE node remains in the outOfServiceNodeBlocks to |
| // to track maintenance expiration. |
| dnAdmin.setInMaintenance(dn); |
| pendingRep.remove(dn); |
| } else if (dn.isInService()) { |
| // Decom / maint was cancelled and the node is yet to be processed |
| // from cancelledNodes |
| LOG.info("Node {} completed decommission and maintenance " + |
| "but has been moved back to in service", dn); |
| pendingRep.remove(dn); |
| outOfServiceNodeBlocks.remove(dn); |
| continue; |
| } else { |
| // Should not happen |
| LOG.error("Node {} is in an unexpected state {} and has been "+ |
| "removed from tracking for decommission or maintenance", |
| dn, dn.getAdminState()); |
| pendingRep.remove(dn); |
| outOfServiceNodeBlocks.remove(dn); |
| continue; |
| } |
| LOG.info("Node {} is sufficiently replicated and healthy, " |
| + "marked as {}.", dn, dn.getAdminState()); |
| } else { |
| LOG.info("Node {} isn't healthy." |
| + " It needs to replicate {} more blocks." |
| + " {} is still in progress.", dn, |
| getPendingCountForNode(dn), dn.getAdminState()); |
| } |
| } |
| } finally { |
| namesystem.writeUnlock("processCompletedNodes"); |
| } |
| } |
| |
| /** |
| * Loop over all nodes and check for any which have zero unprocessed or |
| * pending blocks. If the node has zero blocks pending, the storage is |
| * rescanned to ensure no transient blocks were missed on the first pass. |
| * |
| * If, after rescan the number of blocks pending replication is zero, the |
| * node is added to the passed removeList which will later be processed to |
| * complete the decommission or entering maintenance process. |
| * |
| * @param removeList Nodes which have zero pending blocks are added to this |
| * list. |
| */ |
| private void checkForCompletedNodes(List<DatanodeDescriptor> removeList) { |
| for (DatanodeDescriptor dn : outOfServiceNodeBlocks.keySet()) { |
| // If the node is already in maintenance, we don't need to perform |
| // any further checks on it. |
| if (dn.isInMaintenance()) { |
| LOG.debug("Node {} is currently in maintenance", dn); |
| continue; |
| } else if (!dn.isInService()) { |
| // A node could be inService if decom or maint has been cancelled, but |
| // the cancelled list is yet to be processed. We don't need to check |
| // inService nodes here |
| int outstandingBlocks = getPendingCountForNode(dn); |
| if (outstandingBlocks == 0) { |
| scanDatanodeStorage(dn, false); |
| outstandingBlocks = getPendingCountForNode(dn); |
| } |
| LOG.info("Node {} has {} blocks yet to process", dn, outstandingBlocks); |
| if (outstandingBlocks == 0) { |
| removeList.add(dn); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Returns the number of block pending for the given node by adding those |
| * blocks in pendingRep and outOfServiceNodeBlocks. |
| * |
| * @param dn The datanode to return the count for |
| * @return The total block count, or zero if none are pending |
| */ |
| private int getPendingCountForNode(DatanodeDescriptor dn) { |
| int count = 0; |
| HashMap<BlockInfo, Integer> blocks = outOfServiceNodeBlocks.get(dn); |
| if (blocks != null) { |
| count += blocks.size(); |
| } |
| List<BlockInfo> pendingBlocks = pendingRep.get(dn); |
| if (pendingBlocks != null) { |
| count += pendingBlocks.size(); |
| } |
| return count; |
| } |
| |
| /** |
| * Iterate across all nodes in outOfServiceNodeBlocks which have blocks yet |
| * to be processed. |
| * |
| * The block is removed from outOfServiceNodeBlocks and if it needs |
| * replication it is added to the pendingRep map and also to the |
| * BlockManager replication queue. |
| * |
| * Any block that does not need replication is discarded. |
| * |
| * The method will return when there are the pendingRep map has |
| * pendingRepLimit blocks or there are no further blocks to process. |
| */ |
| private void moveBlocksToPending() { |
| int blocksProcessed = 0; |
| int pendingCount = getPendingCount(); |
| int yetToBeProcessed = getYetToBeProcessedCount(); |
| |
| if (pendingCount == 0 && yetToBeProcessed == 0) { |
| // There are no blocks to process so just return |
| LOG.debug("There are no pending or blocks yet to be processed"); |
| return; |
| } |
| |
| namesystem.writeLock(); |
| try { |
| long repQueueSize = blockManager.getLowRedundancyBlocksCount(); |
| |
| LOG.info("There are {} blocks pending replication and the limit is "+ |
| "{}. A further {} blocks are waiting to be processed. "+ |
| "The replication queue currently has {} blocks", |
| pendingCount, pendingRepLimit, yetToBeProcessed, repQueueSize); |
| |
| if (pendingCount >= pendingRepLimit) { |
| // Only add more blocks to the replication queue if we don't already |
| // have too many pending |
| return; |
| } |
| |
| // Create a "Block Iterator" for each node decommissioning or entering |
| // maintenance. These iterators will be used "round robined" to add blocks |
| // to the replication queue and PendingRep |
| HashMap<DatanodeDescriptor, Iterator<BlockInfo>> |
| iterators = new HashMap<>(); |
| for (Map.Entry<DatanodeDescriptor, HashMap<BlockInfo, Integer>> e |
| : outOfServiceNodeBlocks.entrySet()) { |
| iterators.put(e.getKey(), e.getValue().keySet().iterator()); |
| } |
| |
| // Now loop until we fill the pendingRep map with pendingRepLimit blocks |
| // or run out of blocks to add. |
| Iterator<DatanodeDescriptor> nodeIter = |
| Iterables.cycle(iterators.keySet()).iterator(); |
| while (nodeIter.hasNext()) { |
| // Cycle through each node with blocks which still need processed |
| DatanodeDescriptor dn = nodeIter.next(); |
| Iterator<BlockInfo> blockIt = iterators.get(dn); |
| while (blockIt.hasNext()) { |
| // Process the blocks for the node until we find one that needs |
| // replication |
| if (blocksProcessed >= blocksPerLock) { |
| blocksProcessed = 0; |
| namesystem.writeUnlock("moveBlocksToPending"); |
| namesystem.writeLock(); |
| } |
| blocksProcessed++; |
| if (nextBlockAddedToPending(blockIt, dn)) { |
| // Exit the inner "block" loop so an iterator for the next datanode |
| // is used for the next block. |
| pendingCount++; |
| break; |
| } |
| } |
| if (!blockIt.hasNext()) { |
| // remove the iterator as there are no blocks left in it |
| nodeIter.remove(); |
| } |
| if (pendingCount >= pendingRepLimit) { |
| // We have scheduled the limit of blocks for replication, so do |
| // not add any more |
| break; |
| } |
| } |
| } finally { |
| namesystem.writeUnlock("moveBlocksToPending"); |
| } |
| LOG.debug("{} blocks are now pending replication", pendingCount); |
| } |
| |
| /** |
| * Takes and removes the next block from the given iterator and checks if it |
| * needs additional replicas. If it does, it will be scheduled for |
| * reconstruction and added to the pendingRep map. |
| * @param it The iterator to take the next block from |
| * @param dn The datanodeDescriptor the iterator applies to |
| * @return True if the block needs replication, otherwise false |
| */ |
| private boolean nextBlockAddedToPending(Iterator<BlockInfo> it, |
| DatanodeDescriptor dn) { |
| BlockInfo block = it.next(); |
| it.remove(); |
| numBlocksChecked++; |
| if (!isBlockReplicatedOk(dn, block, true, null)) { |
| pendingRep.computeIfAbsent(dn, k -> new LinkedList<>()).add(block); |
| return true; |
| } |
| return false; |
| } |
| |
| private int getPendingCount() { |
| if (pendingRep.size() == 0) { |
| return 0; |
| } |
| return pendingRep.values() |
| .stream() |
| .map(a -> a.size()) |
| .reduce(0, (a, b) -> a + b); |
| } |
| |
| private int getYetToBeProcessedCount() { |
| if (outOfServiceNodeBlocks.size() == 0) { |
| return 0; |
| } |
| return outOfServiceNodeBlocks.values() |
| .stream() |
| .map(a -> a.size()) |
| .reduce(0, (a, b) -> a + b); |
| } |
| |
| /** |
| * Scan all the blocks held on a datanodes. For a node being decommissioned |
| * we assume that the majority of blocks on the node will need to have new |
| * replicas made, and therefore we do not check if they are under replicated |
| * here and instead add them to the list of blocks to track. |
| * |
| * For a node being moved into maintenance, we assume most blocks will be |
| * replicated OK and hence we do check their under-replicated status here, |
| * hopefully reducing the number of blocks to track. |
| * |
| * On a re-scan (initalScan = false) we assume the node has been processed |
| * already, and hence there should be few under-replicated blocks, so we |
| * check the under-replicated status before adding the blocks to the |
| * tracking list. |
| * |
| * This means that for a node being decomission there should be a large |
| * number of blocks to process later but for maintenance, a smaller number. |
| * |
| * As this method does not schedule any blocks for reconstuction, this |
| * scan can be performed under the namenode readlock, and the lock is |
| * dropped and reaquired for each storage on the DN. |
| * |
| * @param dn - The datanode to process |
| * @param initialScan - True is this is the first time scanning the node |
| * or false if it is a rescan. |
| */ |
| private void scanDatanodeStorage(DatanodeDescriptor dn, |
| Boolean initialScan) { |
| HashMap<BlockInfo, Integer> blockList = outOfServiceNodeBlocks.get(dn); |
| if (blockList == null) { |
| blockList = new HashMap<>(); |
| outOfServiceNodeBlocks.put(dn, blockList); |
| } |
| |
| DatanodeStorageInfo[] storage; |
| namesystem.readLock(); |
| try { |
| storage = dn.getStorageInfos(); |
| } finally { |
| namesystem.readUnlock("scanDatanodeStorage"); |
| } |
| |
| for (DatanodeStorageInfo s : storage) { |
| namesystem.readLock(); |
| try { |
| // As the lock is dropped and re-taken between each storage, we need |
| // to check the storage is still present before processing it, as it |
| // may have been removed. |
| if (dn.getStorageInfo(s.getStorageID()) == null) { |
| continue; |
| } |
| Iterator<BlockInfo> it = s.getBlockIterator(); |
| while (it.hasNext()) { |
| BlockInfo b = it.next(); |
| if (!initialScan || dn.isEnteringMaintenance()) { |
| // this is a rescan, so most blocks should be replicated now, |
| // or this node is going into maintenance. On a healthy |
| // cluster using racks or upgrade domain, a node should be |
| // able to go into maintenance without replicating many blocks |
| // so we will check them immediately. |
| if (!isBlockReplicatedOk(dn, b, false, null)) { |
| blockList.put(b, null); |
| } |
| } else { |
| blockList.put(b, null); |
| } |
| numBlocksChecked++; |
| } |
| } finally { |
| namesystem.readUnlock("scanDatanodeStorage"); |
| } |
| } |
| } |
| |
| /** |
| * Process the list of pendingReplication Blocks. These are the blocks |
| * which have been moved from outOfServiceNodeBlocks, confirmed to be |
| * under-replicated and were added to the blockManager replication |
| * queue. |
| * |
| * Any blocks which have been confirmed to be replicated sufficiently are |
| * removed from the list. |
| * |
| * The datanode stats are also updated in this method, updating the total |
| * pending block count, the number of blocks in PendingRep which are in |
| * open files and the number of blocks in PendingRep which are only on |
| * out of service nodes. |
| * |
| * As this method makes changes to the replication queue, it acquires the |
| * namenode write lock while it runs. |
| */ |
| private void processPendingReplication() { |
| namesystem.writeLock(); |
| try { |
| for (Iterator<Map.Entry<DatanodeDescriptor, List<BlockInfo>>> |
| entIt = pendingRep.entrySet().iterator(); entIt.hasNext();) { |
| Map.Entry<DatanodeDescriptor, List<BlockInfo>> entry = entIt.next(); |
| DatanodeDescriptor dn = entry.getKey(); |
| List<BlockInfo> blocks = entry.getValue(); |
| if (blocks == null) { |
| // should not be able to happen |
| entIt.remove(); |
| continue; |
| } |
| Iterator<BlockInfo> blockIt = blocks.iterator(); |
| BlockStats suspectBlocks = new BlockStats(); |
| while(blockIt.hasNext()) { |
| BlockInfo b = blockIt.next(); |
| if (isBlockReplicatedOk(dn, b, true, suspectBlocks)) { |
| blockIt.remove(); |
| } |
| numBlocksChecked++; |
| } |
| if (blocks.size() == 0) { |
| entIt.remove(); |
| } |
| // Update metrics for this datanode. |
| dn.getLeavingServiceStatus().set( |
| suspectBlocks.getOpenFileCount(), |
| suspectBlocks.getOpenFiles(), |
| getPendingCountForNode(dn), |
| suspectBlocks.getOutOfServiceBlockCount()); |
| } |
| } finally { |
| namesystem.writeUnlock("processPendingReplication"); |
| } |
| } |
| |
| /** |
| * Checks if a block is sufficiently replicated and optionally schedules |
| * it for reconstruction if it is not. |
| * |
| * If a BlockStats object is passed, this method will also update it if the |
| * block is part of an open file or only on outOfService nodes. |
| * |
| * @param datanode The datanode the block belongs to |
| * @param block The block to check |
| * @param scheduleReconStruction Whether to add the block to the replication |
| * queue if it is not sufficiently replicated. |
| * Passing true will add it to the replication |
| * queue, and false will not. |
| * @param suspectBlocks If non-null check if the block is part of an open |
| * file or only on out of service nodes and update the |
| * passed object accordingly. |
| * @return |
| */ |
| private boolean isBlockReplicatedOk(DatanodeDescriptor datanode, |
| BlockInfo block, Boolean scheduleReconStruction, |
| BlockStats suspectBlocks) { |
| if (blockManager.blocksMap.getStoredBlock(block) == null) { |
| LOG.trace("Removing unknown block {}", block); |
| return true; |
| } |
| |
| long bcId = block.getBlockCollectionId(); |
| if (bcId == INodeId.INVALID_INODE_ID) { |
| // Orphan block, will be invalidated eventually. Skip. |
| return false; |
| } |
| |
| final BlockCollection bc = blockManager.getBlockCollection(block); |
| final NumberReplicas num = blockManager.countNodes(block); |
| final int liveReplicas = num.liveReplicas(); |
| |
| // Schedule low redundancy blocks for reconstruction |
| // if not already pending. |
| boolean isDecommission = datanode.isDecommissionInProgress(); |
| boolean isMaintenance = datanode.isEnteringMaintenance(); |
| |
| boolean neededReconstruction = isDecommission ? |
| blockManager.isNeededReconstruction(block, num) : |
| blockManager.isNeededReconstructionForMaintenance(block, num); |
| if (neededReconstruction && scheduleReconStruction) { |
| if (!blockManager.neededReconstruction.contains(block) && |
| blockManager.pendingReconstruction.getNumReplicas(block) == 0 && |
| blockManager.isPopulatingReplQueues()) { |
| // Process these blocks only when active NN is out of safe mode. |
| blockManager.neededReconstruction.add(block, |
| liveReplicas, num.readOnlyReplicas(), |
| num.outOfServiceReplicas(), |
| blockManager.getExpectedRedundancyNum(block)); |
| } |
| } |
| |
| if (suspectBlocks != null) { |
| // Only if we pass a BlockStats object should we do these |
| // checks, as they should only be checked when processing PendingRep. |
| if (bc.isUnderConstruction()) { |
| INode ucFile = namesystem.getFSDirectory().getInode(bc.getId()); |
| if (!(ucFile instanceof INodeFile) || |
| !ucFile.asFile().isUnderConstruction()) { |
| LOG.warn("File {} is not under construction. Skipping add to " + |
| "low redundancy open files!", ucFile.getLocalName()); |
| } else { |
| suspectBlocks.addOpenFile(ucFile.getId()); |
| } |
| } |
| if ((liveReplicas == 0) && (num.outOfServiceReplicas() > 0)) { |
| suspectBlocks.incrementOutOfServiceBlocks(); |
| } |
| } |
| |
| // Even if the block is without sufficient redundancy, |
| // it might not block decommission/maintenance if it |
| // has sufficient redundancy. |
| if (dnAdmin.isSufficient(block, bc, num, isDecommission, isMaintenance)) { |
| return true; |
| } |
| return false; |
| } |
| |
| @VisibleForTesting |
| @Override |
| public int getPendingRepLimit() { |
| return pendingRepLimit; |
| } |
| |
| public void setPendingRepLimit(int pendingRepLimit) { |
| this.pendingRepLimit = pendingRepLimit; |
| } |
| |
| @VisibleForTesting |
| @Override |
| public int getBlocksPerLock() { |
| return blocksPerLock; |
| } |
| |
| public void setBlocksPerLock(int blocksPerLock) { |
| this.blocksPerLock = blocksPerLock; |
| } |
| |
| static class BlockStats { |
| private LightWeightHashSet<Long> openFiles = |
| new LightWeightLinkedSet<>(); |
| private int openFileBlockCount = 0; |
| private int outOfServiceBlockCount = 0; |
| |
| public void addOpenFile(long id) { |
| // Several blocks can be part of the same file so track how |
| // many adds we get, as the same file could be added several times |
| // for different blocks. |
| openFileBlockCount++; |
| openFiles.add(id); |
| } |
| |
| public void incrementOutOfServiceBlocks() { |
| outOfServiceBlockCount++; |
| } |
| |
| public LightWeightHashSet<Long> getOpenFiles() { |
| return openFiles; |
| } |
| |
| public int getOpenFileCount() { |
| return openFileBlockCount; |
| } |
| |
| public int getOutOfServiceBlockCount() { |
| return outOfServiceBlockCount; |
| } |
| } |
| } |
