HBASE-25799 add clusterReadRequests and clusterWriteRequests jmx (#3188)

* HBASE-25799 add clusterReadRequests and clusterWriteRequests jmx

Co-authored-by: stevenxi <[email protected]>

Author: leyangyueshan

hbase-client/src/main/java/org/apache/hadoop/hbase/ServerMetrics.java

@@ -57,6 +57,16 @@ default String getVersion() {
    */
   long getRequestCount();
 
+  /**
+   * @return total Number of read requests from the start of the region server.
+   */
+  long getReadRequestsCount();
+
+  /**
+   * @return total Number of write requests from the start of the region server.
+   */
+  long getWriteRequestsCount();
+
   /**
    * @return the amount of used heap
    */

hbase-client/src/main/java/org/apache/hadoop/hbase/ServerMetricsBuilder.java

@@ -72,6 +72,8 @@ public static ServerMetrics toServerMetrics(ServerName serverName, int versionNu
       .setRequestCountPerSecond(serverLoadPB.getNumberOfRequests())
       .setRequestCount(serverLoadPB.getTotalNumberOfRequests())
       .setInfoServerPort(serverLoadPB.getInfoServerPort())
+      .setReadRequestCount(serverLoadPB.getReadRequestsCount())
+      .setWriteRequestCount(serverLoadPB.getWriteRequestsCount())
       .setMaxHeapSize(new Size(serverLoadPB.getMaxHeapMB(), Size.Unit.MEGABYTE))
       .setUsedHeapSize(new Size(serverLoadPB.getUsedHeapMB(), Size.Unit.MEGABYTE))
       .setCoprocessorNames(serverLoadPB.getCoprocessorsList().stream()
@@ -128,6 +130,8 @@ public static ServerMetricsBuilder newBuilder(ServerName sn) {
   private String version = "0.0.0";
   private long requestCountPerSecond;
   private long requestCount;
+  private long readRequestCount;
+  private long writeRequestCount;
   private Size usedHeapSize = Size.ZERO;
   private Size maxHeapSize = Size.ZERO;
   private int infoServerPort;
@@ -163,6 +167,17 @@ public ServerMetricsBuilder setRequestCount(long value) {
     return this;
   }
 
+  public ServerMetricsBuilder setReadRequestCount(long value) {
+    this.readRequestCount = value;
+    return this;
+  }
+
+  public ServerMetricsBuilder setWriteRequestCount(long value) {
+    this.writeRequestCount = value;
+    return this;
+  }
+
+
   public ServerMetricsBuilder setUsedHeapSize(Size value) {
     this.usedHeapSize = value;
     return this;
@@ -220,6 +235,8 @@ public ServerMetrics build() {
         version,
         requestCountPerSecond,
         requestCount,
+        readRequestCount,
+        writeRequestCount,
         usedHeapSize,
         maxHeapSize,
         infoServerPort,
@@ -238,6 +255,8 @@ private static class ServerMetricsImpl implements ServerMetrics {
     private final String version;
     private final long requestCountPerSecond;
     private final long requestCount;
+    private final long readRequestsCount;
+    private final long writeRequestsCount;
     private final Size usedHeapSize;
     private final Size maxHeapSize;
     private final int infoServerPort;
@@ -251,15 +270,18 @@ private static class ServerMetricsImpl implements ServerMetrics {
     private final Map<byte[], UserMetrics> userMetrics;
 
     ServerMetricsImpl(ServerName serverName, int versionNumber, String version,
-        long requestCountPerSecond, long requestCount, Size usedHeapSize, Size maxHeapSize,
+        long requestCountPerSecond, long requestCount, long readRequestsCount,
+        long writeRequestsCount, Size usedHeapSize, Size maxHeapSize,
         int infoServerPort, List<ReplicationLoadSource> sources, ReplicationLoadSink sink,
-        Map<byte[], RegionMetrics> regionStatus, Set<String> coprocessorNames, long reportTimestamp,
-        long lastReportTimestamp, Map<byte[], UserMetrics> userMetrics) {
+        Map<byte[], RegionMetrics> regionStatus, Set<String> coprocessorNames,
+        long reportTimestamp, long lastReportTimestamp, Map<byte[], UserMetrics> userMetrics) {
       this.serverName = Preconditions.checkNotNull(serverName);
       this.versionNumber = versionNumber;
       this.version = version;
       this.requestCountPerSecond = requestCountPerSecond;
       this.requestCount = requestCount;
+      this.readRequestsCount = readRequestsCount;
+      this.writeRequestsCount = writeRequestsCount;
       this.usedHeapSize = Preconditions.checkNotNull(usedHeapSize);
       this.maxHeapSize = Preconditions.checkNotNull(maxHeapSize);
       this.infoServerPort = infoServerPort;
@@ -296,6 +318,16 @@ public long getRequestCount() {
       return requestCount;
     }
 
+    @Override
+    public long getReadRequestsCount() {
+      return readRequestsCount;
+    }
+
+    @Override
+    public long getWriteRequestsCount() {
+      return writeRequestsCount;
+    }
+
     @Override
     public Size getUsedHeapSize() {
       return usedHeapSize;

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterSource.java

@@ -69,6 +69,8 @@ public interface MetricsMasterSource extends BaseSource {
   String OFFLINE_REGION_COUNT_NAME = "offlineRegionCount";
 
   String CLUSTER_REQUESTS_NAME = "clusterRequests";
+  String CLUSTER_READ_REQUESTS_NAME = "clusterReadRequests";
+  String CLUSTER_WRITE_REQUESTS_NAME = "clusterWriteRequests";
   String MASTER_ACTIVE_TIME_DESC = "Master Active Time";
   String MASTER_START_TIME_DESC = "Master Start Time";
   String MASTER_FINISHED_INITIALIZATION_TIME_DESC =
@@ -98,6 +100,22 @@ public interface MetricsMasterSource extends BaseSource {
    */
   void incRequests(final long inc);
 
+  /**
+   * Increment the number of read requests the cluster has seen.
+   *
+   * @param inc Ammount to increment the total by.
+   */
+  void incReadRequests(final long inc);
+
+
+  /**
+   * Increment the number of write requests the cluster has seen.
+   *
+   * @param inc Ammount to increment the total by.
+   */
+  void incWriteRequests(final long inc);
+
+
   /**
    * @return {@link OperationMetrics} containing common metrics for server crash operation
    */

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterSourceImpl.java

@@ -38,6 +38,8 @@ public class MetricsMasterSourceImpl
 
   private final MetricsMasterWrapper masterWrapper;
   private MutableFastCounter clusterRequestsCounter;
+  private MutableFastCounter clusterReadRequestsCounter;
+  private MutableFastCounter clusterWriteRequestsCounter;
 
   private OperationMetrics serverCrashMetrics;
 
@@ -63,6 +65,8 @@ public MetricsMasterSourceImpl(String metricsName,
   public void init() {
     super.init();
     clusterRequestsCounter = metricsRegistry.newCounter(CLUSTER_REQUESTS_NAME, "", 0L);
+    clusterReadRequestsCounter = metricsRegistry.newCounter(CLUSTER_READ_REQUESTS_NAME, "", 0L);
+    clusterWriteRequestsCounter = metricsRegistry.newCounter(CLUSTER_WRITE_REQUESTS_NAME, "", 0L);
 
     /*
      * NOTE: Please refer to HBASE-9774 and HBASE-14282. Based on these two issues, HBase is
@@ -77,6 +81,16 @@ public void incRequests(final long inc) {
     this.clusterRequestsCounter.incr(inc);
   }
 
+  @Override
+  public void incReadRequests(final long inc) {
+    this.clusterReadRequestsCounter.incr(inc);
+  }
+
+  @Override
+  public void incWriteRequests(final long inc) {
+    this.clusterWriteRequestsCounter.incr(inc);
+  }
+
   @Override
   public void getMetrics(MetricsCollector metricsCollector, boolean all) {
 

hbase-protocol-shaded/src/main/protobuf/server/ClusterStatus.proto

@@ -285,6 +285,16 @@ message ServerLoad {
    * The metrics for each user on this region server
    */
   repeated UserLoad userLoads = 12;
+
+  /**
+   * The metrics for read requests count on this region server
+   */
+  optional uint64 read_requests_count = 13;
+
+  /**
+   * The metrics for write requests on this region server
+   */
+  optional uint64 write_requests_count = 14;
 }
 
 message LiveServerInfo {

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java

@@ -599,7 +599,12 @@ public RegionServerReportResponse regionServerReport(RpcController controller,
       if (sl != null && master.metricsMaster != null) {
         // Up our metrics.
         master.metricsMaster.incrementRequests(
-          sl.getTotalNumberOfRequests() - (oldLoad != null ? oldLoad.getRequestCount() : 0));
+          sl.getTotalNumberOfRequests() -
+                  (oldLoad != null ? oldLoad.getRequestCount() : 0));
+        master.metricsMaster.incrementReadRequests(sl.getReadRequestsCount() -
+                (oldLoad != null ? oldLoad.getReadRequestsCount() : 0));
+        master.metricsMaster.incrementWriteRequests(sl.getWriteRequestsCount() -
+                (oldLoad != null ? oldLoad.getWriteRequestsCount() : 0));
       }
     } catch (IOException ioe) {
       throw new ServiceException(ioe);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MetricsMaster.java

@@ -75,6 +75,20 @@ public void incrementRequests(final long inc) {
     masterSource.incRequests(inc);
   }
 
+  /**
+   * @param inc How much to add to read requests.
+   */
+  public void incrementReadRequests(final long inc) {
+    masterSource.incReadRequests(inc);
+  }
+
+  /**
+   * @param inc How much to add to write requests.
+   */
+  public void incrementWriteRequests(final long inc) {
+    masterSource.incWriteRequests(inc);
+  }
+
   /**
    * Sets the number of space quotas defined.
    *

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

@@ -1402,6 +1402,8 @@ private ClusterStatusProtos.ServerLoad buildServerLoad(long reportStartTime, lon
     serverLoad.setTotalNumberOfRequests(regionServerWrapper.getTotalRequestCount());
     serverLoad.setUsedHeapMB((int)(usedMemory / 1024 / 1024));
     serverLoad.setMaxHeapMB((int) (maxMemory / 1024 / 1024));
+    serverLoad.setReadRequestsCount(this.metricsRegionServerImpl.getReadRequestsCount());
+    serverLoad.setWriteRequestsCount(this.metricsRegionServerImpl.getWriteRequestsCount());
     Set<String> coprocessors = getWAL(null).getCoprocessorHost().getCoprocessors();
     Builder coprocessorBuilder = Coprocessor.newBuilder();
     for (String coprocessor : coprocessors) {

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterMetrics.java

@@ -149,20 +149,28 @@ public void testClusterRequests() throws Exception {
 
     MetricsMasterSource masterSource = master.getMasterMetrics().getMetricsSource();
     ClusterStatusProtos.ServerLoad sl = ClusterStatusProtos.ServerLoad.newBuilder()
-      .setTotalNumberOfRequests(expectedRequestNumber).build();
+       .setTotalNumberOfRequests(expectedRequestNumber)
+       .setReadRequestsCount(expectedRequestNumber)
+       .setWriteRequestsCount(expectedRequestNumber).build();
     request.setLoad(sl);
 
     master.getMasterRpcServices().regionServerReport(null, request.build());
     metricsHelper.assertCounter("cluster_requests", expectedRequestNumber, masterSource);
+    metricsHelper.assertCounter("cluster_read_requests", expectedRequestNumber, masterSource);
+    metricsHelper.assertCounter("cluster_write_requests", expectedRequestNumber, masterSource);
 
     expectedRequestNumber = 15000;
 
-    sl = ClusterStatusProtos.ServerLoad.newBuilder().setTotalNumberOfRequests(expectedRequestNumber)
-      .build();
+    sl = ClusterStatusProtos.ServerLoad.newBuilder()
+            .setTotalNumberOfRequests(expectedRequestNumber)
+            .setReadRequestsCount(expectedRequestNumber)
+            .setWriteRequestsCount(expectedRequestNumber).build();
     request.setLoad(sl);
 
     master.getMasterRpcServices().regionServerReport(null, request.build());
     metricsHelper.assertCounter("cluster_requests", expectedRequestNumber, masterSource);
+    metricsHelper.assertCounter("cluster_read_requests", expectedRequestNumber, masterSource);
+    metricsHelper.assertCounter("cluster_write_requests", expectedRequestNumber, masterSource);
   }
 
   @Test

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestRegionsRecoveryChore.java

@@ -315,6 +315,16 @@ public long getRequestCount() {
         return 0;
       }
 
+      @Override
+      public long getReadRequestsCount() {
+        return 0;
+      }
+
+      @Override
+      public long getWriteRequestsCount() {
+        return 0;
+      }
+
       @Override
       public Size getUsedHeapSize() {
         return null;