前言

错误注入技术的原理

拦截的层面可以根据应用程序本身而定，比较简单的可以在应用程序同级别进行拦截，比较高级的是在OS层面，在系统回调方法处进行拦截。下图是原理图。

钩子拦截示意图

HDFS内部的几种错误注入方法

结合应用程序规则的错误注入

第一类，改变文件的属性，比如说变为只读。代码如下：

...
    // fail the volume
    // delete/make non-writable one of the directories (failed volume)
    data_fail = new File(dataDir, "data3");
    failedDir = MiniDFSCluster.getFinalizedDir(data_fail,
        cluster.getNamesystem().getBlockPoolId());
    if (failedDir.exists() &&
        //!FileUtil.fullyDelete(failedDir)
        !deteteBlocks(failedDir)
        ) {
      throw new IOException("Could not delete hdfs directory ‘" + failedDir + "‘");
    }
    // 设置文件目录为只读模式
    data_fail.setReadOnly();
    failedDir.setReadOnly();
    System.out.println("Deleteing " + failedDir.getPath() + "; exist=" + failedDir.exists());

    // access all the blocks on the "failed" DataNode,
    // we need to make sure that the "failed" volume is being accessed -
    // and that will cause failure, blocks removal, "emergency" block report
    // 触发一次文件文件访问操作
    triggerFailure(filename, filesize);
    ...

设置为只读方式，在随后的读块操作中，将会发现这个错误。

第二类，重命名原始文件，使系统认为原始文件发生丢失。重命名的好处是还可以再立刻恢复回去，而不是真的删除掉了。操作代码如下：

  public static void injectDataDirFailure(File... dirs) throws IOException {
    for (File dir : dirs) {
      File renamedTo = new File(dir.getPath() + DIR_FAILURE_SUFFIX);
      if (renamedTo.exists()) {
        throw new IOException(String.format(
            "Can not inject failure to dir: %s because %s exists.",
            dir, renamedTo));
      }
      // 重命名文件操作
      if (!dir.renameTo(renamedTo)) {
        throw new IOException(String.format("Failed to rename %s to %s.",
            dir, renamedTo));
      }
      if (!dir.createNewFile()) {
        throw new IOException(String.format(
            "Failed to create file %s to inject disk failure.", dir));
      }
    }
  }

以上2种是结合应用程序本身的特点而构造的错误注入的方法，这个可能还不算特别通用，如果是一些非存储类的分布式程序，可能就不适用了。

钩子方法的引入

而这个“错误发生器”长什么样呢？笔者以DataNode的错误发生器为例：

/**
 * Used for injecting faults in DFSClient and DFSOutputStream tests.
 * Calls into this are a no-op in production code.
 */
@VisibleForTesting
@InterfaceAudience.Private
public class DataNodeFaultInjector {
  // 单例模式
  public static DataNodeFaultInjector instance = new DataNodeFaultInjector();

  public static DataNodeFaultInjector get() {
    return instance;
  }

  public static void set(DataNodeFaultInjector injector) {
    instance = injector;
  }
  ...

然后下面是许多钩子方法，不同的方法到时会放到真正的处理方法中。

public class DataNodeFaultInjector {

  ...
  // 下面是钩子方法，一般不执行具体逻辑
  public void getHdfsBlocksMetadata() {}

  public void writeBlockAfterFlush() throws IOException {}

  public void sendShortCircuitShmResponse() throws IOException {}

  public boolean dropHeartbeatPacket() {
    return false;
  }
  ...

接着这些钩子方法会被放到对应的方法中，以writeBlockAfterFlush方法为例,此方法在DataXceiver的writeBlock方法中被调用：

  @Override
  public void writeBlock(final ExtendedBlock block,
      final StorageType storageType,
      final Token<BlockTokenIdentifier> blockToken,
      final String clientname,
      final DatanodeInfo[] targets,
      final StorageType[] targetStorageTypes,
      final DatanodeInfo srcDataNode,
      final BlockConstructionStage stage,
      final int pipelineSize,
      final long minBytesRcvd,
      final long maxBytesRcvd,
      final long latestGenerationStamp,
      DataChecksum requestedChecksum,
      CachingStrategy cachingStrategy,
      boolean allowLazyPersist,
      final boolean pinning,
      final boolean[] targetPinnings) throws IOException {
    previousOpClientName = clientname;
    ...
          try {
                  DataNodeFaultInjector.get().failMirrorConnection();

          int timeoutValue = dnConf.socketTimeout +
              (HdfsConstants.READ_TIMEOUT_EXTENSION * targets.length);
          int writeTimeout = dnConf.socketWriteTimeout +
              (HdfsConstants.WRITE_TIMEOUT_EXTENSION * targets.length);
          NetUtils.connect(mirrorSock, mirrorTarget, timeoutValue);
          mirrorSock.setTcpNoDelay(dnConf.getDataTransferServerTcpNoDelay());
          mirrorSock.setSoTimeout(timeoutValue);
          mirrorSock.setKeepAlive(true);
          if (dnConf.getTransferSocketSendBufferSize() > 0) {
            mirrorSock.setSendBufferSize(
                dnConf.getTransferSocketSendBufferSize());
          }

          ...

          mirrorOut.flush();

          // 此处调用钩子方法
          DataNodeFaultInjector.get().writeBlockAfterFlush();

          ...
        } catch (IOException e) {
        //...
        }
      }

因为在钩子方法所属区域内的代码块操作在真实环境中是很有可能抛出IO异常的，所以加在此处来模拟异常的发生。但是前面我们也说了，这个错误注入器内的方法实际上是空方法啊，异常怎么可能会抛出呢？所以这里需要矫正一个观点：错误注入器本身不负责注入错误，而是由外界来触发的。这里我们要用到mock方法。Mock操作可以在不更改原始代码的同时，覆盖原始方法操作。所以这里我们会用mock操作来覆盖这些钩子方法。下面是mock方法的构造过程。

  public void testTimeoutMetric() throws Exception {
    final Configuration conf = new HdfsConfiguration();
    final Path path = new Path("/test");

    final MiniDFSCluster cluster =
        new MiniDFSCluster.Builder(conf).numDataNodes(2).build();

    final List<FSDataOutputStream> streams = Lists.newArrayList();
    try {
      final FSDataOutputStream out =
          cluster.getFileSystem().create(path, (short) 2);
      // 构造错误注入器
      final DataNodeFaultInjector injector = Mockito.mock
          (DataNodeFaultInjector.class);
      // 覆盖注入器的writeBlockAfterFlush方法，使之执行时抛出异常
      Mockito.doThrow(new IOException("mock IOException")).
          when(injector).
          writeBlockAfterFlush();
      // 将mock出的注入器赋给原始注入器
      DataNodeFaultInjector.instance = injector;
      streams.add(out);
      out.writeBytes("old gs data\n");
      out.hflush();

      /* Test the metric. */
      final MetricsRecordBuilder dnMetrics =
          getMetrics(cluster.getDataNodes().get(0).getMetrics().name());
      // 检测异常
      assertCounter("DatanodeNetworkErrors", 1L, dnMetrics);
      ...

通过这样一番处理，程序在运行writeBlockAfterFlush时就不是一个空逻辑了，而是会抛出一个IO异常。而此时，原主逻辑代码却什么都没动。这种方式的错误注入在Hadoop代码内部的很多地方都存在，是一种比较通用的错误注入模式。Mock方式的错误注入原理刚好对照了上文提到的错误注入原理。

模拟对象的构造

下面是此类内部会用到的输出流对象：

  static private class SimulatedOutputStream extends OutputStream {
    long length = 0;

    /**
     * constructor for Simulated Output Steram
     */
    SimulatedOutputStream() {
    }

    /**
     *
     * @return the length of the data created so far.
     */
    long getLength() {
      return length;
    }

    /**
     */
    void setLength(long length) {
      this.length = length;
    }

    // 下面写出操作不涉及真实数据的写出
    @Override
    public void write(int arg0) throws IOException {
      length++;
    }

    @Override
    public void write(byte[] b) throws IOException {
      length += b.length;
    }

    @Override
    public void write(byte[] b,
              int off,
              int len) throws IOException  {
      length += len;
    }
  }

通过这个例子，笔者想说的是，测试也是一门技术活，越是在复杂的运行环境中，高质量的测试就显得越为重要。完全生搬硬套式的测试方法可能能解决一时所需，但解决不了根本问题。测试与开放同样重要，在想好开发一个新功能的同时，也请想好如何进行测试的方法。

参考文献