/**
 * 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.tools.dynamometer;

import com.google.common.collect.Sets;
import java.util.Optional;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.apache.hadoop.test.PlatformAssumptions;
import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditLogDirectParser;
import org.apache.hadoop.tools.dynamometer.workloadgenerator.audit.AuditReplayMapper;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.util.JarFinder;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.MiniYARNCluster;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.hadoop.tools.dynamometer.DynoInfraUtils.fetchHadoopTarball;
import static org.apache.hadoop.hdfs.MiniDFSCluster.PROP_TEST_BUILD_DATA;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * Start a Dynamometer cluster in a MiniYARNCluster. Ensure that the NameNode is
 * able to start correctly, exit safemode, and run some commands. Subsequently
 * the workload job is launched and it is verified that it completes
 * successfully and is able to replay commands as expected.
 *
 * To run this test JAVA_HOME must be set correctly, and the {@code tar} utility
 * must be available.
 *
 * You can optionally specify which version of HDFS should be started within the
 * Dynamometer cluster; the default is {@value HADOOP_BIN_VERSION_DEFAULT}. This
 * can be adjusted by setting the {@value HADOOP_BIN_VERSION_KEY} property. This
 * will automatically download the correct Hadoop tarball for the specified
 * version. It downloads from an Apache mirror (by default
 * {@value DynoInfraUtils#APACHE_DOWNLOAD_MIRROR_DEFAULT}); which mirror is used
 * can be controlled with the {@value DynoInfraUtils#APACHE_DOWNLOAD_MIRROR_KEY}
 * property. Note that mirrors normally contain only the latest releases on any
 * given release line; you may need to use
 * {@code http://archive.apache.org/dist/} for older releases. The downloaded
 * tarball will be stored in the test directory and can be reused between test
 * executions. Alternatively, you can specify the {@value HADOOP_BIN_PATH_KEY}
 * property to point directly to a Hadoop tarball which is present locally and
 * no download will occur.
 */
public class TestDynamometerInfra {

  private static final Logger LOG =
      LoggerFactory.getLogger(TestDynamometerInfra.class);

  private static final int MINICLUSTER_NUM_NMS = 3;
  private static final int MINICLUSTER_NUM_DNS = 1;

  private static final String HADOOP_BIN_PATH_KEY = "dyno.hadoop.bin.path";
  private static final String HADOOP_BIN_VERSION_KEY =
      "dyno.hadoop.bin.version";
  private static final String HADOOP_BIN_VERSION_DEFAULT = "3.1.1";
  private static final String FSIMAGE_FILENAME = "fsimage_0000000000000061740";
  private static final String VERSION_FILENAME = "VERSION";

  private static final String HADOOP_BIN_UNPACKED_DIR_PREFIX =
      "hadoop_unpacked_";

  private static final String NAMENODE_NODELABEL = "dyno_namenode";
  private static final String DATANODE_NODELABEL = "dyno_datanode";

  private static MiniDFSCluster miniDFSCluster;
  private static MiniYARNCluster miniYARNCluster;
  private static YarnClient yarnClient;
  private static FileSystem fs;
  private static Configuration conf;
  private static Configuration yarnConf;
  private static Path fsImageTmpPath;
  private static Path fsVersionTmpPath;
  private static Path blockImageOutputDir;
  private static Path auditTraceDir;
  private static Path confZip;
  private static File testBaseDir;
  private static File hadoopTarballPath;
  private static File hadoopUnpackedDir;

  private ApplicationId infraAppId;

  @BeforeClass
  public static void setupClass() throws Exception {
    PlatformAssumptions.assumeNotWindows("Dynamometer will not run on Windows");
    Assume.assumeThat("JAVA_HOME must be set properly",
        System.getenv("JAVA_HOME"), notNullValue());
    try {
      Shell.ShellCommandExecutor tarCheck = new Shell.ShellCommandExecutor(
          new String[]{"bash", "-c", "command -v tar"});
      tarCheck.execute();
      Assume.assumeTrue("tar command is not available",
          tarCheck.getExitCode() == 0);
    } catch (IOException ioe) {
      Assume.assumeNoException("Unable to execute a shell command", ioe);
    }

    conf = new Configuration();
    // Follow the conventions of MiniDFSCluster
    testBaseDir = new File(
        System.getProperty(PROP_TEST_BUILD_DATA, "build/test/data"));
    String hadoopBinVersion = System.getProperty(HADOOP_BIN_VERSION_KEY,
        HADOOP_BIN_VERSION_DEFAULT);
    if (System.getProperty(HADOOP_BIN_PATH_KEY) == null) {
      hadoopTarballPath = fetchHadoopTarball(testBaseDir, hadoopBinVersion,
          conf, LOG);
    } else {
      hadoopTarballPath = new File(System.getProperty(HADOOP_BIN_PATH_KEY));
    }
    if (testBaseDir.exists()) {
      // Delete any old unpacked bin dirs that weren't previously cleaned up
      File[] oldUnpackedDirs = testBaseDir.listFiles(
          (dir, name) -> name.startsWith(HADOOP_BIN_UNPACKED_DIR_PREFIX));
      if (oldUnpackedDirs != null) {
        for (File oldDir : oldUnpackedDirs) {
          FileUtils.deleteQuietly(oldDir);
        }
      }
    }
    // Set up the Hadoop binary to be used as the system-level Hadoop install
    hadoopUnpackedDir = new File(testBaseDir,
        HADOOP_BIN_UNPACKED_DIR_PREFIX + UUID.randomUUID());
    assertTrue("Failed to make temporary directory",
        hadoopUnpackedDir.mkdirs());
    Shell.ShellCommandExecutor shexec = new Shell.ShellCommandExecutor(
        new String[] {"tar", "xzf", hadoopTarballPath.getAbsolutePath(), "-C",
            hadoopUnpackedDir.getAbsolutePath()});
    shexec.execute();
    if (shexec.getExitCode() != 0) {
      fail("Unable to execute tar to expand Hadoop binary");
    }

    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 128);
    conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true);
    for (String q : new String[] {"root", "root.default"}) {
      conf.setInt(CapacitySchedulerConfiguration.PREFIX + q + "."
          + CapacitySchedulerConfiguration.CAPACITY, 100);
      String accessibleNodeLabelPrefix = CapacitySchedulerConfiguration.PREFIX
          + q + "." + CapacitySchedulerConfiguration.ACCESSIBLE_NODE_LABELS;
      conf.set(accessibleNodeLabelPrefix,
          CapacitySchedulerConfiguration.ALL_ACL);
      conf.setInt(accessibleNodeLabelPrefix + "." + DATANODE_NODELABEL + "."
          + CapacitySchedulerConfiguration.CAPACITY, 100);
      conf.setInt(accessibleNodeLabelPrefix + "." + NAMENODE_NODELABEL + "."
          + CapacitySchedulerConfiguration.CAPACITY, 100);
    }
    // This is necessary to have the RM respect our vcore allocation request
    conf.setClass(CapacitySchedulerConfiguration.RESOURCE_CALCULATOR_CLASS,
        DominantResourceCalculator.class, ResourceCalculator.class);
    conf.setBoolean(YarnConfiguration.NM_DISK_HEALTH_CHECK_ENABLE, false);
    miniYARNCluster = new MiniYARNCluster(TestDynamometerInfra.class.getName(),
        1, MINICLUSTER_NUM_NMS, 1, 1);
    miniYARNCluster.init(conf);
    miniYARNCluster.start();

    yarnConf = miniYARNCluster.getConfig();
    miniDFSCluster = new MiniDFSCluster.Builder(conf).format(true)
        .numDataNodes(MINICLUSTER_NUM_DNS).build();
    miniDFSCluster.waitClusterUp();
    FileSystem.setDefaultUri(conf, miniDFSCluster.getURI());
    FileSystem.setDefaultUri(yarnConf, miniDFSCluster.getURI());
    fs = miniDFSCluster.getFileSystem();

    URL url = Thread.currentThread().getContextClassLoader()
        .getResource("yarn-site.xml");
    if (url == null) {
      throw new RuntimeException(
          "Could not find 'yarn-site.xml' dummy file in classpath");
    }
    yarnConf.set(YarnConfiguration.YARN_APPLICATION_CLASSPATH,
        new File(url.getPath()).getParent());
    // Write the XML to a buffer before writing to the file. writeXml() can
    // trigger a read of the existing yarn-site.xml, so writing directly could
    // trigger a read of the file while it is in an inconsistent state
    // (partially written)
    try (ByteArrayOutputStream bytesOut = new ByteArrayOutputStream()) {
      yarnConf.writeXml(bytesOut);
      try (OutputStream fileOut = new FileOutputStream(
          new File(url.getPath()))) {
        fileOut.write(bytesOut.toByteArray());
      }
    }

    yarnClient = YarnClient.createYarnClient();
    yarnClient.init(new Configuration(yarnConf));
    yarnClient.start();

    fsImageTmpPath = fs.makeQualified(new Path("/tmp/" + FSIMAGE_FILENAME));
    fsVersionTmpPath = fs.makeQualified(new Path("/tmp/" + VERSION_FILENAME));
    blockImageOutputDir = fs.makeQualified(new Path("/tmp/blocks"));
    auditTraceDir = fs.makeQualified(new Path("/tmp/audit_trace_direct"));
    confZip = fs.makeQualified(new Path("/tmp/conf.zip"));

    uploadFsimageResourcesToHDFS(hadoopBinVersion);

    miniYARNCluster.waitForNodeManagersToConnect(30000);

    RMNodeLabelsManager nodeLabelManager = miniYARNCluster.getResourceManager()
        .getRMContext().getNodeLabelManager();
    nodeLabelManager.addToCluserNodeLabelsWithDefaultExclusivity(
        Sets.newHashSet(NAMENODE_NODELABEL, DATANODE_NODELABEL));
    Map<NodeId, Set<String>> nodeLabels = new HashMap<>();
    nodeLabels.put(miniYARNCluster.getNodeManager(0).getNMContext().getNodeId(),
        Sets.newHashSet(NAMENODE_NODELABEL));
    nodeLabels.put(miniYARNCluster.getNodeManager(1).getNMContext().getNodeId(),
        Sets.newHashSet(DATANODE_NODELABEL));
    nodeLabelManager.addLabelsToNode(nodeLabels);
  }

  @AfterClass
  public static void teardownClass() throws Exception {
    if (miniDFSCluster != null) {
      miniDFSCluster.shutdown(true);
      miniDFSCluster = null;
    }
    if (yarnClient != null) {
      yarnClient.stop();
      yarnClient = null;
    }
    if (miniYARNCluster != null) {
      miniYARNCluster.getResourceManager().stop();
      miniYARNCluster.getResourceManager().waitForServiceToStop(30000);
      miniYARNCluster.stop();
      miniYARNCluster.waitForServiceToStop(30000);
      FileUtils.deleteDirectory(miniYARNCluster.getTestWorkDir());
      miniYARNCluster = null;
    }
    if (hadoopUnpackedDir != null) {
      FileUtils.deleteDirectory(hadoopUnpackedDir);
    }
  }

  @After
  public void tearDown() throws Exception {
    if (infraAppId != null && yarnClient != null) {
      yarnClient.killApplication(infraAppId);
    }
    infraAppId = null;
  }

  @Test(timeout = 15 * 60 * 1000)
  public void testNameNodeInYARN() throws Exception {
    Configuration localConf = new Configuration(yarnConf);
    localConf.setLong(AuditLogDirectParser.AUDIT_START_TIMESTAMP_KEY, 60000);

    final Client client = createAndStartClient(localConf);

    awaitApplicationStartup();

    long startTime = System.currentTimeMillis();
    long maxWaitTimeMs = TimeUnit.MINUTES.toMillis(10);
    Supplier<Boolean> exitCheckSupplier = () -> {
      if (System.currentTimeMillis() - startTime > maxWaitTimeMs) {
        // Wait at most 10 minutes for the NameNode to start and be ready
        return true;
      }
      try {
        // Exit immediately if the YARN app fails
        return yarnClient.getApplicationReport(infraAppId)
            .getYarnApplicationState() == YarnApplicationState.FAILED;
      } catch (IOException | YarnException e) {
        return true;
      }
    };
    Optional<Properties> namenodeProperties = DynoInfraUtils
        .waitForAndGetNameNodeProperties(exitCheckSupplier, localConf,
            client.getNameNodeInfoPath(), LOG);
    if (!namenodeProperties.isPresent()) {
      fail("Unable to fetch NameNode properties");
    }

    DynoInfraUtils.waitForNameNodeReadiness(namenodeProperties.get(), 3, false,
        exitCheckSupplier, localConf, LOG);

    assertClusterIsFunctional(localConf, namenodeProperties.get());

    Map<ContainerId, Container> namenodeContainers = miniYARNCluster
        .getNodeManager(0).getNMContext().getContainers();
    Map<ContainerId, Container> datanodeContainers = miniYARNCluster
        .getNodeManager(1).getNMContext().getContainers();
    Map<ContainerId, Container> amContainers = miniYARNCluster.getNodeManager(2)
        .getNMContext().getContainers();
    assertEquals(1, namenodeContainers.size());
    assertEquals(2,
        namenodeContainers.keySet().iterator().next().getContainerId());
    assertEquals(2, datanodeContainers.size());
    assertEquals(1, amContainers.size());
    assertEquals(1, amContainers.keySet().iterator().next().getContainerId());

    LOG.info("Waiting for workload job to start and complete");
    GenericTestUtils.waitFor(() -> {
      try {
        return client.getWorkloadJob() != null
            && client.getWorkloadJob().isComplete();
      } catch (IOException | IllegalStateException e) {
        return false;
      }
    }, 3000, 60000);
    LOG.info("Workload job completed");

    if (!client.getWorkloadJob().isSuccessful()) {
      fail("Workload job failed");
    }
    Counters counters = client.getWorkloadJob().getCounters();
    assertEquals(6,
        counters.findCounter(AuditReplayMapper.REPLAYCOUNTERS.TOTALCOMMANDS)
            .getValue());
    assertEquals(1,
        counters
            .findCounter(AuditReplayMapper.REPLAYCOUNTERS.TOTALINVALIDCOMMANDS)
            .getValue());

    LOG.info("Waiting for infra application to exit");
    GenericTestUtils.waitFor(() -> {
      try {
        ApplicationReport report = yarnClient
            .getApplicationReport(infraAppId);
        return report
              .getYarnApplicationState() == YarnApplicationState.KILLED;
      } catch (IOException | YarnException e) {
        return false;
      }
    }, 3000, 300000);

    LOG.info("Waiting for metrics file to be ready");
    // Try to read the metrics file
    Path hdfsStoragePath = new Path(fs.getHomeDirectory(),
        DynoConstants.DYNAMOMETER_STORAGE_DIR + "/" + infraAppId);
    final Path metricsPath = new Path(hdfsStoragePath, "namenode_metrics");
    GenericTestUtils.waitFor(() -> {
      try {
        FSDataInputStream in = fs.open(metricsPath);
        String metricsOutput = in.readUTF();
        in.close();
        // Just assert that there is some metrics content in there
        assertTrue(metricsOutput.contains("JvmMetrics"));
        return true;
      } catch (IOException ioe) {
        return false;
      }
    }, 3000, 60000);
  }

  private void assertClusterIsFunctional(Configuration localConf,
      Properties namenodeProperties) throws IOException {
    // Test that we can successfully write to / read from the cluster
    try {
      URI nameNodeUri = DynoInfraUtils.getNameNodeHdfsUri(namenodeProperties);
      DistributedFileSystem dynoFS =
          (DistributedFileSystem) FileSystem.get(nameNodeUri, localConf);
      Path testFile = new Path("/tmp/test/foo");
      dynoFS.mkdir(testFile.getParent(), FsPermission.getDefault());
      FSDataOutputStream out = dynoFS.create(testFile, (short) 1);
      out.write(42);
      out.hsync();
      out.close();
      FileStatus[] stats = dynoFS.listStatus(testFile.getParent());
      assertEquals(1, stats.length);
      assertEquals("foo", stats[0].getPath().getName());
    } catch (IOException e) {
      LOG.error("Failed to write or read", e);
      throw e;
    }
  }

  private void awaitApplicationStartup()
      throws TimeoutException, InterruptedException {
    LOG.info("Waiting for application ID to become available");
    GenericTestUtils.waitFor(() -> {
      try {
        List<ApplicationReport> apps = yarnClient.getApplications();
        if (apps.size() == 1) {
          infraAppId = apps.get(0).getApplicationId();
          return true;
        } else if (apps.size() > 1) {
          fail("Unexpected: more than one application");
        }
      } catch (IOException | YarnException e) {
        fail("Unexpected exception: " + e);
      }
      return false;
    }, 1000, 60000);
  }

  private Client createAndStartClient(Configuration localConf) {
    final Client client = new Client(JarFinder.getJar(ApplicationMaster.class),
        JarFinder.getJar(Assert.class));
    client.setConf(localConf);
    Thread appThread = new Thread(() -> {
      try {
        client.run(new String[] {"-" + Client.MASTER_MEMORY_MB_ARG, "128",
            "-" + Client.CONF_PATH_ARG, confZip.toString(),
            "-" + Client.BLOCK_LIST_PATH_ARG,
            blockImageOutputDir.toString(), "-" + Client.FS_IMAGE_DIR_ARG,
            fsImageTmpPath.getParent().toString(),
            "-" + Client.HADOOP_BINARY_PATH_ARG,
            hadoopTarballPath.getAbsolutePath(),
            "-" + AMOptions.DATANODES_PER_CLUSTER_ARG, "2",
            "-" + AMOptions.DATANODE_MEMORY_MB_ARG, "128",
            "-" + AMOptions.DATANODE_NODELABEL_ARG, DATANODE_NODELABEL,
            "-" + AMOptions.NAMENODE_MEMORY_MB_ARG, "256",
            "-" + AMOptions.NAMENODE_METRICS_PERIOD_ARG, "1",
            "-" + AMOptions.NAMENODE_NODELABEL_ARG, NAMENODE_NODELABEL,
            "-" + AMOptions.SHELL_ENV_ARG,
            "HADOOP_HOME=" + getHadoopHomeLocation(),
            "-" + AMOptions.SHELL_ENV_ARG,
            "HADOOP_CONF_DIR=" + getHadoopHomeLocation() + "/etc/hadoop",
            "-" + Client.WORKLOAD_REPLAY_ENABLE_ARG,
            "-" + Client.WORKLOAD_INPUT_PATH_ARG,
            fs.makeQualified(new Path("/tmp/audit_trace_direct")).toString(),
            "-" + Client.WORKLOAD_THREADS_PER_MAPPER_ARG, "1",
            "-" + Client.WORKLOAD_START_DELAY_ARG, "10s",
            "-" + AMOptions.NAMENODE_ARGS_ARG,
            "-Ddfs.namenode.safemode.extension=0"});
      } catch (Exception e) {
        LOG.error("Error running client", e);
      }
    });

    appThread.start();
    return client;
  }

  private static URI getResourcePath(String resourceName) {
    try {
      return TestDynamometerInfra.class.getClassLoader()
          .getResource(resourceName).toURI();
    } catch (URISyntaxException e) {
      return null;
    }
  }

  /**
   * Get the Hadoop home location (i.e. for {@code HADOOP_HOME}) as the only
   * directory within the unpacked location of the Hadoop tarball.
   *
   * @return The absolute path to the Hadoop home directory.
   */
  private String getHadoopHomeLocation() {
    File[] files = hadoopUnpackedDir.listFiles();
    if (files == null || files.length != 1) {
      fail("Should be 1 directory within the Hadoop unpacked dir");
    }
    return files[0].getAbsolutePath();
  }

  /**
   * Look for the resource files relevant to {@code hadoopBinVersion} and upload
   * them onto the MiniDFSCluster's HDFS for use by the subsequent jobs.
   *
   * @param hadoopBinVersion
   *          The version string (e.g. "3.1.1") for which to look for resources.
   */
  private static void uploadFsimageResourcesToHDFS(String hadoopBinVersion)
      throws IOException {
    // Keep only the major/minor version for the resources path
    String[] versionComponents = hadoopBinVersion.split("\\.");
    if (versionComponents.length < 2) {
      fail(
          "At least major and minor version are required to be specified; got: "
              + hadoopBinVersion);
    }
    String hadoopResourcesPath = "hadoop_" + versionComponents[0] + "_"
        + versionComponents[1];
    String fsImageResourcePath = hadoopResourcesPath + "/" + FSIMAGE_FILENAME;
    fs.copyFromLocalFile(new Path(getResourcePath(fsImageResourcePath)),
        fsImageTmpPath);
    fs.copyFromLocalFile(
        new Path(getResourcePath(fsImageResourcePath + ".md5")),
        fsImageTmpPath.suffix(".md5"));
    fs.copyFromLocalFile(
        new Path(getResourcePath(hadoopResourcesPath + "/" + VERSION_FILENAME)),
        fsVersionTmpPath);
    fs.mkdirs(auditTraceDir);
    IOUtils.copyBytes(
        TestDynamometerInfra.class.getClassLoader()
            .getResourceAsStream("audit_trace_direct/audit0"),
        fs.create(new Path(auditTraceDir, "audit0")), conf, true);
    fs.mkdirs(blockImageOutputDir);
    for (String blockFile : new String[] {"dn0-a-0-r-00000", "dn1-a-0-r-00001",
        "dn2-a-0-r-00002"}) {
      IOUtils.copyBytes(
          TestDynamometerInfra.class.getClassLoader()
              .getResourceAsStream("blocks/" + blockFile),
          fs.create(new Path(blockImageOutputDir, blockFile)), conf, true);
    }
    File tempConfZip = new File(testBaseDir, "conf.zip");
    ZipOutputStream zos = new ZipOutputStream(
        new FileOutputStream(tempConfZip));
    for (String file : new String[] {"core-site.xml", "hdfs-site.xml",
        "log4j.properties"}) {
      zos.putNextEntry(new ZipEntry("etc/hadoop/" + file));
      InputStream is = TestDynamometerInfra.class.getClassLoader()
          .getResourceAsStream("conf/etc/hadoop/" + file);
      IOUtils.copyBytes(is, zos, conf, false);
      is.close();
      zos.closeEntry();
    }
    zos.close();
    fs.copyFromLocalFile(new Path(tempConfZip.toURI()), confZip);
    tempConfZip.delete();
  }

}
