HBASE-22978 : Online slow response log

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

@@ -2261,4 +2261,29 @@ boolean snapshotCleanupSwitch(final boolean on, final boolean synchronous)
    */
   boolean isSnapshotCleanupEnabled() throws IOException;
 
+
+  /**
+   * Retrieves online slow RPC logs from the provided list of
+   * RegionServers
+   *
+   * @param serverNames Server names to get slowlog responses from
+   * @param slowLogQueryFilter filter to be used if provided
+   * @return online slowlog response list
+   * @throws IOException if a remote or network exception occurs
+   */
+  List<SlowLogRecord> getSlowLogResponses(final Set<ServerName> serverNames,
+      final SlowLogQueryFilter slowLogQueryFilter) throws IOException;
+
+  /**
+   * Clears online slow RPC logs from the provided list of
+   * RegionServers
+   *
+   * @param serverNames Set of Server names to clean slowlog responses from
+   * @return List of booleans representing if online slowlog response buffer is cleaned
+   *   from each RegionServer
+   * @throws IOException if a remote or network exception occurs
+   */
+  List<Boolean> clearSlowLogResponses(final Set<ServerName> serverNames)
+      throws IOException;
+
 }

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

@@ -960,4 +960,16 @@ public boolean isSnapshotCleanupEnabled() throws IOException {
     return get(admin.isSnapshotCleanupEnabled());
   }
 
+  @Override
+  public List<SlowLogRecord> getSlowLogResponses(final Set<ServerName> serverNames,
+      final SlowLogQueryFilter slowLogQueryFilter) throws IOException {
+    return get(admin.getSlowLogResponses(serverNames, slowLogQueryFilter));
+  }
+
+  @Override
+  public List<Boolean> clearSlowLogResponses(final Set<ServerName> serverNames)
+      throws IOException {
+    return get(admin.clearSlowLogResponses(serverNames));
+  }
+
 }

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

@@ -1506,4 +1506,24 @@ default CompletableFuture<List<Boolean>> hasUserPermissions(List<Permission> per
    */
   CompletableFuture<Boolean> isSnapshotCleanupEnabled();
 
+  /**
+   * Retrieves online slow RPC logs from the provided list of
+   * RegionServers
+   *
+   * @param serverNames Server names to get slowlog responses from
+   * @param slowLogQueryFilter filter to be used if provided
+   * @return Online slowlog response list. The return value wrapped by a {@link CompletableFuture}
+   */
+  CompletableFuture<List<SlowLogRecord>> getSlowLogResponses(final Set<ServerName> serverNames,
+      final SlowLogQueryFilter slowLogQueryFilter);
+
+  /**
+   * Clears online slow RPC logs from the provided list of
+   * RegionServers
+   *
+   * @param serverNames Set of Server names to clean slowlog responses from
+   * @return List of booleans representing if online slowlog response buffer is cleaned
+   *   from each RegionServer. The return value wrapped by a {@link CompletableFuture}
+   */
+  CompletableFuture<List<Boolean>> clearSlowLogResponses(final Set<ServerName> serverNames);
 }

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.client;
 
 import com.google.protobuf.RpcChannel;
+
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
@@ -838,4 +839,15 @@ public CompletableFuture<Boolean> isSnapshotCleanupEnabled() {
     return wrap(rawAdmin.isSnapshotCleanupEnabled());
   }
 
+  @Override
+  public CompletableFuture<List<SlowLogRecord>> getSlowLogResponses(
+      final Set<ServerName> serverNames, final SlowLogQueryFilter slowLogQueryFilter) {
+    return wrap(rawAdmin.getSlowLogResponses(serverNames, slowLogQueryFilter));
+  }
+
+  @Override
+  public CompletableFuture<List<Boolean>> clearSlowLogResponses(Set<ServerName> serverNames) {
+    return wrap(rawAdmin.clearSlowLogResponses(serverNames));
+  }
+
 }

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

@@ -23,6 +23,7 @@
 import static org.apache.hadoop.hbase.util.FutureUtils.unwrapCompletionException;
 import com.google.protobuf.Message;
 import com.google.protobuf.RpcChannel;
+import edu.umd.cs.findbugs.annotations.Nullable;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -45,6 +46,7 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.AsyncMetaTableAccessor;
 import org.apache.hadoop.hbase.CacheEvictionStats;
@@ -103,6 +105,8 @@
 import org.apache.hbase.thirdparty.io.netty.util.HashedWheelTimer;
 import org.apache.hbase.thirdparty.io.netty.util.Timeout;
 import org.apache.hbase.thirdparty.io.netty.util.TimerTask;
+import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
+
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos;
@@ -3874,4 +3878,63 @@ public CompletableFuture<Boolean> isSnapshotCleanupEnabled() {
         .call();
   }
 
+  @Override
+  public CompletableFuture<List<SlowLogRecord>> getSlowLogResponses(
+    @Nullable final Set<ServerName> serverNames,
+    final SlowLogQueryFilter slowLogQueryFilter) {
+    if (CollectionUtils.isEmpty(serverNames)) {
+      return CompletableFuture.completedFuture(Collections.emptyList());
+    }
+    return CompletableFuture.supplyAsync(() -> serverNames.stream()
+      .map((ServerName serverName) ->
+        getSlowLogResponseFromServer(serverName, slowLogQueryFilter))
+      .map(CompletableFuture::join)
+      .flatMap(List::stream)
+      .collect(Collectors.toList()));
+  }
+
+  private CompletableFuture<List<SlowLogRecord>> getSlowLogResponseFromServer(
+    final ServerName serverName, final SlowLogQueryFilter slowLogQueryFilter) {
+    return this.<List<SlowLogRecord>>newAdminCaller()
+      .action((controller, stub) -> this
+        .adminCall(
+          controller, stub, RequestConverter.buildSlowLogResponseRequest(slowLogQueryFilter),
+          AdminService.Interface::getSlowLogResponses,
+          ProtobufUtil::toSlowLogPayloads))
+      .serverName(serverName).call();
+  }
+
+  @Override
+  public CompletableFuture<List<Boolean>> clearSlowLogResponses(
+      @Nullable Set<ServerName> serverNames) {
+    if (CollectionUtils.isEmpty(serverNames)) {
+      return CompletableFuture.completedFuture(Collections.emptyList());
+    }
+    List<CompletableFuture<Boolean>> clearSlowLogResponseList = serverNames.stream()
+      .map(this::clearSlowLogsResponses)
+      .collect(Collectors.toList());
+    return convertToFutureOfList(clearSlowLogResponseList);
+  }
+
+  private CompletableFuture<Boolean> clearSlowLogsResponses(final ServerName serverName) {
+    return this.<Boolean>newAdminCaller()
+      .action(((controller, stub) -> this
+        .adminCall(
+          controller, stub, RequestConverter.buildClearSlowLogResponseRequest(),
+          AdminService.Interface::clearSlowLogsResponses,
+          ProtobufUtil::toClearSlowLogPayload))
+      ).serverName(serverName).call();
+  }
+
+  private static <T> CompletableFuture<List<T>> convertToFutureOfList(
+    List<CompletableFuture<T>> futures) {
+    CompletableFuture<Void> allDoneFuture =
+      CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+    return allDoneFuture.thenApply(v ->
+      futures.stream()
+        .map(CompletableFuture::join)
+        .collect(Collectors.toList())
+    );
+  }
+
 }

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

@@ -0,0 +1,89 @@
+/*
+ *
+ * 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.client;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * SlowLog params object that contains detailed info as params and region name : to be used
+ * for filter purpose
+ */
+@InterfaceAudience.Private
+public class SlowLogParams {
+
+  private final String regionName;
+  private final String params;
+
+  public SlowLogParams(String regionName, String params) {
+    this.regionName = regionName;
+    this.params = params;
+  }
+
+  public SlowLogParams(String params) {
+    this.regionName = StringUtils.EMPTY;
+    this.params = params;
+  }
+
+  public String getRegionName() {
+    return regionName;
+  }
+
+  public String getParams() {
+    return params;
+  }
+
+  @Override
+  public String toString() {
+    return new ToStringBuilder(this)
+      .append("regionName", regionName)
+      .append("params", params)
+      .toString();
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    SlowLogParams that = (SlowLogParams) o;
+
+    return new EqualsBuilder()
+      .append(regionName, that.regionName)
+      .append(params, that.params)
+      .isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37)
+      .append(regionName)
+      .append(params)
+      .toHashCode();
+  }
+}

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

@@ -0,0 +1,122 @@
+/*
+ *
+ * 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.client;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * SlowLog Query Filter with all filter and limit parameters
+ */
+@InterfaceAudience.Private
+public class SlowLogQueryFilter {
+
+  private String regionName;
+  private String clientAddress;
+  private String tableName;
+  private String userName;
+  private int limit = 10;
+
+  public String getRegionName() {
+    return regionName;
+  }
+
+  public void setRegionName(String regionName) {
+    this.regionName = regionName;
+  }
+
+  public String getClientAddress() {
+    return clientAddress;
+  }
+
+  public void setClientAddress(String clientAddress) {
+    this.clientAddress = clientAddress;
+  }
+
+  public String getTableName() {
+    return tableName;
+  }
+
+  public void setTableName(String tableName) {
+    this.tableName = tableName;
+  }
+
+  public String getUserName() {
+    return userName;
+  }
+
+  public void setUserName(String userName) {
+    this.userName = userName;
+  }
+
+  public int getLimit() {
+    return limit;
+  }
+
+  public void setLimit(int limit) {
+    this.limit = limit;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    SlowLogQueryFilter that = (SlowLogQueryFilter) o;
+
+    return new EqualsBuilder()
+      .append(limit, that.limit)
+      .append(regionName, that.regionName)
+      .append(clientAddress, that.clientAddress)
+      .append(tableName, that.tableName)
+      .append(userName, that.userName)
+      .isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37)
+      .append(regionName)
+      .append(clientAddress)
+      .append(tableName)
+      .append(userName)
+      .append(limit)
+      .toHashCode();
+  }
+
+  @Override
+  public String toString() {
+    return new ToStringBuilder(this)
+      .append("regionName", regionName)
+      .append("clientAddress", clientAddress)
+      .append("tableName", tableName)
+      .append("userName", userName)
+      .append("limit", limit)
+      .toString();
+  }
+
+}