Add hdfs stats for local and remote rack bytes read

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerSource.java

@@ -523,6 +523,12 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
   String ZEROCOPY_BYTES_READ = "zeroCopyBytesRead";
   String ZEROCOPY_BYTES_READ_DESC = "The number of bytes read through HDFS zero copy";
 
+  String LOCAL_RACK_BYTES_READ = "localRackBytesRead";
+  String LOCAL_RACK_BYTES_READ_DESC = "The number of bytes read from the same rack of the RegionServer, but not the local HDFS DataNode";
+
+  String REMOTE_RACK_BYTES_READ = "remoteRackBytesRead";
+  String REMOTE_RACK_BYTES_READ_DESC = "The number of bytes read from a different rack from that of the RegionServer";
+
   String BLOCKED_REQUESTS_COUNT = "blockedRequestCount";
   String BLOCKED_REQUESTS_COUNT_DESC = "The number of blocked requests because of memstore size is "
     + "larger than blockingMemStoreSize";

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerWrapper.java

@@ -539,6 +539,10 @@ public interface MetricsRegionServerWrapper {
   /** Returns Number of bytes read from the local HDFS DataNode. */
   long getLocalBytesRead();
 
+  long getLocalRackBytesRead();
+
+  long getRemoteRackBytesRead();
+
   /** Returns Number of bytes read locally through HDFS short circuit. */
   long getShortCircuitBytesRead();
 

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerSourceImpl.java

@@ -553,6 +553,8 @@ private MetricsRecordBuilder addGaugesToMetricsRecordBuilder(MetricsRecordBuilde
         PERCENT_FILES_LOCAL_SECONDARY_REGIONS_DESC), rsWrap.getPercentFileLocalSecondaryRegions())
       .addGauge(Interns.info(TOTAL_BYTES_READ, TOTAL_BYTES_READ_DESC), rsWrap.getTotalBytesRead())
       .addGauge(Interns.info(LOCAL_BYTES_READ, LOCAL_BYTES_READ_DESC), rsWrap.getLocalBytesRead())
+      .addGauge(Interns.info(LOCAL_RACK_BYTES_READ, LOCAL_RACK_BYTES_READ_DESC), rsWrap.getLocalRackBytesRead())
+      .addGauge(Interns.info(REMOTE_RACK_BYTES_READ, REMOTE_RACK_BYTES_READ_DESC), rsWrap.getRemoteRackBytesRead())
       .addGauge(Interns.info(SHORTCIRCUIT_BYTES_READ, SHORTCIRCUIT_BYTES_READ_DESC),
         rsWrap.getShortCircuitBytesRead())
       .addGauge(Interns.info(ZEROCOPY_BYTES_READ, ZEROCOPY_BYTES_READ_DESC),

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerWrapperImpl.java

@@ -29,6 +29,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.GlobalStorageStatistics;
+import org.apache.hadoop.fs.StorageStatistics;
 import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
@@ -1037,6 +1039,29 @@ public long getLocalBytesRead() {
     return FSDataInputStreamWrapper.getLocalBytesRead();
   }
 
+  @Override
+  public long getLocalRackBytesRead() {
+    return getGlobalStorageStatistic("bytesReadDistanceOfOneOrTwo");
+  }
+
+  @Override
+  public long getRemoteRackBytesRead() {
+    return getGlobalStorageStatistic("bytesReadDistanceOfThreeOrFour")
+      + getGlobalStorageStatistic("bytesReadDistanceOfFiveOrLarger");
+  }
+
+  private static long getGlobalStorageStatistic(String name) {
+    StorageStatistics stats = GlobalStorageStatistics.INSTANCE.get("hdfs");
+    if (stats == null) {
+      return 0;
+    }
+    Long val = stats.getLong(name);
+    if (val == null) {
+      return 0;
+    }
+    return val;
+  }
+
   @Override
   public long getShortCircuitBytesRead() {
     return FSDataInputStreamWrapper.getShortCircuitBytesRead();

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerWrapperStub.java

@@ -532,6 +532,16 @@ public long getLocalBytesRead() {
     return 0;
   }
 
+  @Override
+  public long getLocalRackBytesRead() {
+    return 0;
+  }
+
+  @Override
+  public long getRemoteRackBytesRead() {
+    return 0;
+  }
+
   @Override
   public long getShortCircuitBytesRead() {
     return 0;