HDFS-710. Added actions with constraints to the pipeline fault injection tests and changed SleepAction to support uniform random sleeping over an interval.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/trunk@826086 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -284,6 +284,10 @@ Release 0.21.0 - Unreleased
     HDFS-705. Create an adapter to access some of package-private methods of
     DataNode from tests (cos)
 
+    HDFS-710. Added actions with constraints to the pipeline fault injection
+    tests and changed SleepAction to support uniform random sleeping over an
+    interval.  (szetszwo)
+
   BUG FIXES
 
     HDFS-76. Better error message to users when commands fail because of 

src/test/aop/org/apache/hadoop/fi/DataTransferTestUtil.java

@@ -17,16 +17,18 @@
  */
 package org.apache.hadoop.fi;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.hadoop.fi.FiTestUtil.Action;
 import org.apache.hadoop.fi.FiTestUtil.ActionContainer;
+import org.apache.hadoop.fi.FiTestUtil.ConstraintSatisfactionAction;
+import org.apache.hadoop.fi.FiTestUtil.CountdownConstraint;
 import org.apache.hadoop.hdfs.protocol.DatanodeID;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
 
-import java.util.ArrayList;
-import java.util.List;
-
 /**
  * Utilities for DataTransferProtocol related tests,
  * e.g. TestFiDataTransferProtocol.
@@ -127,8 +129,7 @@ public class DataTransferTestUtil {
       return currentTest + ", index=" + index;
     }
 
-    /** {@inheritDoc}
-     * @param datanodeID*/
+    /** return a String with this object and the datanodeID. */
     String toString(DatanodeID datanodeID) {
       return "FI: " + this + ", datanode="
           + datanodeID.getName();
@@ -178,16 +179,42 @@ public class DataTransferTestUtil {
    * or sleep forever so that datanode becomes not responding.
    */
   public static class SleepAction extends DataNodeAction {
-    /** In milliseconds, duration <= 0 means sleeping forever.*/
-    final long duration;
+    /** In milliseconds;
+     * must have (0 <= minDuration < maxDuration) or (maxDuration <= 0).
+     */
+    final long minDuration;
+    /** In milliseconds; maxDuration <= 0 means sleeping forever.*/
+    final long maxDuration;
 
     /**
      * Create an action for datanode i in the pipeline.
      * @param duration In milliseconds, duration <= 0 means sleeping forever.
      */
     public SleepAction(String currentTest, int i, long duration) {
+      this(currentTest, i, duration, duration <= 0? duration: duration+1);
+    }
+
+    /**
+     * Create an action for datanode i in the pipeline.
+     * @param minDuration minimum sleep time
+     * @param maxDuration maximum sleep time
+     */
+    public SleepAction(String currentTest, int i,
+        long minDuration, long maxDuration) {
       super(currentTest, i);
-      this.duration = duration;
+
+      if (maxDuration > 0) {
+        if (minDuration < 0) {
+          throw new IllegalArgumentException("minDuration = " + minDuration
+              + " < 0 but maxDuration = " + maxDuration + " > 0");
+        }
+        if (minDuration >= maxDuration) {
+          throw new IllegalArgumentException(
+              minDuration + " = minDuration >= maxDuration = " + maxDuration);
+        }
+      }
+      this.minDuration = minDuration;
+      this.maxDuration = maxDuration;
     }
 
     @Override
@@ -195,15 +222,21 @@ public class DataTransferTestUtil {
       final DataTransferTest test = getDataTransferTest();
       final Pipeline p = test.getPipeline(id);
       if (!test.isSuccess() && p.contains(index, id)) {
-        final String s = toString(id) + ", duration=" + duration;
-        FiTestUtil.LOG.info(s);
-        if (duration <= 0) {
+        FiTestUtil.LOG.info(toString(id));
+        if (maxDuration <= 0) {
           for(; true; FiTestUtil.sleep(1000)); //sleep forever
         } else {
-          FiTestUtil.sleep(duration);
+          FiTestUtil.sleep(minDuration, maxDuration);
         }
       }
     }
+
+    /** {@inheritDoc} */
+    @Override
+    public String toString() {
+      return super.toString()
+          + ", duration = [" + minDuration + "," + maxDuration + ")";
+    }
   }
 
   /** Action for pipeline error verification */
@@ -237,4 +270,48 @@ public class DataTransferTestUtil {
       }
     }
   }
+
+  /**
+   *  Create a OomAction with a CountdownConstraint
+   *  so that it throws OutOfMemoryError if the count is zero.
+   */
+  public static ConstraintSatisfactionAction<DatanodeID> createCountdownOomAction(
+      String currentTest, int i, int count) {
+    return new ConstraintSatisfactionAction<DatanodeID>(
+        new OomAction(currentTest, i), new CountdownConstraint(count));
+  }
+
+  /**
+   *  Create a DoosAction with a CountdownConstraint
+   *  so that it throws DiskOutOfSpaceException if the count is zero.
+   */
+  public static ConstraintSatisfactionAction<DatanodeID> createCountdownDoosAction(
+      String currentTest, int i, int count) {
+    return new ConstraintSatisfactionAction<DatanodeID>(
+        new DoosAction(currentTest, i), new CountdownConstraint(count));
+  }
+
+  /**
+   * Create a SleepAction with a CountdownConstraint
+   * for datanode i in the pipeline.
+   * When the count is zero,
+   * sleep some period of time so that it slows down the datanode
+   * or sleep forever so the that datanode becomes not responding.
+   */
+  public static ConstraintSatisfactionAction<DatanodeID> createCountdownSleepAction(
+      String currentTest, int i, long minDuration, long maxDuration, int count) {
+    return new ConstraintSatisfactionAction<DatanodeID>(
+        new SleepAction(currentTest, i, minDuration, maxDuration),
+        new CountdownConstraint(count));
+  }
+
+  /**
+   * Same as
+   * createCountdownSleepAction(currentTest, i, duration, duration+1, count).
+   */
+  public static ConstraintSatisfactionAction<DatanodeID> createCountdownSleepAction(
+      String currentTest, int i, long duration, int count) {
+    return createCountdownSleepAction(currentTest, i, duration, duration+1,
+        count);
+  }
 }

src/test/aop/org/apache/hadoop/fi/FiTestUtil.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.fi;
 
 import java.io.IOException;
+import java.util.Random;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -27,9 +28,34 @@ public class FiTestUtil {
   /** Logging */
   public static final Log LOG = LogFactory.getLog(FiTestUtil.class);
 
+  /** Random source */
+  public static final ThreadLocal<Random> RANDOM = new ThreadLocal<Random>() {
+    protected Random initialValue() {
+      final Random r = new Random();
+      final long seed = r.nextLong();
+      LOG.info(Thread.currentThread() + ": seed=" + seed);
+      r.setSeed(seed);
+      return r;
+    }
+  };
+
+  /**
+   * Return a random integer uniformly distributed in the interval [min,max).
+   * Assume max - min <= Integer.MAX_VALUE.
+   */
+  public static long nextRandomLong(final long min, final long max) {
+    final long d = max - min;
+    if (d <= 0 || d > Integer.MAX_VALUE) {
+      throw new IllegalArgumentException(
+          "d <= 0 || d > Integer.MAX_VALUE, min=" + min + ", max=" + max);
+    }
+    return d == 1? min: min + RANDOM.get().nextInt((int)d);
+  }
+
   /** Return the method name of the callee. */
   public static String getMethodName() {
-    return Thread.currentThread().getStackTrace()[2].getMethodName();
+    final StackTraceElement[] s = Thread.currentThread().getStackTrace();
+    return s[s.length > 2? 2: s.length - 1].getMethodName();
   }
 
   /**
@@ -44,6 +70,18 @@ public class FiTestUtil {
     }
   }
 
+  /**
+   * Sleep a random number of milliseconds over the interval [min, max).
+   * If there is an InterruptedException, re-throw it as a RuntimeException.
+   */
+  public static void sleep(final long min, final long max) {
+    final long n = nextRandomLong(min, max);
+    LOG.info(Thread.currentThread().getName() + " sleeps for " + n  +"ms");
+    if (n > 0) {
+      sleep(n);
+    }
+  }
+
   /** Action interface */
   public static interface Action<T> {
     /** Run the action with the parameter. */
@@ -67,4 +105,61 @@ public class FiTestUtil {
       }
     }
   }
+
+  /** Constraint interface */
+  public static interface Constraint {
+    /** Is this constraint satisfied? */
+    public boolean isSatisfied();
+  }
+
+  /** Counting down, the constraint is satisfied if the count is zero. */
+  public static class CountdownConstraint implements Constraint {
+    private int count;
+
+    /** Initialize the count. */
+    public CountdownConstraint(int count) {
+      if (count < 0) {
+        throw new IllegalArgumentException(count + " = count < 0");
+      }
+      this.count = count;
+    }
+
+    /** Counting down, the constraint is satisfied if the count is zero. */
+    public boolean isSatisfied() {
+      if (count > 0) {
+        count--;
+        return false;
+      }
+      return true;
+    }
+  }
+  
+  /** An action is fired if all the constraints are satisfied. */
+  public static class ConstraintSatisfactionAction<T> implements Action<T> {
+    private final Action<T> action;
+    private final Constraint[] constraints;
+    
+    /** Constructor */
+    public ConstraintSatisfactionAction(
+        Action<T> action, Constraint... constraints) {
+      this.action = action;
+      this.constraints = constraints;
+    }
+
+    /**
+     * Fire the action if all the constraints are satisfied.
+     * Short-circuit-and is used. 
+     */
+    @Override
+    public final void run(T parameter) throws IOException {
+      for(Constraint c : constraints) {
+        if (!c.isSatisfied()) {
+          return;
+        }
+      }
+
+      //all constraints are satisfied, fire the action
+      action.run(parameter);
+    }
+  }
 }

src/test/aop/org/apache/hadoop/fi/PipelineTest.java

@@ -19,7 +19,6 @@ package org.apache.hadoop.fi;
 
 import org.apache.hadoop.hdfs.protocol.DatanodeID;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
-import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 
 /** A pipeline contains a list of datanodes. */
 public interface PipelineTest {

src/test/aop/org/apache/hadoop/hdfs/TestFiHFlush.java

@@ -18,17 +18,14 @@
 
 package org.apache.hadoop.hdfs;
 
-import org.junit.Test;
+import java.io.IOException;
+
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.HdfsConfiguration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.fi.FiHFlushTestUtil;
 import org.apache.hadoop.fi.FiTestUtil;
 import org.apache.hadoop.fi.FiHFlushTestUtil.DerrAction;
 import org.apache.hadoop.fi.FiHFlushTestUtil.HFlushTest;
-import org.eclipse.jdt.core.dom.ThisExpression;
-
-import java.io.IOException;
+import org.junit.Test;
 
 /** Class provides basic fault injection tests according to the test plan
     of HDFS-265