HADOOP-18170. Collect disk I/Os time on the node

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

@@ -39,6 +39,9 @@
 @InterfaceStability.Evolving
 public class DF extends Shell {
 
+  /** Default DF refresh interval. */
+  public static final long DF_INTERVAL_DEFAULT = 3 * 1000;
+
   private final String dirPath;
   private final File dirFile;
   private String filesystem;

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

@@ -0,0 +1,109 @@
+/**
+ * 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.util;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import java.math.BigInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class IoTimeTracker {
+  public static final int UNAVAILABLE = -1;
+  final long MINIMUM_UPDATE_INTERVAL;
+
+  // IO used time since system is on (ms)
+  BigInteger cumulativeIoTime = BigInteger.ZERO;
+
+  // IO used time read last time (ms)
+  BigInteger lastCumulativeIoTime = BigInteger.ZERO;
+
+  // Unix timestamp while reading the IO time (ms)
+  long sampleTime;
+  long lastSampleTime;
+  float ioUsage;
+  BigInteger jiffyLengthInMillis;
+
+  private ReadWriteLock rwLock = new ReentrantReadWriteLock();
+  private Lock readLock = rwLock.readLock();
+  public Lock writeLock = rwLock.writeLock();
+
+  public IoTimeTracker(long jiffyLengthInMillis) {
+    this.jiffyLengthInMillis = BigInteger.valueOf(jiffyLengthInMillis);
+    this.ioUsage = UNAVAILABLE;
+    this.sampleTime = UNAVAILABLE;
+    this.lastSampleTime = UNAVAILABLE;
+    MINIMUM_UPDATE_INTERVAL =  10 * jiffyLengthInMillis;
+  }
+
+  /**
+   * Return percentage of io time spent over the time since last update.
+   * IO time spent is based on elapsed jiffies multiplied by amount of
+   * time for 1 disk. Thus, if you use 2 disks completely you would have spent
+   * twice the actual time between updates and this will return 200%.
+   *
+   * @return Return percentage of io usage since last update, {@link
+   * IoTimeTracker#UNAVAILABLE} if there haven't been 2 updates more than
+   * {@link IoTimeTracker#MINIMUM_UPDATE_INTERVAL} apart
+   */
+  public float getIoTrackerUsagePercent() {
+    readLock.lock();
+    try {
+      if (lastSampleTime == UNAVAILABLE || lastSampleTime > sampleTime) {
+        // lastSampleTime > sampleTime may happen when the system time is changed
+        lastSampleTime = sampleTime;
+        lastCumulativeIoTime = cumulativeIoTime;
+        return ioUsage;
+      }
+      // When lastSampleTime is sufficiently old, update ioUsage.
+      // Also take a sample of the current time and cumulative IO time for the
+      // use of the next calculation.
+      if (sampleTime > lastSampleTime + MINIMUM_UPDATE_INTERVAL) {
+        ioUsage = ((cumulativeIoTime.subtract(lastCumulativeIoTime)).floatValue())
+            / ((float) (sampleTime - lastSampleTime));
+        lastSampleTime = sampleTime;
+        lastCumulativeIoTime = cumulativeIoTime;
+      }
+    } finally {
+      readLock.unlock();
+    }
+    return ioUsage;
+  }
+
+  public void updateElapsedJiffies(BigInteger totalIoPerDisk, long sampleTime) {
+    this.cumulativeIoTime = this.cumulativeIoTime.add(totalIoPerDisk);
+    this.sampleTime = sampleTime;
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    sb.append("SampleTime " + this.sampleTime);
+    sb.append(" CummulativeIoTime " + this.cumulativeIoTime);
+    sb.append(" LastSampleTime " + this.lastSampleTime);
+    sb.append(" LastCummulativeIoTime " + this.lastCumulativeIoTime);
+    sb.append(" IoUsage " + this.ioUsage);
+    sb.append(" JiffyLengthMillisec " + this.jiffyLengthInMillis);
+    return sb.toString();
+  }
+}

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

@@ -0,0 +1,38 @@
+package org.apache.hadoop.util;
+
+public class NodeResource {
+
+    float cpuUsage;
+    float ioUsage;
+    float memoryUsage;
+
+    public NodeResource(float cpuUsage, float ioUsage, float memoryUsage) {
+        this.cpuUsage = cpuUsage;
+        this.ioUsage = ioUsage;
+        this.memoryUsage = memoryUsage;
+    }
+
+    public float getCpuUsage() {
+        return this.cpuUsage;
+    }
+
+    public float getIoUsage() {
+        return this.ioUsage;
+    }
+
+    public float getMemoryUsage() {
+        return this.memoryUsage;
+    }
+
+    public void setCpuUsage(float cpuUsage) {
+        this.cpuUsage = cpuUsage;
+    }
+
+    public void setIoUsage(float ioUsage) {
+        this.ioUsage = ioUsage;
+    }
+
+    public void setMemoryUsage(float memoryUsage) {
+        this.memoryUsage = memoryUsage;
+    }
+}

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

@@ -20,6 +20,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
 
 /**
  * Plugin to calculate resource information on the system.
@@ -108,6 +109,21 @@ public static SysInfo newInstance() {
    */
   public abstract float getCpuUsagePercentage();
 
+  /**
+   * Obtain the IO usage % of the machine. Return -1 if it is unavailable
+   *
+   * @return IO usage in %
+   */
+  public abstract float getIoUsagePercentage(String[] paths);
+
+  /**
+   * Obtain the node resource of the machine. Return null if it is unavailable
+   *
+   * @return cpu & io & memory usage in %
+   */
+  public abstract NodeResource getNodeResourceLastPeriod(
+      String[] localDirs, long millis);
+
   /**
    * Obtain the number of VCores used. Return -1 if it is unavailable
    *
@@ -132,13 +148,13 @@ public static SysInfo newInstance() {
    *
    * @return total number of bytes read.
    */
-  public abstract long getStorageBytesRead();
+  public abstract long getStorageBytesRead(String[] paths);
 
   /**
    * Obtain the aggregated number of bytes written to disks.
    *
    * @return total number of bytes written.
    */
-  public abstract long getStorageBytesWritten();
+  public abstract long getStorageBytesWritten(String[] paths);
 
 }