HBASE-26730 Extend hbase shell 'status' command to support an option …

…'tasks' Expose monitored tasks state in ClusterStatus API via new option in ServerLoad. Add shell support for interrogating monitored tasks state in ServerLoad.
apache · Feb 16, 2022 · 14be37c · 14be37c
1 parent 7c52895
commit 14be37c
Show file tree

Hide file tree

Showing 15 changed files with 481 additions and 30 deletions.
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ClusterMetrics.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ClusterMetrics.java
@@ -161,6 +161,12 @@ default double getAverageLoad() {
    */
   Map<TableName, RegionStatesCount> getTableRegionStatesCount();
 
+  /**
+   * Provide the list of master tasks
+   */
+  @Nullable
+  List<ServerTask> getMasterTasks();
+
   /**
    * Kinds of ClusterMetrics
    */
@@ -213,5 +219,9 @@ enum Option {
      * metrics about table to no of regions status count
      */
     TABLE_TO_REGIONS_COUNT,
+    /**
+     * metrics about monitored tasks
+     */
+    TASKS,
   }
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ClusterMetricsBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ClusterMetricsBuilder.java
@@ -83,6 +83,10 @@ public static ClusterStatusProtos.ClusterStatus toClusterStatus(ClusterMetrics m
     if (metrics.getMasterName() != null) {
       builder.setMaster(ProtobufUtil.toServerName((metrics.getMasterName())));
     }
+    if (metrics.getMasterTasks() != null) {
+      builder.addAllMasterTasks(metrics.getMasterTasks().stream()
+        .map(t -> ProtobufUtil.toServerTask(t)).collect(Collectors.toList()));
+    }
     if (metrics.getBalancerOn() != null) {
       builder.setBalancerOn(metrics.getBalancerOn());
     }
@@ -122,7 +126,9 @@ public static ClusterMetrics toClusterMetrics(
           proto.getTableRegionStatesCountList().stream()
           .collect(Collectors.toMap(
             e -> ProtobufUtil.toTableName(e.getTableName()),
-            e -> ProtobufUtil.toTableRegionStatesCount(e.getRegionStatesCount()))));
+            e -> ProtobufUtil.toTableRegionStatesCount(e.getRegionStatesCount()))))
+        .setMasterTasks(proto.getMasterTasksList().stream()
+          .map(t -> ProtobufUtil.getServerTask(t)).collect(Collectors.toList()));
     if (proto.hasClusterId()) {
       builder.setClusterId(ClusterId.convert(proto.getClusterId()).toString());
     }
@@ -164,6 +170,7 @@ public static ClusterMetrics.Option toOption(ClusterStatusProtos.Option option)
       case SERVERS_NAME: return ClusterMetrics.Option.SERVERS_NAME;
       case MASTER_INFO_PORT: return ClusterMetrics.Option.MASTER_INFO_PORT;
       case TABLE_TO_REGIONS_COUNT: return ClusterMetrics.Option.TABLE_TO_REGIONS_COUNT;
+      case TASKS: return ClusterMetrics.Option.TASKS;
       // should not reach here
       default: throw new IllegalArgumentException("Invalid option: " + option);
     }
@@ -188,6 +195,7 @@ public static ClusterStatusProtos.Option toOption(ClusterMetrics.Option option)
       case SERVERS_NAME: return Option.SERVERS_NAME;
       case MASTER_INFO_PORT: return ClusterStatusProtos.Option.MASTER_INFO_PORT;
       case TABLE_TO_REGIONS_COUNT: return ClusterStatusProtos.Option.TABLE_TO_REGIONS_COUNT;
+      case TASKS: return ClusterStatusProtos.Option.TASKS;
       // should not reach here
       default: throw new IllegalArgumentException("Invalid option: " + option);
     }
@@ -231,6 +239,8 @@ public static ClusterMetricsBuilder newBuilder() {
   private int masterInfoPort;
   private List<ServerName> serversName = Collections.emptyList();
   private Map<TableName, RegionStatesCount> tableRegionStatesCount = Collections.emptyMap();
+  @Nullable
+  private List<ServerTask> masterTasks;
 
   private ClusterMetricsBuilder() {
   }
@@ -280,6 +290,10 @@ public ClusterMetricsBuilder setServerNames(List<ServerName> serversName) {
     this.serversName = serversName;
     return this;
   }
+  public ClusterMetricsBuilder setMasterTasks(List<ServerTask> masterTasks) {
+    this.masterTasks = masterTasks;
+    return this;
+  }
 
   public ClusterMetricsBuilder setTableRegionStatesCount(
       Map<TableName, RegionStatesCount> tableRegionStatesCount) {
@@ -300,7 +314,8 @@ public ClusterMetrics build() {
         balancerOn,
         masterInfoPort,
         serversName,
-        tableRegionStatesCount
+        tableRegionStatesCount,
+        masterTasks
     );
   }
   private static class ClusterMetricsImpl implements ClusterMetrics {
@@ -320,6 +335,7 @@ private static class ClusterMetricsImpl implements ClusterMetrics {
     private final int masterInfoPort;
     private final List<ServerName> serversName;
     private final Map<TableName, RegionStatesCount> tableRegionStatesCount;
+    private final List<ServerTask> masterTasks;
 
     ClusterMetricsImpl(String hbaseVersion, List<ServerName> deadServerNames,
         Map<ServerName, ServerMetrics> liveServerMetrics,
@@ -331,7 +347,8 @@ private static class ClusterMetricsImpl implements ClusterMetrics {
         Boolean balancerOn,
         int masterInfoPort,
         List<ServerName> serversName,
-        Map<TableName, RegionStatesCount> tableRegionStatesCount) {
+        Map<TableName, RegionStatesCount> tableRegionStatesCount,
+        List<ServerTask> masterTasks) {
       this.hbaseVersion = hbaseVersion;
       this.deadServerNames = Preconditions.checkNotNull(deadServerNames);
       this.liveServerMetrics = Preconditions.checkNotNull(liveServerMetrics);
@@ -344,6 +361,7 @@ private static class ClusterMetricsImpl implements ClusterMetrics {
       this.masterInfoPort = masterInfoPort;
       this.serversName = serversName;
       this.tableRegionStatesCount = Preconditions.checkNotNull(tableRegionStatesCount);
+      this.masterTasks = masterTasks;
     }
 
     @Override
@@ -406,6 +424,11 @@ public Map<TableName, RegionStatesCount> getTableRegionStatesCount() {
       return Collections.unmodifiableMap(tableRegionStatesCount);
     }
 
+    @Override
+    public List<ServerTask> getMasterTasks() {
+      return masterTasks;
+    }
+
     @Override
     public String toString() {
       StringBuilder sb = new StringBuilder(1024);

diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ServerMetrics.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ServerMetrics.java
@@ -124,4 +124,11 @@ default String getVersion() {
    */
   long getLastReportTimestamp();
 
+  /**
+   * Called directly from clients such as the hbase shell
+   * @return the active monitored tasks
+   */
+  @Nullable
+  List<ServerTask> getTasks();
+
 }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ServerMetricsBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ServerMetricsBuilder.java
@@ -87,6 +87,8 @@ public static ServerMetrics toServerMetrics(ServerName serverName, int versionNu
       .setReplicationLoadSink(serverLoadPB.hasReplLoadSink()
         ? ProtobufUtil.toReplicationLoadSink(serverLoadPB.getReplLoadSink())
         : null)
+      .setTasks(serverLoadPB.getTasksList().stream()
+          .map(ProtobufUtil::getServerTask).collect(Collectors.toList()))
       .setReportTimestamp(serverLoadPB.getReportEndTime())
       .setLastReportTimestamp(serverLoadPB.getReportStartTime()).setVersionNumber(versionNumber)
       .setVersion(version).build();
@@ -105,19 +107,24 @@ public static ClusterStatusProtos.ServerLoad toServerLoad(ServerMetrics metrics)
         .setInfoServerPort(metrics.getInfoServerPort())
         .setMaxHeapMB((int) metrics.getMaxHeapSize().get(Size.Unit.MEGABYTE))
         .setUsedHeapMB((int) metrics.getUsedHeapSize().get(Size.Unit.MEGABYTE))
-        .addAllCoprocessors(toCoprocessor(metrics.getCoprocessorNames())).addAllRegionLoads(
+        .addAllCoprocessors(toCoprocessor(metrics.getCoprocessorNames()))
+        .addAllRegionLoads(
             metrics.getRegionMetrics().values().stream().map(RegionMetricsBuilder::toRegionLoad)
-                .collect(Collectors.toList())).addAllUserLoads(
+                .collect(Collectors.toList()))
+        .addAllUserLoads(
             metrics.getUserMetrics().values().stream().map(UserMetricsBuilder::toUserMetrics)
-                .collect(Collectors.toList())).addAllReplLoadSource(
+                .collect(Collectors.toList()))
+        .addAllReplLoadSource(
             metrics.getReplicationLoadSourceList().stream()
                 .map(ProtobufUtil::toReplicationLoadSource).collect(Collectors.toList()))
+        .addAllTasks(
+            metrics.getTasks().stream().map(ProtobufUtil::toServerTask)
+                .collect(Collectors.toList()))
         .setReportStartTime(metrics.getLastReportTimestamp())
         .setReportEndTime(metrics.getReportTimestamp());
     if (metrics.getReplicationLoadSink() != null) {
       builder.setReplLoadSink(ProtobufUtil.toReplicationLoadSink(metrics.getReplicationLoadSink()));
     }
-
     return builder.build();
   }
 
@@ -143,6 +150,8 @@ public static ServerMetricsBuilder newBuilder(ServerName sn) {
   private final Set<String> coprocessorNames = new TreeSet<>();
   private long reportTimestamp = EnvironmentEdgeManager.currentTime();
   private long lastReportTimestamp = 0;
+  private final List<ServerTask> tasks = new ArrayList<>();
+
   private ServerMetricsBuilder(ServerName serverName) {
     this.serverName = serverName;
   }
@@ -228,6 +237,11 @@ public ServerMetricsBuilder setLastReportTimestamp(long value) {
     return this;
   }
 
+  public ServerMetricsBuilder setTasks(List<ServerTask> tasks) {
+    this.tasks.addAll(tasks);
+    return this;
+  }
+
   public ServerMetrics build() {
     return new ServerMetricsImpl(
         serverName,
@@ -246,7 +260,8 @@ public ServerMetrics build() {
         coprocessorNames,
         reportTimestamp,
         lastReportTimestamp,
-        userMetrics);
+        userMetrics,
+        tasks);
   }
 
   private static class ServerMetricsImpl implements ServerMetrics {
@@ -268,13 +283,15 @@ private static class ServerMetricsImpl implements ServerMetrics {
     private final long reportTimestamp;
     private final long lastReportTimestamp;
     private final Map<byte[], UserMetrics> userMetrics;
+    private final List<ServerTask> tasks;
 
     ServerMetricsImpl(ServerName serverName, int versionNumber, String version,
         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) {
+        long reportTimestamp, long lastReportTimestamp, Map<byte[], UserMetrics> userMetrics,
+        List<ServerTask> tasks) {
       this.serverName = Preconditions.checkNotNull(serverName);
       this.versionNumber = versionNumber;
       this.version = version;
@@ -292,6 +309,7 @@ private static class ServerMetricsImpl implements ServerMetrics {
       this.coprocessorNames =Preconditions.checkNotNull(coprocessorNames);
       this.reportTimestamp = reportTimestamp;
       this.lastReportTimestamp = lastReportTimestamp;
+      this.tasks = tasks;
     }
 
     @Override
@@ -388,6 +406,11 @@ public long getLastReportTimestamp() {
       return lastReportTimestamp;
     }
 
+    @Override
+    public List<ServerTask> getTasks() {
+      return tasks;
+    }
+
     @Override
     public String toString() {
       int storeCount = 0;

diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ServerTask.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ServerTask.java
@@ -0,0 +1,59 @@
+/*
+ * 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.hbase;
+
+import org.apache.yetus.audience.InterfaceAudience;
+
+/** Information about active monitored server tasks */
+@InterfaceAudience.Public
+public interface ServerTask {
+
+  /** Task state */
+  enum State {
+    RUNNING,
+    WAITING,
+    COMPLETE,
+    ABORTED;
+  }
+
+  /**
+   * @return the task's description, typically a name
+   */
+  String getDescription();
+
+  /**
+   * @return the task's current status
+   */
+  String getStatus();
+
+  /**
+   * @return the task's current state
+   */
+  State getState();
+
+  /**
+   * @return the time when the task started, or 0 if it has not started yet
+   */
+  long getStartTime();
+
+  /**
+   * @return the time when the task completed, or 0 if it has not completed yet
+   */
+  long getCompletionTime();
+
+}