CDPD-11948. HADOOP-17450. Add Public IOStatistics API. (apache#2577)

This is the API and implementation classes of HADOOP-16830,
which allows callers to query IO object instances
(filesystems, streams, remote iterators, ...) and other classes
for statistics on their I/O Usage: operation count and min/max/mean
durations.

New Packages

org.apache.hadoop.fs.statistics.
  Public API, including:
    IOStatisticsSource
    IOStatistics
    IOStatisticsSnapshot (seralizable to java objects and json)
    +helper classes for logging and integration
    BufferedIOStatisticsInputStream
       implements IOStatisticsSource and StreamCapabilities
     BufferedIOStatisticsOutputStream
       implements IOStatisticsSource, Syncable and StreamCapabilities

org.apache.hadoop.fs.statistics.impl
  Implementation classes for internal use.

org.apache.hadoop.util.functional
  functional programming support for RemoteIterators and
  other operations which raise IOEs; all wrapper classes
  implement and propagate IOStatisticsSource

Contributed by Steve Loughran.

Change-Id: If56e8db2981613ff689c39239135e44feb25f78e

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoInputStream.java

@@ -46,9 +46,13 @@
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.StreamCapabilitiesPolicy;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.util.StringUtils;
 
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
+
 /**
  * CryptoInputStream decrypts data. It is not thread-safe. AES CTR mode is
  * required in order to ensure that the plain text and cipher text have a 1:1
@@ -66,7 +70,7 @@ public class CryptoInputStream extends FilterInputStream implements
     Seekable, PositionedReadable, ByteBufferReadable, HasFileDescriptor, 
     CanSetDropBehind, CanSetReadahead, HasEnhancedByteBufferAccess, 
     ReadableByteChannel, CanUnbuffer, StreamCapabilities,
-    ByteBufferPositionedReadable {
+    ByteBufferPositionedReadable, IOStatisticsSource {
   private final byte[] oneByteBuf = new byte[1];
   private final CryptoCodec codec;
   private final Decryptor decryptor;
@@ -867,8 +871,16 @@ public boolean hasCapability(String capability) {
           + " does not expose its stream capabilities.");
       }
       return ((StreamCapabilities) in).hasCapability(capability);
+    case StreamCapabilities.IOSTATISTICS:
+      return (in instanceof StreamCapabilities)
+          && ((StreamCapabilities) in).hasCapability(capability);
     default:
       return false;
     }
   }
+
+  @Override
+  public IOStatistics getIOStatistics() {
+    return retrieveIOStatistics(in);
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoOutputStream.java

@@ -28,9 +28,13 @@
 import org.apache.hadoop.fs.CanSetDropBehind;
 import org.apache.hadoop.fs.StreamCapabilities;
 import org.apache.hadoop.fs.Syncable;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
 
 import com.google.common.base.Preconditions;
 
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
+
 /**
  * CryptoOutputStream encrypts data. It is not thread-safe. AES CTR mode is
  * required in order to ensure that the plain text and cipher text have a 1:1
@@ -48,7 +52,7 @@
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public class CryptoOutputStream extends FilterOutputStream implements 
-    Syncable, CanSetDropBehind, StreamCapabilities {
+    Syncable, CanSetDropBehind, StreamCapabilities, IOStatisticsSource {
   private final byte[] oneByteBuf = new byte[1];
   private final CryptoCodec codec;
   private final Encryptor encryptor;
@@ -313,4 +317,9 @@ public boolean hasCapability(String capability) {
     }
     return false;
   }
+
+  @Override
+  public IOStatistics getIOStatistics() {
+    return retrieveIOStatistics(out);
+  }
 }

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

@@ -24,6 +24,10 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
 
 
 /**
@@ -33,7 +37,8 @@
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 public class BufferedFSInputStream extends BufferedInputStream
-implements Seekable, PositionedReadable, HasFileDescriptor {
+    implements Seekable, PositionedReadable, HasFileDescriptor,
+    IOStatisticsSource, StreamCapabilities {
   /**
    * Creates a <code>BufferedFSInputStream</code>
    * with the specified buffer size,
@@ -126,4 +131,26 @@ public FileDescriptor getFileDescriptor() throws IOException {
       return null;
     }
   }
+
+  /**
+   * If the inner stream supports {@link StreamCapabilities},
+   * forward the probe to it.
+   * Otherwise: return false.
+   *
+   * @param capability string to query the stream support for.
+   * @return true if a capability is known to be supported.
+   */
+  @Override
+  public boolean hasCapability(final String capability) {
+    if (in instanceof StreamCapabilities) {
+      return ((StreamCapabilities) in).hasCapability(capability);
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public IOStatistics getIOStatistics() {
+    return retrieveIOStatistics(in);
+  }
 }

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

@@ -38,6 +38,9 @@
 import org.apache.hadoop.fs.impl.OpenFileParameters;
 import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.statistics.IOStatisticsSupport;
 import org.apache.hadoop.util.DataChecksum;
 import org.apache.hadoop.util.LambdaUtils;
 import org.apache.hadoop.util.Progressable;
@@ -134,7 +137,8 @@ private int getSumBufferSize(int bytesPerSum, int bufferSize) {
    * For open()'s FSInputStream
    * It verifies that data matches checksums.
    *******************************************************/
-  private static class ChecksumFSInputChecker extends FSInputChecker {
+  private static class ChecksumFSInputChecker extends FSInputChecker implements
+      IOStatisticsSource {
     private ChecksumFileSystem fs;
     private FSDataInputStream datas;
     private FSDataInputStream sums;
@@ -270,6 +274,17 @@ protected int readChunk(long pos, byte[] buf, int offset, int len,
       }
       return nread;
     }
+
+    /**
+     * Get the IO Statistics of the nested stream, falling back to
+     * null if the stream does not implement the interface
+     * {@link IOStatisticsSource}.
+     * @return an IOStatistics instance or null
+     */
+    @Override
+    public IOStatistics getIOStatistics() {
+      return IOStatisticsSupport.retrieveIOStatistics(datas);
+    }
   }
 
   private static class FSDataBoundedInputStream extends FSDataInputStream {
@@ -389,7 +404,8 @@ public static long getChecksumLength(long size, int bytesPerSum) {
 
   /** This class provides an output stream for a checksummed file.
    * It generates checksums for data. */
-  private static class ChecksumFSOutputSummer extends FSOutputSummer {
+  private static class ChecksumFSOutputSummer extends FSOutputSummer
+      implements IOStatisticsSource, StreamCapabilities {
     private FSDataOutputStream datas;    
     private FSDataOutputStream sums;
     private static final float CHKSUM_AS_FRACTION = 0.01f;
@@ -443,6 +459,28 @@ protected void checkClosed() throws IOException {
         throw new ClosedChannelException();
       }
     }
+
+    /**
+     * Get the IO Statistics of the nested stream, falling back to
+     * null if the stream does not implement the interface
+     * {@link IOStatisticsSource}.
+     * @return an IOStatistics instance or null
+     */
+    @Override
+    public IOStatistics getIOStatistics() {
+      return IOStatisticsSupport.retrieveIOStatistics(datas);
+    }
+
+    /**
+     * Probe the inner stream for a capability.
+     *
+     * @param capability string to query the stream support for.
+     * @return true if a capability is known to be supported.
+     */
+    @Override
+    public boolean hasCapability(final String capability) {
+      return datas.hasCapability(capability);
+    }
   }
 
   @Override

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

@@ -28,6 +28,9 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.statistics.IOStatisticsSupport;
 import org.apache.hadoop.io.ByteBufferPool;
 import org.apache.hadoop.util.IdentityHashStore;
 
@@ -39,7 +42,7 @@ public class FSDataInputStream extends DataInputStream
     implements Seekable, PositionedReadable, 
       ByteBufferReadable, HasFileDescriptor, CanSetDropBehind, CanSetReadahead,
       HasEnhancedByteBufferAccess, CanUnbuffer, StreamCapabilities,
-      ByteBufferPositionedReadable {
+      ByteBufferPositionedReadable, IOStatisticsSource {
   /**
    * Map ByteBuffers that we have handed out to readers to ByteBufferPool 
    * objects
@@ -266,4 +269,15 @@ public void readFully(long position, ByteBuffer buf) throws IOException {
               "unsupported by " + in.getClass().getCanonicalName());
     }
   }
+
+  /**
+   * Get the IO Statistics of the nested stream, falling back to
+   * null if the stream does not implement the interface
+   * {@link IOStatisticsSource}.
+   * @return an IOStatistics instance or null
+   */
+  @Override
+  public IOStatistics getIOStatistics() {
+    return IOStatisticsSupport.retrieveIOStatistics(in);
+  }
 }

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

@@ -24,13 +24,17 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.statistics.IOStatisticsSupport;
 
 /** Utility that wraps a {@link OutputStream} in a {@link DataOutputStream}.
  */
 @InterfaceAudience.Public
 @InterfaceStability.Stable
 public class FSDataOutputStream extends DataOutputStream
-    implements Syncable, CanSetDropBehind, StreamCapabilities {
+    implements Syncable, CanSetDropBehind, StreamCapabilities,
+      IOStatisticsSource {
   private final OutputStream wrappedStream;
 
   private static class PositionCache extends FilterOutputStream {
@@ -155,4 +159,15 @@ public void setDropBehind(Boolean dropBehind) throws IOException {
           "not support setting the drop-behind caching setting.");
     }
   }
+
+  /**
+   * Get the IO Statistics of the nested stream, falling back to
+   * empty statistics if the stream does not implement the interface
+   * {@link IOStatisticsSource}.
+   * @return an IOStatistics instance.
+   */
+  @Override
+  public IOStatistics getIOStatistics() {
+    return IOStatisticsSupport.retrieveIOStatistics(wrappedStream);
+  }
 }

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

@@ -24,6 +24,9 @@
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -134,4 +137,23 @@ public void readFully(long position, byte[] buffer)
     throws IOException {
     readFully(position, buffer, 0, buffer.length);
   }
+
+  /**
+   * toString method returns the superclass toString, but if the subclass
+   * implements {@link IOStatisticsSource} then those statistics are
+   * extracted and included in the output.
+   * That is: statistics of subclasses are automatically reported.
+   * @return a string value.
+   */
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder(super.toString());
+    sb.append('{');
+    if (this instanceof IOStatisticsSource) {
+      sb.append(IOStatisticsLogging.ioStatisticsSourceToString(
+          (IOStatisticsSource) this));
+    }
+    sb.append('}');
+    return sb.toString();
+  }
 }