mapred/gridmix/TestResourceUsageEmulators.java

/**
 * 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.mapred.gridmix;

import java.io.IOException;

import org.junit.Test;
import static org.junit.Assert.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.MapContext;
import org.apache.hadoop.mapreduce.StatusReporter;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskInputOutputContext;
import org.apache.hadoop.util.ResourceCalculatorPlugin;
import org.apache.hadoop.util.ResourceCalculatorPlugin.ProcResourceValues;
import org.apache.hadoop.tools.rumen.ResourceUsageMetrics;
import org.apache.hadoop.util.DummyResourceCalculatorPlugin;
import org.apache.hadoop.mapred.TaskTracker;
import org.apache.hadoop.mapred.gridmix.LoadJob.ResourceUsageMatcherRunner;
import org.apache.hadoop.mapred.gridmix.emulators.resourceusage.CumulativeCpuUsageEmulatorPlugin;
import org.apache.hadoop.mapred.gridmix.emulators.resourceusage.ResourceUsageEmulatorPlugin;
import org.apache.hadoop.mapred.gridmix.emulators.resourceusage.ResourceUsageMatcher;
import org.apache.hadoop.mapred.gridmix.emulators.resourceusage.CumulativeCpuUsageEmulatorPlugin.DefaultCpuUsageEmulator;

/**
 * Test Gridmix's resource emulator framework and supported plugins.
 */
public class TestResourceUsageEmulators {
  /**
   * A {@link ResourceUsageEmulatorPlugin} implementation for testing purpose.
   * It essentially creates a file named 'test' in the test directory.
   */
  static class TestResourceUsageEmulatorPlugin
  implements ResourceUsageEmulatorPlugin {
    static final Path rootTempDir =
        new Path(System.getProperty("test.build.data", "/tmp"));
    static final Path tempDir =
      new Path(rootTempDir, "TestResourceUsageEmulatorPlugin");
    static final String DEFAULT_IDENTIFIER = "test";

    private Path touchPath = null;
    private FileSystem fs = null;

    @Override
    public void emulate() throws IOException, InterruptedException {
      // add some time between 2 calls to emulate()
      try {
        Thread.sleep(1000); // sleep for 1s
      } catch (Exception e){}

      try {
        fs.delete(touchPath, false); // delete the touch file
        //TODO Search for a better touch utility
        fs.create(touchPath).close(); // recreate it
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }

    protected String getIdentifier() {
      return DEFAULT_IDENTIFIER;
    }

    private static Path getFilePath(String id) {
      return new Path(tempDir, id);
    }

    private static Path getInitFilePath(String id) {
      return new Path(tempDir, id + ".init");
    }

    @Override
    public void initialize(Configuration conf, ResourceUsageMetrics metrics,
        ResourceCalculatorPlugin monitor, Progressive progress) {
      // add some time between 2 calls to initialize()
      try {
        Thread.sleep(1000); // sleep for 1s
      } catch (Exception e){}

      try {
        fs = FileSystem.getLocal(conf);

        Path initPath = getInitFilePath(getIdentifier());
        fs.delete(initPath, false); // delete the old file
        fs.create(initPath).close(); // create a new one

        touchPath = getFilePath(getIdentifier());
        fs.delete(touchPath, false);
      } catch (Exception e) {

      } finally {
        if (fs != null) {
          try {
            fs.deleteOnExit(tempDir);
          } catch (IOException ioe){}
        }
      }
    }

    // test if the emulation framework successfully loaded this plugin
    static long testInitialization(String id, Configuration conf)
    throws IOException {
      Path testPath = getInitFilePath(id);
      FileSystem fs = FileSystem.getLocal(conf);
      return fs.exists(testPath)
             ? fs.getFileStatus(testPath).getModificationTime()
             : 0;
    }

    // test if the emulation framework successfully loaded this plugin
    static long testEmulation(String id, Configuration conf)
    throws IOException {
      Path testPath = getFilePath(id);
      FileSystem fs = FileSystem.getLocal(conf);
      return fs.exists(testPath)
             ? fs.getFileStatus(testPath).getModificationTime()
             : 0;
    }

    @Override
    public float getProgress() {
      try {
        return fs.exists(touchPath) ? 1.0f : 0f;
      } catch (IOException ioe) {}
      return 0f;
    }
  }

  /**
   * Test implementation of {@link ResourceUsageEmulatorPlugin} which creates
   * a file named 'others' in the test directory.
   */
  static class TestOthers extends TestResourceUsageEmulatorPlugin {
    static final String ID = "others";

    @Override
    protected String getIdentifier() {
      return ID;
    }
  }

  /**
   * Test implementation of {@link ResourceUsageEmulatorPlugin} which creates
   * a file named 'cpu' in the test directory.
   */
  static class TestCpu extends TestResourceUsageEmulatorPlugin {
    static final String ID = "cpu";

    @Override
    protected String getIdentifier() {
      return ID;
    }
  }

  /**
   * Test {@link ResourceUsageMatcher}.
   */
  @Test
  public void testResourceUsageMatcher() throws Exception {
    ResourceUsageMatcher matcher = new ResourceUsageMatcher();
    Configuration conf = new Configuration();
    conf.setClass(ResourceUsageMatcher.RESOURCE_USAGE_EMULATION_PLUGINS,
                  TestResourceUsageEmulatorPlugin.class,
                  ResourceUsageEmulatorPlugin.class);
    long currentTime = System.currentTimeMillis();

    matcher.configure(conf, null, null, null);

    matcher.matchResourceUsage();

    String id = TestResourceUsageEmulatorPlugin.DEFAULT_IDENTIFIER;
    long result =
      TestResourceUsageEmulatorPlugin.testInitialization(id, conf);
    assertTrue("Resource usage matcher failed to initialize the configured"
               + " plugin", result > currentTime);
    result = TestResourceUsageEmulatorPlugin.testEmulation(id, conf);
    assertTrue("Resource usage matcher failed to load and emulate the"
               + " configured plugin", result > currentTime);

    // test plugin order to first emulate cpu and then others
    conf.setStrings(ResourceUsageMatcher.RESOURCE_USAGE_EMULATION_PLUGINS,
                    TestCpu.class.getName() + "," + TestOthers.class.getName());

    matcher.configure(conf, null, null, null);

    // test the initialization order
    long time1 =
           TestResourceUsageEmulatorPlugin.testInitialization(TestCpu.ID, conf);
    long time2 =
           TestResourceUsageEmulatorPlugin.testInitialization(TestOthers.ID,
                                                              conf);
    assertTrue("Resource usage matcher failed to initialize the configured"
               + " plugins in order", time1 < time2);

    matcher.matchResourceUsage();

    // Note that the cpu usage emulator plugin is configured 1st and then the
    // others plugin.
    time1 =
      TestResourceUsageEmulatorPlugin.testInitialization(TestCpu.ID, conf);
    time2 =
      TestResourceUsageEmulatorPlugin.testInitialization(TestOthers.ID,
                                                         conf);
    assertTrue("Resource usage matcher failed to load the configured plugins",
               time1 < time2);
  }

  /**
   * Fakes the cumulative usage using {@link FakeCpuUsageEmulatorCore}.
   */
  static class FakeResourceUsageMonitor extends DummyResourceCalculatorPlugin {
    private FakeCpuUsageEmulatorCore core;

    public FakeResourceUsageMonitor(FakeCpuUsageEmulatorCore core) {
      this.core = core;
    }

    /**
     * A dummy CPU usage monitor. Every call to
     * {@link ResourceCalculatorPlugin#getCumulativeCpuTime()} will return the
     * value of {@link FakeCpuUsageEmulatorCore#getNumCalls()}.
     */
    @Override
    public long getCumulativeCpuTime() {
      return core.getCpuUsage();
    }

    /**
     * Returns a {@link ProcResourceValues} with cumulative cpu usage
     * computed using {@link #getCumulativeCpuTime()}.
     */
    @Override
    public ProcResourceValues getProcResourceValues() {
      long usageValue = getCumulativeCpuTime();
      return new ProcResourceValues(usageValue, -1, -1);
    }
  }

  /**
   * A dummy {@link Progressive} implementation that allows users to set the
   * progress for testing. The {@link Progressive#getProgress()} call will
   * return the last progress value set using
   * {@link FakeProgressive#setProgress(float)}.
   */
  static class FakeProgressive implements Progressive {
    private float progress = 0F;
    @Override
    public float getProgress() {
      return progress;
    }

    void setProgress(float progress) {
      this.progress = progress;
    }
  }

  /**
   * A dummy reporter for {@link LoadJob.ResourceUsageMatcherRunner}.
   */
  private static class DummyReporter extends StatusReporter {
    private Progressive progress;

    DummyReporter(Progressive progress) {
      this.progress = progress;
    }

    @Override
    public org.apache.hadoop.mapreduce.Counter getCounter(Enum<?> name) {
      return null;
    }

    @Override
    public org.apache.hadoop.mapreduce.Counter getCounter(String group,
                                                          String name) {
      return null;
    }

    @Override
    public void progress() {
    }

    @Override
    public float getProgress() {
      return progress.getProgress();
    }

    @Override
    public void setStatus(String status) {
    }
  }

  // Extends ResourceUsageMatcherRunner for testing.
  @SuppressWarnings("unchecked")
  private static class FakeResourceUsageMatcherRunner
  extends ResourceUsageMatcherRunner {
    FakeResourceUsageMatcherRunner(TaskInputOutputContext context,
                                   ResourceUsageMetrics metrics) {
      super(context, metrics);
    }

    // test ResourceUsageMatcherRunner
    void test() throws Exception {
      super.match();
    }
  }

  /**
   * Test {@link LoadJob.ResourceUsageMatcherRunner}.
   */
  @Test
  @SuppressWarnings("unchecked")
  public void testResourceUsageMatcherRunner() throws Exception {
    Configuration conf = new Configuration();
    FakeProgressive progress = new FakeProgressive();

    // set the resource calculator plugin
    conf.setClass(TaskTracker.TT_RESOURCE_CALCULATOR_PLUGIN,
                  DummyResourceCalculatorPlugin.class,
                  ResourceCalculatorPlugin.class);
    // set the resources
    // set the resource implementation class
    conf.setClass(ResourceUsageMatcher.RESOURCE_USAGE_EMULATION_PLUGINS,
                  TestResourceUsageEmulatorPlugin.class,
                  ResourceUsageEmulatorPlugin.class);

    long currentTime = System.currentTimeMillis();

    // initialize the matcher class
    TaskAttemptID id = new TaskAttemptID("test", 1, true, 1, 1);
    StatusReporter reporter = new DummyReporter(progress);
    TaskInputOutputContext context =
      new MapContext(conf, id, null, null, null, reporter, null);
    FakeResourceUsageMatcherRunner matcher =
      new FakeResourceUsageMatcherRunner(context, null);

    // check if the matcher initialized the plugin
    String identifier = TestResourceUsageEmulatorPlugin.DEFAULT_IDENTIFIER;
    long initTime =
      TestResourceUsageEmulatorPlugin.testInitialization(identifier, conf);
    assertTrue("ResourceUsageMatcherRunner failed to initialize the"
               + " configured plugin", initTime > currentTime);

    // check the progress
    assertEquals("Progress mismatch in ResourceUsageMatcherRunner",
                 0, progress.getProgress(), 0D);

    // call match() and check progress
    progress.setProgress(0.01f);
    currentTime = System.currentTimeMillis();
    matcher.test();
    long emulateTime =
      TestResourceUsageEmulatorPlugin.testEmulation(identifier, conf);
    assertTrue("ProgressBasedResourceUsageMatcher failed to load and emulate"
               + " the configured plugin", emulateTime > currentTime);
  }

  /**
   * Test {@link CumulativeCpuUsageEmulatorPlugin}'s core CPU usage emulation
   * engine.
   */
  @Test
  public void testCpuUsageEmulator() throws IOException {
    // test CpuUsageEmulator calibration with fake resource calculator plugin
    long target = 100000L; // 100 secs
    int unitUsage = 50;
    FakeCpuUsageEmulatorCore fakeCpuEmulator = new FakeCpuUsageEmulatorCore();
    fakeCpuEmulator.setUnitUsage(unitUsage);
    FakeResourceUsageMonitor fakeMonitor =
      new FakeResourceUsageMonitor(fakeCpuEmulator);

    // calibrate for 100ms
    fakeCpuEmulator.calibrate(fakeMonitor, target);

    // by default, CpuUsageEmulator.calibrate() will consume 100ms of CPU usage
    assertEquals("Fake calibration failed",
                 100, fakeMonitor.getCumulativeCpuTime());
    assertEquals("Fake calibration failed",
                 100, fakeCpuEmulator.getCpuUsage());
    // by default, CpuUsageEmulator.performUnitComputation() will be called
    // twice
    assertEquals("Fake calibration failed",
                 2, fakeCpuEmulator.getNumCalls());
  }

  /**
   * This is a dummy class that fakes CPU usage.
   */
  private static class FakeCpuUsageEmulatorCore
  extends DefaultCpuUsageEmulator {
    private int numCalls = 0;
    private int unitUsage = 1;
    private int cpuUsage = 0;

    @Override
    protected void performUnitComputation() {
      ++numCalls;
      cpuUsage += unitUsage;
    }

    int getNumCalls() {
      return numCalls;
    }

    int getCpuUsage() {
      return cpuUsage;
    }

    void reset() {
      numCalls = 0;
      cpuUsage = 0;
    }

    void setUnitUsage(int unitUsage) {
      this.unitUsage = unitUsage;
    }
  }

  // Creates a ResourceUsageMetrics object from the target usage
  static ResourceUsageMetrics createMetrics(long target) {
    ResourceUsageMetrics metrics = new ResourceUsageMetrics();
    metrics.setCumulativeCpuUsage(target);
    metrics.setVirtualMemoryUsage(target);
    metrics.setPhysicalMemoryUsage(target);
    metrics.setHeapUsage(target);
    return metrics;
  }

  /**
   * Test {@link CumulativeCpuUsageEmulatorPlugin}.
   */
  @Test
  public void testCumulativeCpuUsageEmulatorPlugin() throws Exception {
    Configuration conf = new Configuration();
    long targetCpuUsage = 1000L;
    int unitCpuUsage = 50;

    // fake progress indicator
    FakeProgressive fakeProgress = new FakeProgressive();

    // fake cpu usage generator
    FakeCpuUsageEmulatorCore fakeCore = new FakeCpuUsageEmulatorCore();
    fakeCore.setUnitUsage(unitCpuUsage);

    // a cumulative cpu usage emulator with fake core
    CumulativeCpuUsageEmulatorPlugin cpuPlugin =
      new CumulativeCpuUsageEmulatorPlugin(fakeCore);

    // test with invalid or missing resource usage value
    ResourceUsageMetrics invalidUsage = createMetrics(0);
    cpuPlugin.initialize(conf, invalidUsage, null, null);

    // test if disabled cpu emulation plugin's emulate() call is a no-operation
    // this will test if the emulation plugin is disabled or not
    int numCallsPre = fakeCore.getNumCalls();
    long cpuUsagePre = fakeCore.getCpuUsage();
    cpuPlugin.emulate();
    int numCallsPost = fakeCore.getNumCalls();
    long cpuUsagePost = fakeCore.getCpuUsage();

    //  test if no calls are made cpu usage emulator core
    assertEquals("Disabled cumulative CPU usage emulation plugin works!",
                 numCallsPre, numCallsPost);

    //  test if no calls are made cpu usage emulator core
    assertEquals("Disabled cumulative CPU usage emulation plugin works!",
                 cpuUsagePre, cpuUsagePost);

    // test with valid resource usage value
    ResourceUsageMetrics metrics = createMetrics(targetCpuUsage);

    // fake monitor
    ResourceCalculatorPlugin monitor = new FakeResourceUsageMonitor(fakeCore);

    // test with default emulation interval
    testEmulationAccuracy(conf, fakeCore, monitor, metrics, cpuPlugin,
                          targetCpuUsage, targetCpuUsage / unitCpuUsage);

    // test with custom value for emulation interval of 20%
    conf.setFloat(CumulativeCpuUsageEmulatorPlugin.CPU_EMULATION_PROGRESS_INTERVAL,
                  0.2F);
    testEmulationAccuracy(conf, fakeCore, monitor, metrics, cpuPlugin,
                          targetCpuUsage, targetCpuUsage / unitCpuUsage);

    // test if emulation interval boundary is respected (unit usage = 1)
    //  test the case where the current progress is less than threshold
    fakeProgress = new FakeProgressive(); // initialize
    fakeCore.reset();
    fakeCore.setUnitUsage(1);
    conf.setFloat(CumulativeCpuUsageEmulatorPlugin.CPU_EMULATION_PROGRESS_INTERVAL,
                  0.25F);
    cpuPlugin.initialize(conf, metrics, monitor, fakeProgress);
    // take a snapshot after the initialization
    long initCpuUsage = monitor.getCumulativeCpuTime();
    long initNumCalls = fakeCore.getNumCalls();
    // test with 0 progress
    testEmulationBoundary(0F, fakeCore, fakeProgress, cpuPlugin, initCpuUsage,
                          initNumCalls, "[no-op, 0 progress]");
    // test with 24% progress
    testEmulationBoundary(0.24F, fakeCore, fakeProgress, cpuPlugin,
                          initCpuUsage, initNumCalls, "[no-op, 24% progress]");
    // test with 25% progress
    //  target = 1000ms, target emulation at 25% = 250ms,
    //  weighed target = 1000 * 0.25^4 (we are using progress^4 as the weight)
    //                 ~ 4
    //  but current usage = init-usage = 100, hence expected = 100
    testEmulationBoundary(0.25F, fakeCore, fakeProgress, cpuPlugin,
                          initCpuUsage, initNumCalls, "[op, 25% progress]");

    // test with 80% progress
    //  target = 1000ms, target emulation at 80% = 800ms,
    //  weighed target = 1000 * 0.25^4 (we are using progress^4 as the weight)
    //                 ~ 410
    //  current-usage = init-usage = 100, hence expected-usage = 410
    testEmulationBoundary(0.80F, fakeCore, fakeProgress, cpuPlugin, 410, 410,
                          "[op, 80% progress]");

    // now test if the final call with 100% progress ramps up the CPU usage
    testEmulationBoundary(1F, fakeCore, fakeProgress, cpuPlugin, targetCpuUsage,
                          targetCpuUsage, "[op, 100% progress]");

    // test if emulation interval boundary is respected (unit usage = 50)
    //  test the case where the current progress is less than threshold
    fakeProgress = new FakeProgressive(); // initialize
    fakeCore.reset();
    fakeCore.setUnitUsage(unitCpuUsage);
    conf.setFloat(CumulativeCpuUsageEmulatorPlugin.CPU_EMULATION_PROGRESS_INTERVAL,
                  0.40F);
    cpuPlugin.initialize(conf, metrics, monitor, fakeProgress);
    // take a snapshot after the initialization
    initCpuUsage = monitor.getCumulativeCpuTime();
    initNumCalls = fakeCore.getNumCalls();
    // test with 0 progress
    testEmulationBoundary(0F, fakeCore, fakeProgress, cpuPlugin, initCpuUsage,
                          initNumCalls, "[no-op, 0 progress]");
    // test with 39% progress
    testEmulationBoundary(0.39F, fakeCore, fakeProgress, cpuPlugin,
                          initCpuUsage, initNumCalls, "[no-op, 39% progress]");
    // test with 40% progress
    //  target = 1000ms, target emulation at 40% = 4000ms,
    //  weighed target = 1000 * 0.40^4 (we are using progress^4 as the weight)
    //                 ~ 26
    // current-usage = init-usage = 100, hence expected-usage = 100
    testEmulationBoundary(0.40F, fakeCore, fakeProgress, cpuPlugin,
                          initCpuUsage, initNumCalls, "[op, 40% progress]");

    // test with 90% progress
    //  target = 1000ms, target emulation at 90% = 900ms,
    //  weighed target = 1000 * 0.90^4 (we are using progress^4 as the weight)
    //                 ~ 657
    //  current-usage = init-usage = 100, hence expected-usage = 657 but
    //  the fake-core increases in steps of 50, hence final target = 700
    testEmulationBoundary(0.90F, fakeCore, fakeProgress, cpuPlugin, 700,
                          700 / unitCpuUsage, "[op, 90% progress]");

    // now test if the final call with 100% progress ramps up the CPU usage
    testEmulationBoundary(1F, fakeCore, fakeProgress, cpuPlugin, targetCpuUsage,
                          targetCpuUsage / unitCpuUsage, "[op, 100% progress]");
  }

  // test whether the CPU usage emulator achieves the desired target using
  // desired calls to the underling core engine.
  private static void testEmulationAccuracy(Configuration conf,
                        FakeCpuUsageEmulatorCore fakeCore,
                        ResourceCalculatorPlugin monitor,
                        ResourceUsageMetrics metrics,
                        CumulativeCpuUsageEmulatorPlugin cpuPlugin,
                        long expectedTotalCpuUsage, long expectedTotalNumCalls)
  throws Exception {
    FakeProgressive fakeProgress = new FakeProgressive();
    fakeCore.reset();
    cpuPlugin.initialize(conf, metrics, monitor, fakeProgress);
    int numLoops = 0;
    while (fakeProgress.getProgress() < 1) {
      ++numLoops;
      float progress = (float)numLoops / 100;
      fakeProgress.setProgress(progress);
      cpuPlugin.emulate();
    }

    // test if the resource plugin shows the expected invocations
    assertEquals("Cumulative cpu usage emulator plugin failed (num calls)!",
                 expectedTotalNumCalls, fakeCore.getNumCalls(), 0L);
    // test if the resource plugin shows the expected usage
    assertEquals("Cumulative cpu usage emulator plugin failed (total usage)!",
                 expectedTotalCpuUsage, fakeCore.getCpuUsage(), 0L);
  }

  // tests if the CPU usage emulation plugin emulates only at the expected
  // progress gaps
  private static void testEmulationBoundary(float progress,
      FakeCpuUsageEmulatorCore fakeCore, FakeProgressive fakeProgress,
      CumulativeCpuUsageEmulatorPlugin cpuPlugin, long expectedTotalCpuUsage,
      long expectedTotalNumCalls, String info) throws Exception {
    fakeProgress.setProgress(progress);
    cpuPlugin.emulate();

    assertEquals("Emulation interval test for cpu usage failed " + info + "!",
                 expectedTotalCpuUsage, fakeCore.getCpuUsage(), 0L);
    assertEquals("Emulation interval test for num calls failed " + info + "!",
                 expectedTotalNumCalls, fakeCore.getNumCalls(), 0L);
  }
}