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.

Author: apurtell

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,
   }
 }

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);

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();
+
 }

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;

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();
+
+}