HADOOP-16830. HADOOP-17271. Checkstyle, findbugs and mock failures.

Test failure was just a mock brittleness issue, just took
a long time with the debugger to work out what.
An NPE was being raised in S3AFileSystem.getDurationFactory()
as the instrumentation was null; this was happening in commit abort
and so swallowed.
Fix:
* getDurationFactory() to return null if instrumentation is null;
  trackDuration code is all robust to this
* Mock overrides the new method

(This is why I hate mocks BTW. They can simulate failures,
 but are so, so brittle).

Also:

* Tests on the trackDuration calls now make sure they are all
   invoked.
* Some copy and paste of the trackDuration logic which somewhat
  reduces the depth of lambda-expressions.
* Serialization of IOStatisticsSnapshot: method to list required
  classes and a test to verify this holds. To aid secure use.
* IOStatisticsLogging.logToStortedString is now ioStatisticsToPrettyString
  and filters out all empty/unset values (e.g counters == 0,
  min/max unset, means with samples == 0. Makes for leaner logs

Findbugs. Once S3AInstrumentation declares itself an IOStatisticsSource,
all nested classes need to do this.getIOStatistics to stop findbugs
warning.

Change-Id: I7d1da267c442b1e22dd025be97457cd4a6ae5306

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/IOStatisticsLogging.java

@@ -21,6 +21,7 @@
 import javax.annotation.Nullable;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.function.Predicate;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -86,22 +87,28 @@ public static String ioStatisticsToString(
   }
 
   /**
-   * Convert IOStatistics to a string form, with all the metrics sorted.
+   * Convert IOStatistics to a string form, with all the metrics sorted
+   * and empty value stripped.
    * This is more expensive than the simple conversion, so should only
    * be used for logging/output where it's known/highly likely that the
    * caller wants to see the values. Not for debug logging.
    * @param statistics A statistics instance.
    * @return string value or the empty string if null
    */
-  public static String ioStatisticsToSortedString(
+  public static String ioStatisticsToPrettyString(
       @Nullable final IOStatistics statistics) {
     if (statistics != null) {
       StringBuilder sb = new StringBuilder();
-      mapToSortedString(sb, "counters", statistics.counters());
-      mapToSortedString(sb, "\ngauges", statistics.gauges());
-      mapToSortedString(sb, "\nminimums", statistics.minimums());
-      mapToSortedString(sb, "\nmaximums", statistics.maximums());
-      mapToSortedString(sb, "\nmeans", statistics.meanStatistics());
+      mapToSortedString(sb, "counters", statistics.counters(),
+          p -> p == 0);
+      mapToSortedString(sb, "\ngauges", statistics.gauges(),
+          p -> p == 0);
+      mapToSortedString(sb, "\nminimums", statistics.minimums(),
+          p -> p  < 0);
+      mapToSortedString(sb, "\nmaximums", statistics.maximums(),
+          p -> p < 0);
+      mapToSortedString(sb, "\nmeans", statistics.meanStatistics(),
+          MeanStatistic::isEmpty);
 
       return sb.toString();
     } else {
@@ -147,8 +154,9 @@ private static <E> void mapToString(StringBuilder sb,
    */
   private static <E> void mapToSortedString(StringBuilder sb,
       final String type,
-      final Map<String, E> map) {
-    mapToString(sb, type, sortedMap(map), "\n");
+      final Map<String, E> map,
+      final Predicate<E> isEmpty) {
+    mapToString(sb, type, sortedMap(map, isEmpty), "\n");
   }
 
   /**
@@ -160,9 +168,14 @@ private static <E> void mapToSortedString(StringBuilder sb,
    * @return a treemap with all the entries.
    */
   private static <E> Map<String, E> sortedMap(
-      final Map<String, E> source) {
+      final Map<String, E> source,
+      final Predicate<E> isEmpty) {
     Map<String, E> tm = new TreeMap<>();
-    tm.putAll(source);
+    for (Map.Entry<String, E> entry : source.entrySet()) {
+      if (!isEmpty.test(entry.getValue())) {
+        tm.put(entry.getKey(), entry.getValue());
+      }
+    }
     return tm;
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/IOStatisticsSnapshot.java

@@ -23,9 +23,12 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -41,13 +44,19 @@
 
 /**
  * Snapshot of statistics from a different source.
- * <p></p>
+ * <p>
  * It is serializable so that frameworks which can use java serialization
  * to propagate data (Spark, Flink...) can send the statistics
  * back. For this reason, TreeMaps are explicitly used as field types,
  * even though IDEs can recommend use of Map instead.
- * <p></p>
+ * For security reasons, untrusted java object streams should never be
+ * deserialized. If for some reason this is required, use
+ * {@link #requiredSerializationClasses()} to get the list of classes
+ * used when deserializing instances of this object.
+ * <p>
+ * <p>
  * It is annotated for correct serializations with jackson2.
+ * </p>
  */
 @SuppressWarnings("CollectionDeclaredAsConcreteClass")
 @InterfaceAudience.Public
@@ -57,6 +66,16 @@ public final class IOStatisticsSnapshot
 
   private static final long serialVersionUID = -1762522703841538084L;
 
+  /**
+   * List of chasses needed to deserialize.
+   */
+  private static final Class[] DESERIALIZATION_CLASSES = {
+      IOStatisticsSnapshot.class,
+      TreeMap.class,
+      Long.class,
+      MeanStatistic.class,
+  };
+
   /**
    * Counters.
    */
@@ -253,4 +272,14 @@ private void readObject(final ObjectInputStream s)
         (TreeMap<String, MeanStatistic>) s.readObject());
   }
 
+  /**
+   * What classes are needed to deserialize this class?
+   * Needed to securely unmarshall this from untrusted sources.
+   * @return a list of required classes to deserialize the data.
+   */
+  public static List<Class> requiredSerializationClasses() {
+    return Arrays.stream(DESERIALIZATION_CLASSES)
+        .collect(Collectors.toList());
+  }
+
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StoreStatisticNames.java

@@ -179,7 +179,7 @@ public final class StoreStatisticNames {
       = "object_delete_request";
 
   /**
-   * The count of objects deleted in delete requests
+   * The count of objects deleted in delete requests.
    */
   public static final String OBJECT_DELETE_OBJECTS
       = "object_delete_objects";

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/IOStatisticsBinding.java

@@ -445,8 +445,23 @@ public static void trackDurationOfInvocation(
       DurationTrackerFactory factory,
       String statistic,
       InvocationRaisingIOE input) throws IOException {
-    trackDurationOfOperation(factory, statistic,
-        input.asCallable()).apply();
+      // create the tracker outside try-with-resources so
+      // that failures can be set in the catcher.
+      DurationTracker tracker = createTracker(factory, statistic);
+      try {
+        // exec the input function and return its value
+        input.apply();
+      } catch (IOException | RuntimeException e) {
+        // input function failed: note it
+        tracker.failed();
+        // and rethrow
+        throw e;
+      } finally {
+        // update the tracker.
+        // this is called after the catch() call would have
+        // set the failed flag.
+        tracker.close();
+      }
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/functional/InvocationRaisingIOE.java

@@ -39,14 +39,4 @@ public interface InvocationRaisingIOE {
    */
   void apply() throws IOException;
 
-  /**
-   * Convert to a callable for ease of passing around.
-   * @return the callable.
-   */
-  default CallableRaisingIOE<Void> asCallable() {
-    return () -> {
-      apply();
-      return null;
-    };
-  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractStreamIOStatisticsTest.java

@@ -22,6 +22,7 @@
 import java.util.List;
 
 import org.assertj.core.api.Assertions;
+import org.junit.AfterClass;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,11 +32,15 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSnapshot;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
 
 import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.extractStatistics;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
 import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.demandStringifyIOStatisticsSource;
+import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsToPrettyString;
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.snapshotIOStatistics;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_BYTES;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_WRITE_BYTES;
 
@@ -54,6 +59,33 @@ public abstract class AbstractContractStreamIOStatisticsTest
   private static final Logger LOG =
       LoggerFactory.getLogger(AbstractContractStreamIOStatisticsTest.class);
 
+  /**
+   * FileSystem statistics are collected across every test case.
+   */
+  protected static final IOStatisticsSnapshot FILESYSTEM_IOSTATS =
+      snapshotIOStatistics();
+
+  @Override
+  public void teardown() throws Exception {
+    final FileSystem fs = getFileSystem();
+    if (fs instanceof IOStatisticsSource) {
+      FILESYSTEM_IOSTATS.aggregate(((IOStatisticsSource)fs).getIOStatistics());
+    }
+    super.teardown();
+  }
+
+  /**
+   * Dump the filesystem statistics after the class if contains any values
+   */
+  @AfterClass
+  public static void dumpFileSystemIOStatistics() {
+    if (!FILESYSTEM_IOSTATS.counters().isEmpty()) {
+      // if there is at least one counter
+      LOG.info("Aggregate FileSystem Statistics {}",
+          ioStatisticsToPrettyString(FILESYSTEM_IOSTATS));
+    }
+  }
+
   @Test
   public void testOutputStreamStatisticKeys() throws Throwable {
     describe("Look at the statistic keys of an output stream");

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/statistics/IOStatisticAssertions.java

@@ -21,10 +21,14 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.io.ObjectStreamClass;
 import java.io.Serializable;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.assertj.core.api.AbstractLongAssert;
 import org.assertj.core.api.ObjectAssert;
@@ -489,10 +493,37 @@ public static IOStatistics statisticsJavaRoundTrip(final IOStatistics stat)
     }
     ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
     IOStatistics deser;
-    try (ObjectInputStream ois = new ObjectInputStream(bais)) {
+    try (ObjectInputStream ois = new RestrictedInput(bais,
+        IOStatisticsSnapshot.requiredSerializationClasses())) {
       deser = (IOStatistics) ois.readObject();
     }
     return deser;
   }
 
+  private static final class RestrictedInput extends ObjectInputStream {
+
+    private final List<String> allowedClasses;
+
+    private RestrictedInput(final InputStream in,
+        final List<Class> allowedClasses) throws IOException {
+
+      super(in);
+      this.allowedClasses = allowedClasses.stream()
+          .map(Class::getName)
+          .collect(Collectors.toList());
+    }
+
+    @Override
+    protected Class<?> resolveClass(final ObjectStreamClass desc)
+        throws IOException, ClassNotFoundException {
+      final String classname = desc.getName();
+      if (!allowedClasses.contains(classname)) {
+        throw new ClassNotFoundException("Class " + classname
+            + " Not in list of allowed classes");
+      }
+
+      return super.resolveClass(desc);
+    }
+  }
+
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/statistics/TestDurationTracking.java

@@ -21,6 +21,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 
 import org.junit.After;
@@ -57,6 +58,8 @@ public class TestDurationTracking extends AbstractHadoopTestBase {
 
   private IOStatisticsStore stats;
 
+  private final AtomicInteger invocationCounter = new AtomicInteger(0);
+
   @Before
   public void setup() {
     stats = iostatisticsStore()
@@ -91,31 +94,47 @@ public void testDurationTryWithResources() throws Throwable {
 
   /**
    * A little sleep method; exceptions are swallowed.
+   * Increments {@link #invocationCounter}.
+   * Increments {@inheritDoc #atomicCounter}.
    */
   public void sleep() {
     sleepf(10);
   }
 
   /**
    * A little sleep function; exceptions are swallowed.
+   * Increments {@link #invocationCounter}.
    */
   protected int sleepf(final int millis) {
+    invocationCounter.incrementAndGet();
     try {
       Thread.sleep(millis);
     } catch (InterruptedException ignored) {
     }
     return millis;
   }
 
+  /**
+   * Assert that the sleep counter has been invoked
+   * the expected number of times.
+   * @param expected expected value
+   */
+  private void assertCounterValue(final int expected) {
+    assertThat(invocationCounter.get())
+        .describedAs("Sleep invocation Counter")
+        .isEqualTo(expected);
+  }
+
   /**
    * Test that a function raising an IOE can be wrapped.
    */
   @Test
   public void testDurationFunctionIOE() throws Throwable {
     FunctionRaisingIOE<Integer, Integer> fn =
         trackFunctionDuration(stats, REQUESTS,
-            (Integer x) -> x);
-    assertThat(fn.apply(1)).isEqualTo(1);
+            (Integer x) -> invocationCounter.getAndSet(x));
+    assertThat(fn.apply(1)).isEqualTo(0);
+    assertCounterValue(1);
     assertSummaryValues(
         fetchSuccessSummary(stats, REQUESTS),
         1, 0, 0);
@@ -194,6 +213,7 @@ public void testCallableFailureDuration() throws Throwable {
           sleepf(100);
           throw new RuntimeException("oops");
         }));
+    assertCounterValue(1);
     assertSummaryValues(
         fetchSuccessSummary(stats, REQUESTS),
         1, -1, -1);
@@ -208,7 +228,10 @@ public void testCallableFailureDuration() throws Throwable {
   @Test
   public void testInvocationDuration() throws Throwable {
     // call the operation
-    trackDurationOfInvocation(stats, REQUESTS, () -> sleepf(100));
+    trackDurationOfInvocation(stats, REQUESTS, () -> {
+      sleepf(100);
+    });
+    assertCounterValue(1);
     DurationStatisticSummary summary = fetchSuccessSummary(stats, REQUESTS);
     assertSummaryValues(summary, 1, 0, 0);
     assertSummaryMean(summary, 1, 0);