Namenode 会定期将文件系统的命名空间（文件目录树、文件／目录元信息）保存到 fsimage 文件中，以防止Namenode 掉电或者进程崩溃。但如果 Namenode 实时地将内存中的元数据同步到 fsimage 文件中，将会非常消耗资源且造成 Namenode 运行缓慢。所以 Namenode 会先将命名空间的修改操作保存在 editlog 文件中， 然后定期合并 fsimage 和 editlog 文件

FSImage类主要实现了以下功能：

  • 保存命名空间：将当前时刻 Namenode 内存中的命名空间保存到 fsimage 文件中
  • 加载 fsimage 文件：将磁盘上 fsimage 文件中保存的命名空间加载到 Namenode 内存中，这个操作是保存命名空间操作的逆操作
  • 加载 editlog 文件：Namenode 加载了 fsimage 文件后，内存中只包含了命名空间在保存 fsimage 文件时的信息，Namenode 还需要加载后续对命名空间的修改操作，即 editlog 文件中记录的内容。所以 FSImage 类还提供了加载 editlog 文件到 Namenode 内存中的功能

构建和初始化

FSImage 是在 FSNamesystem 中的 loadFromDisk 方法进行初始化的

  1. static FSNamesystem loadFromDisk(Configuration conf) throws IOException {
  3.     checkConfiguration(conf);
  5.     // 构建FSImage
  6.     FSImage fsImage = new FSImage(conf,
  7.         FSNamesystem.getNamespaceDirs(conf),
  8.         FSNamesystem.getNamespaceEditsDirs(conf));
  10.     FSNamesystem namesystem = new FSNamesystem(conf, fsImage, false);
  11.     StartupOption startOpt = NameNode.getStartupOption(conf);
  13.     if (startOpt == StartupOption.RECOVER) {
  14.       namesystem.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
  15.     }
  17.     long loadStart = monotonicNow();
  19.     try {
  20.       namesystem.loadFSImage(startOpt);
  21.     } catch (IOException ioe) {
  22.       LOG.warn("Encountered exception loading fsimage", ioe);
  23.       fsImage.close();
  24.       throw ioe;
  25.     }
  27.     long timeTakenToLoadFSImage = monotonicNow() - loadStart;
  28.     LOG.info("Finished loading FSImage in " + timeTakenToLoadFSImage + " msecs");
  29.     NameNodeMetrics nnMetrics = NameNode.getNameNodeMetrics();
  31.     if (nnMetrics != null) {
  32.       nnMetrics.setFsImageLoadTime((int) timeTakenToLoadFSImage);
  33.     }
  34.     namesystem.getFSDirectory().createReservedStatuses(namesystem.getCTime());
  36.     return namesystem;
  37. }
  1. protected FSImage(Configuration conf,
  2.                     Collection<URI> imageDirs,
  3.                     List<URI> editsDirs)
  4.       throws IOException {
  5.     this.conf = conf;
  7.     //构建NNStorage ==> NNStorage负责管理NameNode使用的 StorageDirectories。
  8.     storage = new NNStorage(conf, imageDirs, editsDirs);
  10.     // dfs.namenode.name.dir.restore  默认: false
  11.     // 设置为true可使NameNode尝试恢复以前失败的dfs.NameNode.name.dir。
  12.     // 启用后，将在检查点期间尝试恢复任何失败的目录。
  13.     if(conf.getBoolean(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_RESTORE_KEY,
  14.                        DFSConfigKeys.DFS_NAMENODE_NAME_DIR_RESTORE_DEFAULT)) {
  15.         storage.setRestoreFailedStorage(true);
  16.     }
  19.     // 构建 FSEditLog
  20.     this.editLog = FSEditLog.newInstance(conf, storage, editsDirs);
  22.     archivalManager = new NNStorageRetentionManager(conf, storage, editLog);
  23. }

保存命名空间

FSImage 类最重要的功能之一就是将当前时刻 Namenode 的命名空间保存到 fsimage 文件中

SaveFSImageInAllDirs

Namenode 可以定义多个存储路径来保存 fsimage 文件，对于每一个存储路径，saveFSImageInAllDirs() 方法都会启动一个线程负责在这个路径上保存 fsimage 文件。同时，为了防止保存过程中出现错误，命名空间信息首先会被保存在一个 fsimage.ckpt 文件中，当保存操作全部完成之后，才会将 fsimage.ckpt 重命名为 fsimage 文件。 之后 saveFSImageInAllDirs() 方法会清理 Namenode 元数据存储文件夹中过期的 editlog 文件和 fsimage 文件

  1. /**
  2.    *
  3.    * @param source
  4.    * @param nnf
  5.    * @param txid
  6.    * @param canceler
  7.    * @throws IOException
  8.    */
  9. private synchronized void saveFSImageInAllDirs(FSNamesystem source,
  10.       NameNodeFile nnf, long txid, Canceler canceler) throws IOException {
  12.     StartupProgress prog = NameNode.getStartupProgress();
  13.     prog.beginPhase(Phase.SAVING_CHECKPOINT);
  14.     if (storage.getNumStorageDirs(NameNodeDirType.IMAGE) == 0) {
  15.       throw new IOException("No image directories available!");
  16.     }
  17.     if (canceler == null) {
  18.       canceler = new Canceler();
  19.     }
  21.     // 构造保存命名空间操作的上下文
  22.     SaveNamespaceContext ctx = new SaveNamespaceContext(source, txid, canceler);
  24.     try {
  26.       // 在每一个保存路径上启动一个线程，该线程使用FSImageSaver类保存fsimage文件
  27.       List<Thread> saveThreads = new ArrayList<Thread>();
  28.       // save images into current
  29.       for (Iterator<StorageDirectory> it = storage.dirIterator(NameNodeDirType.IMAGE); 
  30.            it.hasNext();) {
  31.         StorageDirectory sd = it.next();
  33.         // 命名空间具体的保存操作是由 FSImageSaver 这个类来承担的，
  34.         // FSImageSaver是FSImage中的内部类，也是一个线程类，
  35.         // 它的run()方法调用了saveFSImage()方法来保存fsimage文件。
  36.         FSImageSaver saver = new FSImageSaver(ctx, sd, nnf);
  38.         Thread saveThread = new Thread(saver, saver.toString());
  40.         saveThreads.add(saveThread);
  41.         saveThread.start();
  42.       }
  44.       // 等待所有线程执行完毕
  45.       waitForThreads(saveThreads);
  47.       saveThreads.clear();
  48.       storage.reportErrorsOnDirectories(ctx.getErrorSDs());
  50.       // 保存文件失败则抛出异常
  51.       if (storage.getNumStorageDirs(NameNodeDirType.IMAGE) == 0) {
  52.         throw new IOException(
  53.           "Failed to save in any storage directories while saving namespace.");
  54.       }
  55.       if (canceler.isCancelled()) {
  56.         deleteCancelledCheckpoint(txid);
  57.         ctx.checkCancelled(); // throws
  58.         assert false : "should have thrown above!";
  59.       }
  61.       // 将fsimage.ckpt 改名为 fsimage
  62.       renameCheckpoint(txid, NameNodeFile.IMAGE_NEW, nnf, false);
  64.       // Since we now have a new checkpoint, we can clean up some
  65.       // old edit logs and checkpoints.
  66.       // Do not purge anything if we just wrote a corrupted FsImage.
  67.       if (!exitAfterSave.get()) {
  69.         // 我们已经完成了fsimage的保存，那么可以将存储上的一部分editlog和fsimage删除
  70.         // 如果没有成功,则失败.
  71.         purgeOldStorage(nnf);
  72.         archivalManager.purgeCheckpoints(NameNodeFile.IMAGE_NEW);
  73.       }
  74.     } finally {
  75.       // Notify any threads waiting on the checkpoint to be canceled
  76.       // that it is complete.
  78.       // 通知所有等待的线程
  79.       ctx.markComplete();
  80.       ctx = null;
  81.     }
  83.     prog.endPhase(Phase.SAVING_CHECKPOINT);
  85. }

FSImageSaver

命名空间具体的保存操作是由 FSImageSaver 这个类来承担的，FSImageSaver 是 FSImage 中的内部类，也是一个线程类，它的 run() 方法调用了 saveFSImage() 方法来保存 fsimage 文件

  1. private class FSImageSaver implements Runnable {
  2.     private final SaveNamespaceContext context;
  3.     private final StorageDirectory sd;
  4.     private final NameNodeFile nnf;
  6.     public FSImageSaver(SaveNamespaceContext context, StorageDirectory sd,
  7.         NameNodeFile nnf) {
  8.       this.context = context;
  9.       this.sd = sd;
  10.       this.nnf = nnf;
  11.     }
  13.     @Override
  14.     public void run() {
  15.       // Deletes checkpoint file in every storage directory when shutdown.
  16.       Runnable cancelCheckpointFinalizer = () -> {
  17.         try {
  18.           deleteCancelledCheckpoint(context.getTxId());
  19.           LOG.info("FSImageSaver clean checkpoint: txid={} when meet " +
  20.               "shutdown.", context.getTxId());
  21.         } catch (IOException e) {
  22.           LOG.error("FSImageSaver cancel checkpoint threw an exception:", e);
  23.         }
  24.       };
  25.       ShutdownHookManager.get().addShutdownHook(cancelCheckpointFinalizer,
  26.           SHUTDOWN_HOOK_PRIORITY);
  27.       try {
  28.         // 保存fsimage文件
  29.         System.out.println("context : "+ context);
  30.         System.out.println("sd : "+ sd);
  31.         System.out.println("nnf : "+ nnf);
  32.         saveFSImage(context, sd, nnf);
  33.       } catch (SaveNamespaceCancelledException snce) {
  34.         LOG.info("Cancelled image saving for " + sd.getRoot() +
  35.             ": " + snce.getMessage());
  36.         // don't report an error on the storage dir!
  37.       } catch (Throwable t) {
  38.         LOG.error("Unable to save image for " + sd.getRoot(), t);
  39.         context.reportErrorOnStorageDirectory(sd);
  40.         try {
  41.           deleteCancelledCheckpoint(context.getTxId());
  42.           LOG.info("FSImageSaver clean checkpoint: txid={} when meet " +
  43.               "Throwable.", context.getTxId());
  44.         } catch (IOException e) {
  45.           LOG.error("FSImageSaver cancel checkpoint threw an exception:", e);
  46.         }
  47.       }
  48.     }
  50.     @Override
  51.     public String toString() {
  52.       return "FSImageSaver for " + sd.getRoot() +
  53.              " of type " + sd.getStorageDirType();
  54.     }
  55.   }

SaveFSImage(context, sd, nnf)

SaveFSImage() 方法会使用一个 FSImageFormat.Saver 对象来完成保存操作，FSImageFormat.Saver 类会以fsimage 文件定义的格式保存 Namenode 的命名空间信息，需要注意命名空空间信息会先写入 fsimage.ckpt 文件中。saveFSImage() 方法还会生成 fsimage 文件的 md5 校验文件，以确保 fsimage 文件的正确性

  2.   /**
  3.    * 将 FS image 的内容保存到文件中。
  4.    * Save the contents of the FS image to the file.
  5.    */
  6. void saveFSImage(SaveNamespaceContext context, StorageDirectory sd,
  7.       NameNodeFile dstType) throws IOException {
  9.     // 获取当前命名空间中记录的最新事务的txid
  10.     long txid = context.getTxId();
  12.     // fsimage文件
  13.     File newFile = NNStorage.getStorageFile(sd, NameNodeFile.IMAGE_NEW, txid);
  15.     File dstFile = NNStorage.getStorageFile(sd, dstType, txid);
  17.     // FSImageFormatProtobuf.Saver类负责保存fsimage
  18.     FSImageFormatProtobuf.Saver saver = new FSImageFormatProtobuf.Saver(context);
  20.     // 压缩类
  21.     FSImageCompression compression = FSImageCompression.createCompression(conf);
  23.     // 调用Saver类保存fsimage文件
  24.     long numErrors = saver.save(newFile, compression);
  26.     if (numErrors > 0) {
  27.       // The image is likely corrupted.
  28.       LOG.error("Detected " + numErrors + " errors while saving FsImage " +
  29.           dstFile);
  30.       exitAfterSave.set(true);
  31.     }
  33.     // 保存MD5校验值
  34.     MD5FileUtils.saveMD5File(dstFile, saver.getSavedDigest());
  35.     storage.setMostRecentCheckpointInfo(txid, Time.now());
  37. }

saveFSImage() 方法构造了一个 FSImageFormatProtobuf.Saver 对象来保存命名空间，FSImageFormatProtobuf是一个工具类，它提供了以 protobuf 格式读取和写入 fsimage 文件的方法

  1. /**
  2.  * save()方法会打开fsimage文件的输出流并且获得文件通道，
  3.  * 然后调用saveInternal()方法将命名空间保存到fsimage文件中。
  4.  *
  5.  * @return number of non-fatal errors detected while writing the image.
  6.  * @throws IOException on fatal error.
  7.  */
  8. long save(File file, FSImageCompression compression) throws IOException {
  11.       FileOutputStream fout = new FileOutputStream(file);
  13.       fileChannel = fout.getChannel();
  15.       try {
  16.         LOG.info("Saving image file {} using {}", file, compression);
  17.         long startTime = monotonicNow();
  19.         // 保存到fsimage文件
  20.         long numErrors = saveInternal( fout, compression, file.getAbsolutePath());
  22.         LOG.info("Image file {} of size {} bytes saved in {} seconds {}.", file,
  23.             file.length(), (monotonicNow() - startTime) / 1000,
  24.             (numErrors > 0 ? (" with" + numErrors + " errors") : ""));
  25.         return numErrors;
  26.       } finally {
  27.         fout.close();
  28.       }
  29. }

FSImageFormatProtobuf.Saver

使用 protobuf 定义的 fsimage 文件的格式，它包括了4个部分的信息

  • MAGIC：fsimage 的文件头，是“HDFSIMG1”这个字符串的二进制形式，MAGIC 头标识了当前 fsimage 文件是使用 protobuf 格式序列化的。 FSImage 类在读取 fsimage 文件时，会先判断 fsimage 文件是否包含了MAGIC头，如果包含了则使用 protobuf 格式反序列化 fsimage 文件

  • SECTIONS：fsimage 文件会将同一类型的 Namenode 元信息保存在一个 section 中，例如将文件系统元信息保存在 NameSystemSection 中，将文件系统目录树中的所有 INode 信息保存在 INodeSection 中，将快照信息保存在 SnapshotSection 中等。fsimage 文件的第二个部分就是 Namenode 各类元信息对应的所有section，每类 section 中都包含了对应 Namenode 元信息的属性

  • FileSummary：FileSummary 记录了 fsimage 文件的元信息，以及 fsimage 文件保存的所有 section 的信息。 FileSummary 中的 ondiskVersion 字段记录了 fsimage 文件的版本号，layoutVersion 字段记录了当前HDFS 的文件系统布局版本号，codec 字段记录了 fsimage 文件的压缩编码，sections 字段则记录了 fsimage文件中各个 section 字段的元信息，每个 fsimage 文件中记录的 section 在 FileSummary 中都有一个与之对应的 section 字段。FileSummary 的 section 字段记录了对应的 fsimage 中 section 的名称、在 fsimage 文件中的长度，以及这个 section 在 fsimage 中的起始位置。FSImage 类在读取 fsimage 文件时，会先从 fsimage中读取出 FileSummary 部分，然后利用 FileSummary 记录的元信息指导 fsimage 文件的反序列化操作

  • FileSummaryLength：FileSummaryLength 记录了 FileSummary 在 fsimage 文件中所占的长度， FSImage 类在读取 fsimage 文件时，会首先读取 FileSummaryLength 获取 FileSummary 部分的长度， 然后根据这个长度从 fsimage 中反序列化出 FileSummary

FSImageFormatProtobuf.Saver 类就是以 protobuf 格式将 Namenode 的命名空间保存至 fsimage 文件的工具类。这个类的入口方法是 save() 方法。save() 方法会打开 fsimage 文件的输出流并且获得文件通道，然后调用saveInternal() 方法将命名空间保存到 fsimage 文件中

SaveInternal

saveInternal() 方法首先构造底层 fsimage 文件的输出流，构造 fsimage 文件的描述类 FileSummary，然后在FileSummary 中记录 ondiskVersion、layoutVersion、codec等信息

接下来 saveInternal() 方法依次向 fsimage 文件中写入命名空间信息、inode信息、快照信息、安全信息、缓存信息、StringTable信息等。上述信息都是以 section 为单位写入

saveInternal() 方法以 section 为单位写入元数据信息时，还会在 FileSummary 中记录这个 section 的长度，以及section 在 fsimage 文件中的起始位置等信息。当完成了所有 section 的写入后，FileSummary 对象也就构造完毕了。saveInternal() 最后会将 FileSummary 对象写入 fsimage 文件中

  1. private long saveInternal(FileOutputStream fout, FSImageCompression compression, 
  2.                           String filePath) throws IOException {
  4.       StartupProgress prog = NameNode.getStartupProgress();
  6.       // 构造输出流，一边写入数据，一边写入校验值
  7.       MessageDigest digester = MD5Hash.getDigester();
  9.       int layoutVersion = context.getSourceNamesystem().getEffectiveLayoutVersion();
  11.       underlyingOutputStream = new DigestOutputStream(new BufferedOutputStream(fout), 
  12.                                                       digester);
  14.       underlyingOutputStream.write(FSImageUtil.MAGIC_HEADER);
  16.       fileChannel = fout.getChannel();
  18.       // FileSummary为fsimage文件的描述部分，也是protobuf定义的
  19.       FileSummary.Builder b = FileSummary.newBuilder()
  20.           .setOndiskVersion(FSImageUtil.FILE_VERSION)
  21.           .setLayoutVersion(
  22.               context.getSourceNamesystem().getEffectiveLayoutVersion());
  24.       // 获取压缩格式，并装饰输出流
  25.       codec = compression.getImageCodec();
  27.       if (codec != null) {
  28.         b.setCodec(codec.getClass().getCanonicalName());
  29.         sectionOutputStream = codec.createOutputStream(underlyingOutputStream);
  30.       } else {
  31.         sectionOutputStream = underlyingOutputStream;
  32.       }
  34.       //保存命名空间信息
  35.       saveNameSystemSection(b);
  37.       // Check for cancellation right after serializing the name system section.
  38.       // Some unit tests, such as TestSaveNamespace#testCancelSaveNameSpace
  39.       // depends on this behavior.
  41.       // 检查是否取消了保存操作
  42.       context.checkCancelled();
  44.       Step step;
  46.       // Erasure coding policies should be saved before inodes
  47.       if (NameNodeLayoutVersion.supports(
  48.           NameNodeLayoutVersion.Feature.ERASURE_CODING, layoutVersion)) {
  50.         step = new Step(StepType.ERASURE_CODING_POLICIES, filePath);
  51.         prog.beginStep(Phase.SAVING_CHECKPOINT, step);
  53.         //保存 ErasureCoding 信息
  54.         saveErasureCodingSection(b);
  55.         prog.endStep(Phase.SAVING_CHECKPOINT, step);
  57.       }
  59.       //保存命名空间中的inode信息
  60.       step = new Step(StepType.INODES, filePath);
  62.       prog.beginStep(Phase.SAVING_CHECKPOINT, step);
  63.       // Count number of non-fatal errors when saving inodes and snapshots.
  64.       // 保存命名空间中的inode信息
  65.       long numErrors = saveInodes(b);
  67.       // 保存快照信息
  68.       numErrors += saveSnapshots(b);
  69.       prog.endStep(Phase.SAVING_CHECKPOINT, step);
  71.       // 保存安全信息
  72.       step = new Step(StepType.DELEGATION_TOKENS, filePath);
  73.       prog.beginStep(Phase.SAVING_CHECKPOINT, step);
  74.       saveSecretManagerSection(b);
  75.       prog.endStep(Phase.SAVING_CHECKPOINT, step);
  77.       // 保存缓存信息
  78.       step = new Step(StepType.CACHE_POOLS, filePath);
  79.       prog.beginStep(Phase.SAVING_CHECKPOINT, step);
  80.       saveCacheManagerSection(b);
  81.       prog.endStep(Phase.SAVING_CHECKPOINT, step);
  83.       // 保存StringTable
  84.       saveStringTableSection(b);
  86.       // We use the underlyingOutputStream to write the header. Therefore flush
  87.       // the buffered stream (which is potentially compressed) first.
  89.       // flush输出流
  90.       flushSectionOutputStream();
  92.       FileSummary summary = b.build();
  93.       //将FileSummary写入文件
  94.       saveFileSummary(underlyingOutputStream, summary);
  96.       //关闭底层输出流
  97.       underlyingOutputStream.close();
  99.       savedDigest = new MD5Hash(digester.digest());
  100.       return numErrors;
  102. }

写入文件头

fsimage 的文件头，是“HDFSIMG1”这个字符串的二进制形式。MAGIC 头标识了当前 fsimage 文件是使用protobuf 格式序列化的

  1. private long saveInternal(FileOutputStream fout, FSImageCompression compression, 
  2.                           String filePath) throws IOException {
  4.       StartupProgress prog = NameNode.getStartupProgress();
  6.       // 构造输出流，一边写入数据，一边写入校验值
  7.       MessageDigest digester = MD5Hash.getDigester();
  9.       int layoutVersion = context.getSourceNamesystem().getEffectiveLayoutVersion();
  11.       // 构造underlyingOutputStream
  12.       underlyingOutputStream = new DigestOutputStream(new BufferedOutputStream(fout), 
  13.                                                       digester);
  14.       // 写入头信息
  15.       underlyingOutputStream.write(FSImageUtil.MAGIC_HEADER);
  17.       // 打开文件通道
  18.       fileChannel = fout.getChannel();

构建FileSummary对象

  1. // FileSummary为fsimage文件的描述部分，也是protobuf定义的
  2. FileSummary.Builder b = FileSummary.newBuilder()
  3.   .setOndiskVersion(FSImageUtil.FILE_VERSION)
  4.   .setLayoutVersion(
  5.       context.getSourceNamesystem().getEffectiveLayoutVersion());

处理数据的压缩方式

  1.   // 获取压缩格式，并装饰输出流
  2.   codec = compression.getImageCodec();
  4.   if (codec != null) {
  5.     b.setCodec(codec.getClass().getCanonicalName());
  6.     sectionOutputStream = codec.createOutputStream(underlyingOutputStream);
  7.   } else {
  8.     sectionOutputStream = underlyingOutputStream;
  9.   }

Save

SaveInternal() 方法调用了多个 save*() 方法来记录不同 section 的元数据信息，这些方法除了在 fsimage 文件中写入对应种类的元数据信息外，还会在 FileSummary 中记录 section 的大小，以及在fsimage中的起始位置

以 saveINodes() 方法举例，该方法构造了一个 FSImageFormatPBINode.Saver 对象，并调用这个对象对应的方法保存文件系统目录树中的 INode 信息、INodeDirectory 信息，以及处于构建状态的文件信息

  1. private long saveInodes(FileSummary.Builder summary) throws IOException {
  3.   FSImageFormatPBINode.Saver saver = new FSImageFormatPBINode.Saver(this,summary);
  5.   // 保存INode信息是由FSImageFormatPBINode.Saver.serializeINodeSection()方法实现的
  6.   saver.serializeINodeSection(sectionOutputStream);
  7.   // 保存info目录信息
  8.   saver.serializeINodeDirectorySection(sectionOutputStream);
  9.   // 租约管理
  10.   saver.serializeFilesUCSection(sectionOutputStream);
  12.   return saver.getNumImageErrors();
  13. }

serializeINodeSection() 方法会首先构造一个 INodeSection 对象，记录文件系统目录树中保存的最后一个 inode的 inodeid，以及命名空间中所有 inode 的个数。之后迭代处理将所有 inode 信息写入 fsimage 文件中，最后将INodeSection 的属性信息记录在 FileSummary 中

  1. void serializeINodeSection(OutputStream out) throws IOException {
  3.       INodeMap inodesMap = fsn.dir.getINodeMap();
  5.       // 构造一个INodeSection，保存最后一个inode的inodeid，以及这个命名空间中所有inode的个数
  6.       INodeSection.Builder b = INodeSection.newBuilder()
  7.           .setLastInodeId(fsn.dir.getLastInodeId()).setNumInodes(inodesMap.size());
  9.       INodeSection s = b.build();
  11.       // 序列化至输出流
  12.       s.writeDelimitedTo(out);
  14.       int i = 0;
  16.       // 迭代处理inodeMap中所有的inode，调用save()方法将inode信息保存到fsimage中
  17.       Iterator<INodeWithAdditionalFields> iter = inodesMap.getMapIterator();
  18.       while (iter.hasNext()) {
  19.         INodeWithAdditionalFields n = iter.next();
  20.         // 将所有inode信息写入fsimage文件中
  21.         save(out, n);
  22.         ++i;
  23.         if (i % FSImageFormatProtobuf.Saver.CHECK_CANCEL_INTERVAL == 0) {
  24.           context.checkCancelled();
  25.         }
  26.       }
  28.       // 调用commitSection()方法在FileSummary中写入inode section
  29.       parent.commitSection(summary, FSImageFormatProtobuf.SectionName.INODE);
  30.     }

save() 方法首先将当前 INode 对象分为目录、文件以及符号链接等几类，然后调用各个类型对应的 save() 重载方法。重载方法的实现也非常简单，就是构造不同的 protobuf Builder 类，然后设置相应字段的值，并将序列化之后的对象写入 fsimage 文件的输出流中

以INodeFile为例，首先构造 protobuf Builder 类，INodeSection.INodeFile.Builder，然后设置blocks，也就是当前文件有哪些数据块，如果当前的 INodeFile 处于构建状态，则设置对应的构建信息。最后将序列化后的 inode 信息写入输出流中

  1.     private void save(OutputStream out, INode n) throws IOException {
  2.       if (n.isDirectory()) {
  3.         save(out, n.asDirectory());
  4.       } else if (n.isFile()) {
  5.         save(out, n.asFile());
  6.       } else if (n.isSymlink()) {
  7.         save(out, n.asSymlink());
  8.       }
  9.     }
  11.     private void save(OutputStream out, INodeDirectory n) throws IOException {
  12.       INodeSection.INodeDirectory.Builder b = buildINodeDirectory(n,
  13.           parent.getSaverContext());
  14.       INodeSection.INode r = buildINodeCommon(n)
  15.           .setType(INodeSection.INode.Type.DIRECTORY).setDirectory(b).build();
  16.       r.writeDelimitedTo(out);
  17.     }
  19.     // 首先构造protobuf Builder类——INodeSection.INodeFile.Builder，
  20.     // 然后设置blocks——也就是当前文件有哪些数据块，
  21.     // 如果当前的INodeFile处于构建状态， 则设置对应的构建信息。
  22.     //
  23.     // 最后将序列化后的inode信息写入输出流中。
  25.     private void save(OutputStream out, INodeFile n) throws IOException {
  26.       INodeSection.INodeFile.Builder b = buildINodeFile(n,
  27.           parent.getSaverContext());
  28.       BlockInfo[] blocks = n.getBlocks();
  30.       if (blocks != null) {
  31.         for (Block block : n.getBlocks()) {
  32.           b.addBlocks(PBHelperClient.convert(block));
  33.         }
  34.       }
  36.       FileUnderConstructionFeature uc = n.getFileUnderConstructionFeature();
  37.       if (uc != null) {
  38.         INodeSection.FileUnderConstructionFeature f =
  39.             INodeSection.FileUnderConstructionFeature
  40.             .newBuilder().setClientName(uc.getClientName())
  41.             .setClientMachine(uc.getClientMachine()).build();
  42.         b.setFileUC(f);
  43.       }
  45.       INodeSection.INode r = buildINodeCommon(n)
  46.           .setType(INodeSection.INode.Type.FILE).setFile(b).build();
  47.       r.writeDelimitedTo(out);
  48.     }
  51.     private void save(OutputStream out, INodeSymlink n) throws IOException {
  52.       INodeSection.INodeSymlink.Builder b = INodeSection.INodeSymlink
  53.           .newBuilder()
  54.           .setPermission(buildPermissionStatus(n))
  55.           .setTarget(ByteString.copyFrom(n.getSymlink()))
  56.           .setModificationTime(n.getModificationTime())
  57.           .setAccessTime(n.getAccessTime());
  59.       INodeSection.INode r = buildINodeCommon(n)
  60.           .setType(INodeSection.INode.Type.SYMLINK).setSymlink(b).build();
  61.       r.writeDelimitedTo(out);
  62.     }

将FileSummary写入文件

  1. private static void saveFileSummary(OutputStream out, FileSummary summary)
  2.     throws IOException {
  3.     summary.writeDelimitedTo(out);
  4.     int length = getOndiskTrunkSize(summary);
  5.     byte[] lengthBytes = new byte[4];
  6.     ByteBuffer.wrap(lengthBytes).asIntBuffer().put(length);
  7.     out.write(lengthBytes);
  8. }

FSImage.loadFSImage

当 Namenode 启动时，首先会将 fsimage 文件中记录的命名空间加载到 Namenode 内存中，然后再一条一条地将 editlog 文件中记录的更新操作加载并合并到命名空间中
20200927004349193.png
Namenode 会等待各个 Datanode 向自己汇报数据块信息来组装 blockMap，从而离开安全模式。Namenode 每次启动时都会调用 FSImage.loadFSImage() 方法执行加载 fsimage 和 editlog 文件的操作

这里需要关注两个地方，一个是加载 FSImage 文件 fsImage.recoverTransitionRead(startOpt, this, recovery); 另外一个是根据条件判断是否要合并edits文件——间隔周期1小时或者 edits 数量操作100w

loadFSImage方法，其实就是加载最后一个fsimage文件。加载分两步:

  • 获取加载器：FSImageFormat.LoaderDelegator

  • 加载文件：loader.load(curFile, requireSameLayoutVersion)

    1. private void loadFSImage(File curFile, MD5Hash expectedMd5,
    2.     FSNamesystem target, MetaRecoveryContext recovery,
    3.     boolean requireSameLayoutVersion) throws IOException {
    4.   // BlockPoolId is required when the FsImageLoader loads the rolling upgrade
    5.   // information. Make sure the ID is properly set.
    6.   target.setBlockPoolId(this.getBlockPoolID());
    8.   // 获取加载器 FSImageFormat.LoaderDelegator
    9.   FSImageFormat.LoaderDelegator loader = FSImageFormat.newLoader(conf, target);
    11.   // 加载文件
    12.   loader.load(curFile, requireSameLayoutVersion);
    14.   // Check that the image digest we loaded matches up with what we expected
    15.   MD5Hash readImageMd5 = loader.getLoadedImageMd5();
    16.   if (expectedMd5 != null &&
    17.       !expectedMd5.equals(readImageMd5)) {
    18.     throw new IOException("Image file " + curFile +
    19.         " is corrupt with MD5 checksum of " + readImageMd5 +
    20.         " but expecting " + expectedMd5);
    21.   }
    23.   long txId = loader.getLoadedImageTxId();
    24.   LOG.info("Loaded image for txid " + txId + " from " + curFile);
    25.   lastAppliedTxId = txId;
    26.   storage.setMostRecentCheckpointInfo(txId, curFile.lastModified());
    27. }
    1. public void load(File file, boolean requireSameLayoutVersion)
    2.       throws IOException {
    3.     Preconditions.checkState(impl == null, "Image already loaded!");
    5.     FileInputStream is = null;
    6.     try {
    7.       is = new FileInputStream(file);
    8.       byte[] magic = new byte[FSImageUtil.MAGIC_HEADER.length];
    9.       IOUtils.readFully(is, magic, 0, magic.length);
    11.       // fsimage文件中包括 magicHeader，使用的是protobuf序列化方式
    12.       if (Arrays.equals(magic, FSImageUtil.MAGIC_HEADER)) {
    14.         // 构造FSImageFormatProtobuf.Loader加载fsimage文件
    15.         FSImageFormatProtobuf.Loader loader = new FSImageFormatProtobuf.Loader(
    16.             conf, fsn, requireSameLayoutVersion);
    17.         impl = loader;
    18.         loader.load(file);
    20.       } else {
    21.         // 否则构造FSImageFormat.Loader加载fsimage文件
    22.         Loader loader = new Loader(conf, fsn);
    23.         impl = loader;
    24.         loader.load(file);
    25.       }
    26.     } finally {
    27.       IOUtils.cleanupWithLogger(LOG, is);
    28.     }
    29. }

    这里面是构造 FSImageFormatProtobuf.Loader 对象，使用它的 load 方法加载 fsimage 文件。在 load 方法中最终调用了 loadInternal(raFile, fin); 方法。这个方法是加载 fsimage 文件的相对最底层的方法了

在加载 fsimage 操作中，最终会调用 FSImageFormatProtobuf.Loader 作为 fsimage 文件的加载类FSImageFormatProtobuf.Loader.loadInternal() 方法执行了加载 fsimage 文件的操作，loadInternal() 方法会打开fsimage 文件通道，然后读取 fsimage 文件中的 FileSummary 对象，FileSummary 对象中记录了 fsimage 中保存的所有section的信息。loadInternal() 会对 FileSummary 对象中保存的 section 排序， 然后遍历每个section并调用对应的方法从 fsimage 文件中加载这个 section
20200927005049903.png

  1. // loadInternal()方法会打开fsimage文件通道，
  2. // 然后读取fsimage文件中的FileSummary对象，
  3. // FileSummary对象中记录了fsimage中保存的所有section的信息。
  4. // loadInternal()会对FileSummary对象中保存的section排序，
  5. // 然后遍历每个section并调用对应的方法从fsimage文件中加载这个section。
  6. private void loadInternal(RandomAccessFile raFile, FileInputStream fin)
  7.         throws IOException {
  8.       if (!FSImageUtil.checkFileFormat(raFile)) {
  9.         throw new IOException("Unrecognized file format");
  10.       }
  12.       // 从fsimage文件末尾加载FileSummary，也就是fsimage文件内容的描述
  13.       FileSummary summary = FSImageUtil.loadSummary(raFile);
  15.       if (requireSameLayoutVersion && summary.getLayoutVersion() !=
  16.           HdfsServerConstants.NAMENODE_LAYOUT_VERSION) {
  17.         throw new IOException("Image version " + summary.getLayoutVersion() +
  18.             " is not equal to the software version " +
  19.             HdfsServerConstants.NAMENODE_LAYOUT_VERSION);
  20.       }
  22.       // 获取通道
  23.       FileChannel channel = fin.getChannel();
  25.       // 构造FSImageFormatPBINode.Loader和FSImageFormatPBSnapshot.
  26.       // Loader加载INode以及Snapshot
  27.       FSImageFormatPBINode.Loader inodeLoader = new FSImageFormatPBINode.Loader(
  28.           fsn, this);
  29.       FSImageFormatPBSnapshot.Loader snapshotLoader = new 
  30.           FSImageFormatPBSnapshot.Loader(fsn, this);
  32.       // 对fsimage文件描述中记录的sections进行排序
  33.       ArrayList<FileSummary.Section> sections = Lists.newArrayList(summary
  34.           .getSectionsList());
  35.       Collections.sort(sections, new Comparator<FileSummary.Section>() {
  36.         @Override
  37.         public int compare(FileSummary.Section s1, FileSummary.Section s2) {
  38.           SectionName n1 = SectionName.fromString(s1.getName());
  39.           SectionName n2 = SectionName.fromString(s2.getName());
  40.           if (n1 == null) {
  41.             return n2 == null ? 0 : -1;
  42.           } else if (n2 == null) {
  43.             return -1;
  44.           } else {
  45.             return n1.ordinal() - n2.ordinal();
  46.           }
  47.         }
  48.       });
  50.       StartupProgress prog = NameNode.getStartupProgress();
  51.       /**
  52.        * beginStep() and the endStep() calls do not match the boundary of the
  53.        * sections. This is because that the current implementation only allows
  54.        * a particular step to be started for once.
  55.        */
  56.       Step currentStep = null;
  58.       // 遍历每个section，并调用对应的方法加载这个 section
  59.       for (FileSummary.Section s : sections) {
  61.         // 在通道中定位这个 section 的起始位置
  62.         channel.position(s.getOffset());
  63.         InputStream in = new BufferedInputStream(new LimitInputStream(fin,
  64.             s.getLength()));
  66.         in = FSImageUtil.wrapInputStreamForCompression(conf,
  67.             summary.getCodec(), in);
  69.         String n = s.getName();
  71.         // 调用对应的方法加载不同的section
  72.         switch (SectionName.fromString(n)) {
  73.         case NS_INFO:
  74.           loadNameSystemSection(in);
  75.           break;
  76.         case STRING_TABLE:
  77.           loadStringTableSection(in);
  78.           break;
  79.         case INODE: {
  80.           currentStep = new Step(StepType.INODES);
  81.           prog.beginStep(Phase.LOADING_FSIMAGE, currentStep);
  82.           inodeLoader.loadINodeSection(in, prog, currentStep);
  83.         }
  84.           break;
  85.         case INODE_REFERENCE:
  86.           snapshotLoader.loadINodeReferenceSection(in);
  87.           break;
  88.         case INODE_DIR:
  89.           inodeLoader.loadINodeDirectorySection(in);
  90.           break;
  91.         case FILES_UNDERCONSTRUCTION:
  92.           inodeLoader.loadFilesUnderConstructionSection(in);
  93.           break;
  94.         case SNAPSHOT:
  95.           snapshotLoader.loadSnapshotSection(in);
  96.           break;
  97.         case SNAPSHOT_DIFF:
  98.           snapshotLoader.loadSnapshotDiffSection(in);
  99.           break;
  100.         case SECRET_MANAGER: {
  101.           prog.endStep(Phase.LOADING_FSIMAGE, currentStep);
  102.           Step step = new Step(StepType.DELEGATION_TOKENS);
  103.           prog.beginStep(Phase.LOADING_FSIMAGE, step);
  104.           loadSecretManagerSection(in, prog, step);
  105.           prog.endStep(Phase.LOADING_FSIMAGE, step);
  106.         }
  107.           break;
  108.         case CACHE_MANAGER: {
  109.           Step step = new Step(StepType.CACHE_POOLS);
  110.           prog.beginStep(Phase.LOADING_FSIMAGE, step);
  111.           loadCacheManagerSection(in, prog, step);
  112.           prog.endStep(Phase.LOADING_FSIMAGE, step);
  113.         }
  114.           break;
  115.         case ERASURE_CODING:
  116.           Step step = new Step(StepType.ERASURE_CODING_POLICIES);
  117.           prog.beginStep(Phase.LOADING_FSIMAGE, step);
  118.           loadErasureCodingSection(in);
  119.           prog.endStep(Phase.LOADING_FSIMAGE, step);
  120.           break;
  121.         default:
  122.           LOG.warn("Unrecognized section {}", n);
  123.           break;
  124.         }
  125.       }
  126. }

对于不同类型的 section，loadInternal() 方法会调用不同的方法加载这个 section，例如对于 INodeSection 会调用 InodeLoader.loadINodeSection() 方法加载。load*() 方法的实现都比较简单，就是按照 protobuf 格式加载不同的 section。慢慢恢复/构建 FSNamesystem 对象中的内容

  1. void loadINodeSection(InputStream in, StartupProgress prog,
  2.         Step currentStep) throws IOException {
  4.       INodeSection s = INodeSection.parseDelimitedFrom(in);
  6.       fsn.dir.resetLastInodeId(s.getLastInodeId());
  8.       long numInodes = s.getNumInodes();
  10.       LOG.info("Loading " + numInodes + " INodes.");
  11.       prog.setTotal(Phase.LOADING_FSIMAGE, currentStep, numInodes);
  12.       Counter counter = prog.getCounter(Phase.LOADING_FSIMAGE, currentStep);
  13.       for (int i = 0; i < numInodes; ++i) {
  14.         INodeSection.INode p = INodeSection.INode.parseDelimitedFrom(in);
  15.         if (p.getId() == INodeId.ROOT_INODE_ID) {
  16.           // 加载root
  17.           loadRootINode(p);
  18.         } else {
  19.           // 加载子节点
  20.           INode n = loadINode(p);
  21.           dir.addToInodeMap(n);
  22.         }
  23.         counter.increment();
  24.       }
  25. }
  1. // 构建目录体系
  2. void loadINodeDirectorySection(InputStream in) throws IOException {
  3.       final List<INodeReference> refList = parent.getLoaderContext()
  4.           .getRefList();
  5.       while (true) {
  6.         INodeDirectorySection.DirEntry e = INodeDirectorySection.DirEntry
  7.             .parseDelimitedFrom(in);
  8.         // note that in is a LimitedInputStream
  9.         if (e == null) {
  10.           break;
  11.         }
  12.         INodeDirectory p = dir.getInode(e.getParent()).asDirectory();
  13.         for (long id : e.getChildrenList()) {
  14.           INode child = dir.getInode(id);
  15.           addToParent(p, child);
  16.         }
  17.         for (int refId : e.getRefChildrenList()) {
  18.           INodeReference ref = refList.get(refId);
  19.           addToParent(p, ref);
  20.         }
  21.       }
  22. }

加载Edits文件

FSImage.loadEdits() 方法会构造一个 FSEditLogLoader 对象。然后遍历 Namenode 所有存储路径上保存的 editlog 文件的输入流并调用 FSEditLogLoader.loadFSEdits() 方法加载指定路径上的 editlog 文件

  1. public long loadEdits(Iterable<EditLogInputStream> editStreams,
  2.       FSNamesystem target, long maxTxnsToRead,
  3.       StartupOption startOpt, MetaRecoveryContext recovery)
  4.       throws IOException {
  5.     LOG.debug("About to load edits:\n  " + Joiner.on("\n  ").join(editStreams));
  6.     StartupProgress prog = NameNode.getStartupProgress();
  7.     prog.beginPhase(Phase.LOADING_EDITS);
  9.     //记录命名空间中加载的最新的事务id
  10.     long prevLastAppliedTxId = lastAppliedTxId;
  11.     long remainingReadTxns = maxTxnsToRead;
  12.     try {
  14.       //构造FSEditLogLoader对象用于加栽editlog文件
  15.       FSEditLogLoader loader = new FSEditLogLoader(target, lastAppliedTxId);
  18.       //遍历所有存储路径上editlog文件对应的输入流
  19.       // Load latest edits
  20.       for (EditLogInputStream editIn : editStreams) {
  21.         LogAction logAction = loadEditLogHelper.record();
  22.         if (logAction.shouldLog()) {
  23.           String logSuppressed = "";
  24.           if (logAction.getCount() > 1) {
  25.             logSuppressed = "; suppressed logging for " +
  26.                 (logAction.getCount() - 1) + " edit reads";
  27.           }
  28.           LOG.info("Reading " + editIn + " expecting start txid #" +
  29.               (lastAppliedTxId + 1) + logSuppressed);
  30.         }
  31.         try {
  33.           //调用FSEditLogLoader.loadFSEdits()从某个存储路径上的editlog文件加载修改操作
  34.           remainingReadTxns -= loader.loadFSEdits(editIn, lastAppliedTxId + 1,
  35.                   remainingReadTxns, startOpt, recovery);
  36.         } finally {
  38.           // lastAppliedTxId记录从editlog加载的最新的事务id
  39.           // Update lastAppliedTxId even in case of error, since some ops may
  40.           // have been successfully applied before the error.
  41.           lastAppliedTxId = loader.getLastAppliedTxId();
  42.         }
  45.         // If we are in recovery mode, we may have skipped over some txids.
  46.         if (editIn.getLastTxId() != HdfsServerConstants.INVALID_TXID
  47.             && recovery != null) {
  48.           lastAppliedTxId = editIn.getLastTxId();
  49.         }
  50.         if (remainingReadTxns <= 0) {
  51.           break;
  52.         }
  53.       }
  54.     } finally {
  55.       //关闭所有editlog文件的输入流
  56.       FSEditLog.closeAllStreams(editStreams);
  57.     }
  58.     prog.endPhase(Phase.LOADING_EDITS);
  59.     return lastAppliedTxId - prevLastAppliedTxId;
  60.   }