Hadoop3.2.1 【 HDFS 】源码分析 : Datanode升级、回滚、提交

本是古典何须时尚 2022-12-20 02:25 184阅读 0赞

### List of articles ###

*  一.前言
 *  一.doUpgrade()
 *  三.doRollback()
 *  四.doFinalize()

# 一.前言 #

本文主要介绍DataNode中的 doRollback()、 doUpgrade()以及doFinalize()这三个DataStorage中与升级相关的方法.分别负责回滚,升级,提交三个操作.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

# 一.doUpgrade() #

doUpgrade()方法用于对单个存储目录下的指定块池目录进行升级， 这个存储目录由doUpgrade()方法的参数sd（StorageDirectory类型） 指定， 而块池信息则由nsInfo参数（NamespaceInfo类型） 指定。  
doUpgrade()方法首先会判断当前存储目录的文件系统布局版本号是否支持Federation部署， 如果支持则直接更新存储目录的VERSION文件， 然后返回， 也就是将块池目录的升级操作委托给BlockPoolSliceStorage.doUpgrade()方法负责（recoverTransitionRea()方法会  
调用BlockPoolSliceStorage.recoverTransitionRead()触发BlockPoolSliceStorage.doUpgrade()方法） 。 如果原有版本号不支持Federation部署， 也就是说， 原有的存储目录中并不存在块池目录。 则先确保存储目录处于正常状态， 并且确认存储目录的current文件夹存在， 然后再进行升级操作。

整个升级操作分为5个步骤进行。
    ■ 删除存储目录的previous文件夹， 并确认previous.tmp文件夹不存在。
    ■ 将存储目录的current目录重命名为previous.tmp。
    ■ 创建BlockPoolSliceStorage对象， 在当前存储目录下创建块池目录， 然后格式化
    （写入VERSION文件） 。 接下来创建块池目录的current文件夹， 并与存储目录
    的previous.tmp文件夹建立硬链接。
    ■ 在存储目录中写入新的VERSION文件。
    ■ 最后将存储目录的previous.tmp文件夹改名为previous， 完成整个升级操作

/**
       * Upgrade to any release after 0.22 (0.22 included) release e.g. 0.22 => 0.23
       * Upgrade procedure is as follows:
       * <ol>
       * <li>If <SD>/current/<bpid>/previous exists then delete it</li>
       * <li>Rename <SD>/current/<bpid>/current to
       * <SD>/current/bpid/current/previous.tmp</li>
       * <li>Create new <SD>current/<bpid>/current directory</li>
       * <ol>
       * <li>Hard links for block files are created from previous.tmp to current</li>
       * <li>Save new version file in current directory</li>
       * </ol>
       * <li>Rename previous.tmp to previous</li> </ol>
       * 
       * @param bpSd storage directory <SD>/current/<bpid>
       * @param nsInfo Namespace Info from the namenode
       * @throws IOException on error
       */
      private void doUpgrade(final StorageDirectory bpSd,
          final NamespaceInfo nsInfo,
          final List<Callable<StorageDirectory>> callables,
          final Configuration conf) throws IOException {
    
        // 如果当前存储目录的文件系统布局版本号支持Federation部署
        // Upgrading is applicable only to release with federation or after
        // 直接更新layoutVersion， 并写入存储目录的VERSION文件中
        // 其他升级操作由BlockPoolSliceStorage.doUpgrade()方法负责
        if (!DataNodeLayoutVersion.supports(
            LayoutVersion.Feature.FEDERATION, layoutVersion)) {
          return;
        }
        // no upgrades for storage directories that are PROVIDED
        if (bpSd.getRoot() == null) {
          return;
        }
        final int oldLV = getLayoutVersion();
        LOG.info("Upgrading block pool storage directory {}.\n   old LV = {}; old"
            + " CTime = {}.\n   new LV = {}; new CTime = {}",
            bpSd.getRoot(), oldLV, this.getCTime(), HdfsServerConstants
                .DATANODE_LAYOUT_VERSION, nsInfo.getCTime());
        // get <SD>/previous directory
        String dnRoot = getDataNodeStorageRoot(bpSd.getRoot().getCanonicalPath());
        StorageDirectory dnSdStorage = new StorageDirectory(new File(dnRoot));
        File dnPrevDir = dnSdStorage.getPreviousDir();
        
        // If <SD>/previous directory exists delete it
        if (dnPrevDir.exists()) {
          deleteDir(dnPrevDir);
        }
        final File bpCurDir = bpSd.getCurrentDir();
        final File bpPrevDir = bpSd.getPreviousDir();
    
        //确认current目录存在
        assert bpCurDir.exists() : "BP level current directory must exist.";
        cleanupDetachDir(new File(bpCurDir, DataStorage.STORAGE_DIR_DETACHED));
    
        //1． 删除previous目录
        // 1. Delete <SD>/current/<bpid>/previous dir before upgrading
        if (bpPrevDir.exists()) {
          deleteDir(bpPrevDir);
        }
        final File bpTmpDir = bpSd.getPreviousTmp();
        //确认previous.tmp目录不存在
        assert !bpTmpDir.exists() : "previous.tmp directory must not exist.";
    
        // 2． 将current目录重命名为previous.tmp
        // 2. Rename <SD>/current/<bpid>/current to
        //    <SD>/current/<bpid>/previous.tmp
        rename(bpCurDir, bpTmpDir);
        
        final String name = "block pool " + blockpoolID + " at " + bpSd.getRoot();
        if (callables == null) {
          //执行升级操作
          doUpgrade(name, bpSd, nsInfo, bpPrevDir, bpTmpDir, bpCurDir, oldLV, conf);
        } else {
          callables.add(new Callable<StorageDirectory>() {
            @Override
            public StorageDirectory call() throws Exception {
              doUpgrade(name, bpSd, nsInfo, bpPrevDir, bpTmpDir, bpCurDir, oldLV,
                  conf);
              return bpSd;
            }
          });
        }
      }
    
      private void doUpgrade(String name, final StorageDirectory bpSd,
          NamespaceInfo nsInfo, final File bpPrevDir, final File bpTmpDir,
          final File bpCurDir, final int oldLV, Configuration conf)
              throws IOException {
    
        //3． 将块池目录的current文件夹与存储目录的previous.tmp文件夹建立硬链接
    
        // 3. Create new <SD>/current with block files hardlinks and VERSION
        linkAllBlocks(bpTmpDir, bpCurDir, oldLV, conf);
        this.layoutVersion = HdfsServerConstants.DATANODE_LAYOUT_VERSION;
        assert this.namespaceID == nsInfo.getNamespaceID() 
            : "Data-node and name-node layout versions must be the same.";
        this.cTime = nsInfo.getCTime();
        writeProperties(bpSd);
    
    
        // 4. 将previous.tmp目录改名为previous
        // 4. rename <SD>/current/<bpid>/previous.tmp to
        // <SD>/current/<bpid>/previous
        rename(bpTmpDir, bpPrevDir);
        LOG.info("Upgrade of {} is complete", name);
      }

从代码分析中可以看到， doUpgrade()方法的执行分为两种情况。

■ 一种情况是当前Datanode存储的文件系统布局版本号支持Federation部署。 也就是说， 当前Datanode存储目录中已经保存了若干块池目录， 那么当前的升级操作只需要在存储目录的VERSION文件中更新layoutVersion字段即可， 接下来的操作都是在块池目录中进行的， 需要由BlockPoolSliceStorage.doUpgrade()方法执行。
    ■ 另一种情况是当前Datanode存储的文件系统版本号比较低， 并不支持Federation部署。 也就是说， 当前存储目录中并不存在块池目录， 存储目录中只保存了当前唯一的命名空间的所有数据块。 在这种情况下， doUpgrade()方法就需要先在当前存储目录中创建并格式化（ 写入VERSION文件） 一个新的块池目录， 使存储目录的结构符合Federation目录结构（ 请参考Datanode磁盘存储结构小节） ， 然后创建BlockPoolSliceStorage对象管理这个块池目录， 最后doUpgrade()方法才可以执行升级操作。 这里还需要注意一点， 升级前的数据块文件都是保存在存储目录的current文件夹中的（ current文件夹在升级操作中被重命名为previous.tmp） ， 这些数据块并不属于任何块池。 而升级之后这些数据块要保存在块池目录的current文件夹中， 也就是原有的数据块都将属于新建立的块池，所以doUpgrade()方法需要建立存储目录的previous.tmp文件夹与块池目录current文件夹的硬链接。

# 三.doRollback() #

doRollback()方法用于对单个存储目录下的指定块池目录进行回滚操作， 这个存储目录由doUpgrade()方法的参数sd(StorageDirectory类型） 指定， 而块池信息则由nsInfo参数（NamespaceInfo类型） 指定。

doRollback()方法的流程与doUpgrade()非常类似。 doRollback()方法首先判断当前回滚操作是不是在Federation部署条件下进行的， 如果是则更新存储目录的VERSION文件， 然后返回， 也就是将块池目录的回滚操作委托给BlockPoolSliceStorage.doRollback()方法负  
责。 如果不是， doRollback()方法会检查存储目录的previous目录是否存在， 检查是否可以执行回滚操作。 完成了上述判断操作后， doRollback()方法才可以进行回滚操作。 整个回滚操作分为如下几个步骤进行。

■ 首先将存储目录的current目录重命名为removed.tmp。  
■ 然后将previous目录重命名为current。  
■ 最后删除removed.tmp目录， 完成回滚操作。

/*
       * Roll back to old snapshot at the block pool level
       * If previous directory exists: 
       * <ol>
       * <li>Rename <SD>/current/<bpid>/current to removed.tmp</li>
       * <li>Rename * <SD>/current/<bpid>/previous to current</li>
       * <li>Remove removed.tmp</li>
       * </ol>
       * 
       * Do nothing if previous directory does not exist.
       * @param bpSd Block pool storage directory at <SD>/current/<bpid>
       */
      void doRollback(StorageDirectory bpSd, NamespaceInfo nsInfo)
          throws IOException {
        File prevDir = bpSd.getPreviousDir();
        // regular startup if previous dir does not exist
        if (prevDir == null || !prevDir.exists()) {
          return;
        }
        // read attributes out of the VERSION file of previous directory
        BlockPoolSliceStorage prevInfo = new BlockPoolSliceStorage();
        prevInfo.readPreviousVersionProperties(bpSd);
    
        // We allow rollback to a state, which is either consistent with
        // the namespace state or can be further upgraded to it.
        // In another word, we can only roll back when ( storedLV >= software LV)
        // && ( DN.previousCTime <= NN.ctime)
        if (!(prevInfo.getLayoutVersion() >= HdfsServerConstants.DATANODE_LAYOUT_VERSION &&
            prevInfo.getCTime() <= nsInfo.getCTime())) { // cannot rollback
          throw new InconsistentFSStateException(bpSd.getRoot(),
              "Cannot rollback to a newer state.\nDatanode previous state: LV = "
                  + prevInfo.getLayoutVersion() + " CTime = " + prevInfo.getCTime()
                  + " is newer than the namespace state: LV = "
                  + HdfsServerConstants.DATANODE_LAYOUT_VERSION + " CTime = " + nsInfo.getCTime());
        }
    
        LOG.info("Rolling back storage directory {}.\n   target LV = {}; target "
                + "CTime = {}", bpSd.getRoot(), nsInfo.getLayoutVersion(),
            nsInfo.getCTime());
    
        //确认removed.tmp不存在
        File tmpDir = bpSd.getRemovedTmp();
        assert !tmpDir.exists() : "removed.tmp directory must not exist.";
        // 1. 先将current目录改名为removed.tmp
        // 1. rename current to tmp
        File curDir = bpSd.getCurrentDir();
        assert curDir.exists() : "Current directory must exist.";
        rename(curDir, tmpDir);
        // 2. 将previous目录重命名为current
        // 2. rename previous to current
        rename(prevDir, curDir);
    
        // 3. 删除removed.tmp， 完成回滚操作
        // 3. delete removed.tmp dir
        deleteDir(tmpDir);
        LOG.info("Rollback of {} is complete", bpSd.getRoot());
      }

# 四.doFinalize() #

doFinalize()方法用于执行升级提交操作。 需要注意的是， doFinalize()方法的入口并不是doTransition()， 而是Namenode通过心跳携带回的名字节点指令。 doFinalize()方法的执行逻辑也是比较简单的， 首先将previous目录改名为finalized.tmp， 然后删除finalized.tmp即完成了升级。 这里的一个小优化是finalized.tmp的删除操作是通过单独启动一个新的删除线程执行的， 这是由于finalized.tmp文件夹中可能包含比较多的数据块， 删除操作会很慢而导致doFinalize()方法阻塞， 所以这里单独启动了一个删除线程执行finalized.tmp的删除操作。

/*
       * Finalize the block pool storage by deleting <BP>/previous directory
       * that holds the snapshot.
       */
      void doFinalize(File dnCurDir) throws IOException {
    
        if (dnCurDir == null) {
          return; //we do nothing if the directory is null
        }
        File bpRoot = getBpRoot(blockpoolID, dnCurDir);
        StorageDirectory bpSd = new StorageDirectory(bpRoot);
    
    
        // block pool level previous directory
        File prevDir = bpSd.getPreviousDir();
        // 如果previous目录不存在， 则没有必要执行提交操作， 直接退出执行
        if (!prevDir.exists()) {
          return; // already finalized
        }
        final String dataDirPath = bpSd.getRoot().getCanonicalPath();
        LOG.info("Finalizing upgrade for storage directory {}.\n   cur LV = {}; "
                + "cur CTime = {}", dataDirPath, this.getLayoutVersion(),
            this.getCTime());
        assert bpSd.getCurrentDir().exists() : "Current directory must exist.";
        //1． 将previous目录改名为finalized.tmp
        //1.  rename previous to finalized.tmp
        final File tmpDir = bpSd.getFinalizedTmp();
        rename(prevDir, tmpDir);
    
        // 2． 启动一个新的线程， 并发删除finalized.tmp文件夹
        // 2.  delete finalized.tmp dir in a separate thread
        new Daemon(new Runnable() {
          @Override
          public void run() {
            try {
              deleteDir(tmpDir);
            } catch (IOException ex) {
              LOG.error("Finalize upgrade for {} failed.", dataDirPath, ex);
            }
            LOG.info("Finalize upgrade for {} is complete.", dataDirPath);
          }
    
          @Override
          public String toString() {
            return "Finalize " + dataDirPath;
          }
        }).start();
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/6e7321e0a1474509a8752f0b9f632b6d.png