HADOOP-16830. Add public IOStatistics API + S3A implementation

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

@@ -29,6 +29,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;
 
@@ -40,7 +43,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
@@ -267,4 +270,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();
+  }
 }

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

@@ -19,6 +19,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.StoreStatisticNames;
 
 import java.util.Iterator;
 
@@ -27,6 +28,7 @@
  * instance.
  */
 @InterfaceAudience.Public
+@InterfaceStability.Stable
 public abstract class StorageStatistics {
 
   /**
@@ -43,39 +45,40 @@ public abstract class StorageStatistics {
    *   underscores.</li>
    *   <li>the value of the constants are lowercase of the constant names.</li>
    * </ul>
+   * See {@link StoreStatisticNames} for the field names used here and elsewhere.
    */
   @InterfaceStability.Evolving
   public interface CommonStatisticNames {
     // The following names are for file system operation invocations
-    String OP_APPEND = "op_append";
-    String OP_COPY_FROM_LOCAL_FILE = "op_copy_from_local_file";
-    String OP_CREATE = "op_create";
-    String OP_CREATE_NON_RECURSIVE = "op_create_non_recursive";
-    String OP_DELETE = "op_delete";
-    String OP_EXISTS = "op_exists";
-    String OP_GET_CONTENT_SUMMARY = "op_get_content_summary";
-    String OP_GET_DELEGATION_TOKEN = "op_get_delegation_token";
-    String OP_GET_FILE_CHECKSUM = "op_get_file_checksum";
-    String OP_GET_FILE_STATUS = "op_get_file_status";
-    String OP_GET_STATUS = "op_get_status";
-    String OP_GLOB_STATUS = "op_glob_status";
-    String OP_IS_FILE = "op_is_file";
-    String OP_IS_DIRECTORY = "op_is_directory";
-    String OP_LIST_FILES = "op_list_files";
-    String OP_LIST_LOCATED_STATUS = "op_list_located_status";
-    String OP_LIST_STATUS = "op_list_status";
-    String OP_MKDIRS = "op_mkdirs";
-    String OP_MODIFY_ACL_ENTRIES = "op_modify_acl_entries";
-    String OP_OPEN = "op_open";
-    String OP_REMOVE_ACL = "op_remove_acl";
-    String OP_REMOVE_ACL_ENTRIES = "op_remove_acl_entries";
-    String OP_REMOVE_DEFAULT_ACL = "op_remove_default_acl";
-    String OP_RENAME = "op_rename";
-    String OP_SET_ACL = "op_set_acl";
-    String OP_SET_OWNER = "op_set_owner";
-    String OP_SET_PERMISSION = "op_set_permission";
-    String OP_SET_TIMES = "op_set_times";
-    String OP_TRUNCATE = "op_truncate";
+    String OP_APPEND = StoreStatisticNames.OP_APPEND;
+    String OP_COPY_FROM_LOCAL_FILE = StoreStatisticNames.OP_COPY_FROM_LOCAL_FILE;
+    String OP_CREATE = StoreStatisticNames.OP_CREATE;
+    String OP_CREATE_NON_RECURSIVE = StoreStatisticNames.OP_CREATE_NON_RECURSIVE;
+    String OP_DELETE = StoreStatisticNames.OP_DELETE;
+    String OP_EXISTS = StoreStatisticNames.OP_EXISTS;
+    String OP_GET_CONTENT_SUMMARY = StoreStatisticNames.OP_GET_CONTENT_SUMMARY;
+    String OP_GET_DELEGATION_TOKEN = StoreStatisticNames.OP_GET_DELEGATION_TOKEN;
+    String OP_GET_FILE_CHECKSUM = StoreStatisticNames.OP_GET_FILE_CHECKSUM;
+    String OP_GET_FILE_STATUS = StoreStatisticNames.OP_GET_FILE_STATUS;
+    String OP_GET_STATUS = StoreStatisticNames.OP_GET_STATUS;
+    String OP_GLOB_STATUS = StoreStatisticNames.OP_GLOB_STATUS;
+    String OP_IS_FILE = StoreStatisticNames.OP_IS_FILE;
+    String OP_IS_DIRECTORY = StoreStatisticNames.OP_IS_DIRECTORY;
+    String OP_LIST_FILES = StoreStatisticNames.OP_LIST_FILES;
+    String OP_LIST_LOCATED_STATUS = StoreStatisticNames.OP_LIST_LOCATED_STATUS;
+    String OP_LIST_STATUS = StoreStatisticNames.OP_LIST_STATUS;
+    String OP_MKDIRS = StoreStatisticNames.OP_MKDIRS;
+    String OP_MODIFY_ACL_ENTRIES = StoreStatisticNames.OP_MODIFY_ACL_ENTRIES;
+    String OP_OPEN = StoreStatisticNames.OP_OPEN;
+    String OP_REMOVE_ACL = StoreStatisticNames.OP_REMOVE_ACL;
+    String OP_REMOVE_ACL_ENTRIES = StoreStatisticNames.OP_REMOVE_ACL_ENTRIES;
+    String OP_REMOVE_DEFAULT_ACL = StoreStatisticNames.OP_REMOVE_DEFAULT_ACL;
+    String OP_RENAME = StoreStatisticNames.OP_RENAME;
+    String OP_SET_ACL = StoreStatisticNames.OP_SET_ACL;
+    String OP_SET_OWNER = StoreStatisticNames.OP_SET_OWNER;
+    String OP_SET_PERMISSION = StoreStatisticNames.OP_SET_PERMISSION;
+    String OP_SET_TIMES = StoreStatisticNames.OP_SET_TIMES;
+    String OP_TRUNCATE = StoreStatisticNames.OP_TRUNCATE;
   }
 
   /**

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

@@ -0,0 +1,151 @@
+/*
+ * 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.fs.statistics;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * IO Statistics.
+ * <p>
+ *   These are low-cost per-instance statistics provided by any IO components.
+ * <p>
+ * The statistics MUST BE for the specific instance of the source;
+ * possibly including aggregate statistics from other objects
+ * created by that stores.
+ * For example, the statistics of a filesystem instance must be unique
+ * to that instant and not shared with any other.
+ * However, those statistics may also collect and aggregate statistics
+ * generated in the use of input and output streams created by that
+ * file system instance.
+ *
+ * <p>
+ * The iterator is a possibly empty iterator over all monitored statistics.
+ * <ol>
+ *   <li>
+ *     The attributes of an instance can be probed for with
+ *     {@link #hasAttribute(Attributes)}.
+ *   </li>
+ *   <li>
+ *     The set of statistic keys SHOULD remain unchanged, and MUST NOT
+ *     ever remove keys.
+ *   </li>
+ *     The statistics MAY BE dynamic: every call to {@code iterator()}
+ *     MAY return a current/recent set of statistics.
+ *     This
+ *   </li>
+ *   <li>
+ *     The values MAY change across invocations of {@code iterator()}.
+ *   </li>
+ *   <li>
+ *     The update MAY be in the iterable() call, or MAY be in the actual
+ *     Iterable.next() operation.
+ *   </li>
+ *   <li>
+ *     The returned Map.Entry instances MUST return the same value on
+ *     repeated getValue() calls.
+ *   </li>
+ *   <li>
+ *     Queries of statistics SHOULD Be fast and Nonblocking to the extent
+ *     that if invoked during a long operation, they will prioritize
+ *     returning fast over most timely values.
+ *   </li>
+ *   <li>
+ *     The statistics MAY lag; especially for statistics collected in separate
+ *     operations (e.g stream IO statistics as provided by a filesystem
+ *     instance).
+ *   </li>
+ *   <li>
+ *     Thread safety: an instance of IOStatistics can be shared across threads;
+ *     a call to @code iterator()} is thread safe.
+ *     The actual Iterable returned MUST NOT be shared across threads.
+ *   </li>
+ *   <li>
+ *     If the instance declares that it has the attribute {@link Attributes#Snapshotted},
+ *     then it will take a snapshot of the attribute values in the call {@link #snapshot()}.
+ *     These values MUST NOT change until a subsequent snapshot() operation.
+ *   </li>
+ *   <li>
+ *     A snapshot MAY NOT be consistent, i.e. during the snapshot operation
+ *     the underlying values may change.
+ *   </li>
+ *
+ * </ol>
+ */
+@InterfaceStability.Unstable
+public interface IOStatistics extends Iterable<Map.Entry<String, Long>> {
+
+  /**
+   * Get the value of a statistic.
+   *
+   * @return The value of the statistic, or null if not tracked.
+   */
+  Long getStatistic(String key);
+
+  /**
+   * Return true if a statistic is being tracked.
+   *
+   * @return True only if the statistic is being tracked.
+   */
+  boolean isTracked(String key);
+
+  /**
+   * Probe for an attribute of this statistics set.
+   * @return true if the source has the attribute.
+   */
+  default boolean hasAttribute(Attributes attr) {
+    return false;
+  }
+
+  /**
+   * Create a snapshot; no-op if not supported.
+   * @return true if this call had any effect
+   */
+  default boolean snapshot() {
+    return false;
+  }
+
+  /**
+   * Get the set of keys.
+   * No guarantees are made about the mutability/immutability
+   * of this set.
+   * @return the set of keys.
+   */
+  Set<String> keys();
+
+  /**
+   * Possible attributes of the statistics.
+   * This is very limited right now
+   */
+  enum Attributes {
+    /** The attributes never update. */
+    Static,
+
+    /** The statistics are dynamic: when you re-read a value it may change. */
+    Dynamic,
+
+    /**
+     * The statistics are actually snapshots, updated when you call snapshot(),
+     * or iterator();
+     */
+    Snapshotted
+  }
+}