第七章：小朱笔记hadoop之源码分析-hdfs分析 第四节：namenode-DecommissionManager

xmxm76

第七章：小朱笔记hadoop之源码分析-hdfs分析

第四节：namenode分析

4.6 namenode任务线程之DecommissionManager$Monitor

       DecommissionManager主要是负责节点退役或者说节点停用，Monitor负责定时来检测这些节点的退役状态。DataNode节点退役得等该节点上的所有数据块Blocks被复制完成之后，才能允许退役。这就需要Monitor负责定时地检测这些节点中Blocks的状态，当这些Blocks都满足副本因子之后，才能将该DataNode节点置为退役状态。
       DataNode的状态：

enum AdminStates {NORMAL, DECOMMISSION_INPROGRESS, DECOMMISSIONED; }  
//AdminStates.NORMAL：正常态  
//AdminStates.DECOMMISSION_INPROGRESS：表示DataNode节点处于退役处理状态  
//AdminStates.DECOMMISSIONED：后一个变量表示节点处于已退役状态  
（1）NORMAL> DECOMMISSION_INPROGRESS 转换
         NameNode为客户端提供了一个方法refreshNodes()，当客户端调用该接口的时候，NameNode节点会重新加载配置文件中的 dfs.hosts和dfs.hosts.exclude两个选项，更新HostsFileReader，之后根据HostsFileReader的新主 机列表来判断哪些DataNode节点被允许连接到NameNode，哪些不允许被连接，对于不允许连接的DataNode节点，先将其标记为 AdminStates.DECOMMISSION_INPROGRESS
（2） DECOMMISSION_INPROGRESS> DECOMMISSIONED转换
         这正是DecommissionManager$Monitor所做的事情
（3）DECOMMISSIONED>关闭
         当数据节点重启或者重新初始化之后，都会首先向NameNode节点注册，如果此时主节点不允许该数据节点加入集群，那么它就会对该数据节点进行强制退役处理。当DecommissionManager$Monitor线程把一个DataNode节点标记为 AdminStates.DECOMMISSIONED状态之后，NameNode节点会在当该DataNode节点下一次发送heart的时候，就明确的给他一个command，让它shutdown。
         主要属性：

    // 一次轮询检查的时间间隔  
private final long recheckInterval;  
// 每次轮询检查处于正在退役状态的数据节点数量  
private final int numNodesPerCheck;  
// 上一次轮询结束时处理到的最后一个数据节点   
private String firstkey = "";  
           DataNode多并且检测耗时，所以采用迭代的步长检测，每次只检测处于正在退役状态的numNodesPerCheck个数据节点，下次检测时就从上次检测的的最后一个数据节点开始，而这两次检测的时间间隔是recheckInterval s。
    为了保证检测的连续性，所以检测的数据构成一个环形数据结构，然后需要记录每次迭代的起始点，所以有个firstkey属性，然后环形数据结构由hadoop自己实现的CyclicIteration来提供。
    检测逻辑：
    （a）判断节点是否处于DECOMMISSION_INPROGRESS状态;
    （b）如果处于DECOMMISSION_INPROGRESS状态，则检测该节点上的所有数据块是否已经到了复制因子;
    （c）如果已经到了复制因子块的个数则说明这个节点上的所有块都是多余了，也就是说这个节点可以退役了，将节点状态设置为退役即 DECOMMISSIONED
 

    private void check() {  
int count = 0;  
for(Map.Entry<String, DatanodeDescriptor> entry : new CyclicIteration<String, DatanodeDescriptor>(fsnamesystem.datanodeMap, firstkey)) {  
final DatanodeDescriptor d = entry.getValue();  
firstkey = entry.getKey();  
////数据节点处于正在退役状态   
if (d.isDecommissionInProgress()) {  
try {  
//检查数据节点是否可以退役，如果可以则使其退役   
//检测该节点上的所有数据块是否已经到了复制因子  
fsnamesystem.checkDecommissionStateInternal(d);  
} catch(Exception e) {  
LOG.warn("entry=" + entry, e);  
}  
//检查处于正在退役状态的数据节点达到设定的阈值就退出本次轮询   
if (++count == numNodesPerCheck) {  
return;  
}  
}  
}  
}  
}  
    检查存储在它上面的所有数据块Block的副本是否已经满足用户设置的副本值，如果不满足还要讲该数据块Block添加到待复制副本的队列中。

   

    /**
* Change, if appropriate, the admin state of a datanode to  
* decommission completed. Return true if decommission is complete.
*  
*  从上面的检测逻辑可以看出，其实就是为了确定是否这个节点上的数据块已经被完全复制过了到了复制因子，
*  如果完全复制过了，那这个节点其实就没用了，所以就标示为退役了，主要起到一个审查的目的
*  DECOMMISSION_INPROGRESS   
*  DFSAdmin的 -refreshNodes 命令或者说传递这个命令参数的其他调用者例如Balancer ，DFSck等。   
*  refreshNode会依据exclude文件里配置的ip来更新node节点的状态，例如这里是调用setDECOMMISSION_INPROGRESS（datanode）   
*   
*  需要检查存储在它上面的所有数据块Block的副本是否已经满足用户设置的副本值，如果不满足还要讲该数据块Block添加到待复制副本的队列中，这个检查过程是比较耗时的
*/  
boolean checkDecommissionStateInternal(DatanodeDescriptor node) {  
//  
// Check to see if all blocks in this decommissioned  
// node has reached their target replication factor.  
//  
if (node.isDecommissionInProgress()) {  
if (!isReplicationInProgress(node)) {  
node.setDecommissioned();  
LOG.info("Decommission complete for node " + node.getName());  
}  
}  
if (node.isDecommissioned()) {  
return true;  
}  
return false;  
}  

/**
* Return true if there are any blocks on this node that have not
* yet reached their replication factor. Otherwise returns false.
*/  
/**  
* 检查一个数据节点上的每一个数据块的副本数量是否满足设置的值，如果都满足则返回false，否则返回ture;  
* 对于副本数量不满足设置值的数据块，如果还没有对该数据块进行副本复制，则将其挂载到副本复制线程的任务队列中  
*/   
private boolean isReplicationInProgress(DatanodeDescriptor srcNode) {  
boolean status = false;  
int underReplicatedBlocks = 0;  
int decommissionOnlyReplicas = 0;  
int underReplicatedInOpenFiles = 0;  
for(final Iterator<Block> i = srcNode.getBlockIterator(); i.hasNext(); ) {  
final Block block = i.next();  
INode fileINode = blocksMap.getINode(block);  
if (fileINode != null) {  
NumberReplicas num = countNodes(block);  
int curReplicas = num.liveReplicas();  
int curExpectedReplicas = getReplication(block);  
if (curExpectedReplicas > curReplicas) {  
// Log info about one block for this node which needs replication  
if (!status) {  
status = true;  
logBlockReplicationInfo(block, srcNode, num);  
}  
underReplicatedBlocks++;  
if ((curReplicas == 0) && (num.decommissionedReplicas() > 0)) {  
decommissionOnlyReplicas++;  
}  
if (fileINode.isUnderConstruction()) {  
underReplicatedInOpenFiles++;  
}  
//数据块不在待(副本)复制队列中，也不在正在(副本)复制队列中   
if (!neededReplications.contains(block) &&  
pendingReplications.getNumReplicas(block) == 0) {  
//  
// These blocks have been reported from the datanode  
// after the startDecommission method has been executed. These  
// blocks were in flight when the decommission was started.  
//  
neededReplications.add(block,   
curReplicas,  
num.decommissionedReplicas(),  
curExpectedReplicas);  
}  
}  
}  
}  
srcNode.decommissioningStatus.set(underReplicatedBlocks,  
decommissionOnlyReplicas, underReplicatedInOpenFiles);  
return status;  
}