HBASE-27536: improve slowlog payload

Author: Ray Mattingly

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.client;
 
+import java.util.List;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
@@ -71,6 +72,10 @@ final public class OnlineLogRecord extends LogEntry {
   private final int multiGetsCount;
   private final int multiMutationsCount;
   private final int multiServiceCalls;
+  private final Scan scan;
+  private final List<Operation> multi;
+  private final Get get;
+  private final Mutation mutate;
 
   public long getStartTime() {
     return startTime;
@@ -128,11 +133,28 @@ public int getMultiServiceCalls() {
     return multiServiceCalls;
   }
 
+  public Scan getScan() {
+    return scan;
+  }
+
+  public List<Operation> getMulti() {
+    return multi;
+  }
+
+  public Get getGet() {
+    return get;
+  }
+
+  public Mutation getMutate() {
+    return mutate;
+  }
+
   private OnlineLogRecord(final long startTime, final int processingTime, final int queueTime,
     final long responseSize, final String clientAddress, final String serverClass,
     final String methodName, final String callDetails, final String param, final String regionName,
     final String userName, final int multiGetsCount, final int multiMutationsCount,
-    final int multiServiceCalls) {
+    final int multiServiceCalls, final Scan scan, final List<Operation> multi, final Get get,
+    final Mutation mutate) {
     this.startTime = startTime;
     this.processingTime = processingTime;
     this.queueTime = queueTime;
@@ -147,6 +169,10 @@ private OnlineLogRecord(final long startTime, final int processingTime, final in
     this.multiGetsCount = multiGetsCount;
     this.multiMutationsCount = multiMutationsCount;
     this.multiServiceCalls = multiServiceCalls;
+    this.scan = scan;
+    this.multi = multi;
+    this.get = get;
+    this.mutate = mutate;
   }
 
   public static class OnlineLogRecordBuilder {
@@ -164,6 +190,10 @@ public static class OnlineLogRecordBuilder {
     private int multiGetsCount;
     private int multiMutationsCount;
     private int multiServiceCalls;
+    private Scan scan = null;
+    private List<Operation> multi = null;
+    private Get get = null;
+    private Mutation mutate = null;
 
     public OnlineLogRecordBuilder setStartTime(long startTime) {
       this.startTime = startTime;
@@ -235,10 +265,30 @@ public OnlineLogRecordBuilder setMultiServiceCalls(int multiServiceCalls) {
       return this;
     }
 
+    public OnlineLogRecordBuilder setScan(Scan scan) {
+      this.scan = scan;
+      return this;
+    }
+
+    public OnlineLogRecordBuilder setMulti(List<Operation> multi) {
+      this.multi = multi;
+      return this;
+    }
+
+    public OnlineLogRecordBuilder setGet(Get get) {
+      this.get = get;
+      return this;
+    }
+
+    public OnlineLogRecordBuilder setMutate(Mutation mutate) {
+      this.mutate = mutate;
+      return this;
+    }
+
     public OnlineLogRecord build() {
       return new OnlineLogRecord(startTime, processingTime, queueTime, responseSize, clientAddress,
         serverClass, methodName, callDetails, param, regionName, userName, multiGetsCount,
-        multiMutationsCount, multiServiceCalls);
+        multiMutationsCount, multiServiceCalls, scan, multi, get, mutate);
     }
   }
 
@@ -261,15 +311,17 @@ public boolean equals(Object o) {
       .append(multiServiceCalls, that.multiServiceCalls).append(clientAddress, that.clientAddress)
       .append(serverClass, that.serverClass).append(methodName, that.methodName)
       .append(callDetails, that.callDetails).append(param, that.param)
-      .append(regionName, that.regionName).append(userName, that.userName).isEquals();
+      .append(regionName, that.regionName).append(userName, that.userName).append(scan, that.scan)
+      .append(multi, that.multi).append(get, that.get).append(mutate, that.mutate).isEquals();
   }
 
   @Override
   public int hashCode() {
     return new HashCodeBuilder(17, 37).append(startTime).append(processingTime).append(queueTime)
       .append(responseSize).append(clientAddress).append(serverClass).append(methodName)
       .append(callDetails).append(param).append(regionName).append(userName).append(multiGetsCount)
-      .append(multiMutationsCount).append(multiServiceCalls).toHashCode();
+      .append(multiMutationsCount).append(multiServiceCalls).append(scan).append(multi).append(get)
+      .append(mutate).toHashCode();
   }
 
   @Override
@@ -286,7 +338,8 @@ public String toString() {
       .append("callDetails", callDetails).append("param", param).append("regionName", regionName)
       .append("userName", userName).append("multiGetsCount", multiGetsCount)
       .append("multiMutationsCount", multiMutationsCount)
-      .append("multiServiceCalls", multiServiceCalls).toString();
+      .append("multiServiceCalls", multiServiceCalls).append("scan", scan).append("multi", multi)
+      .append("get", get).append("mutate", mutate).toString();
   }
 
 }

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

@@ -23,6 +23,8 @@
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.yetus.audience.InterfaceAudience;
 
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
+
 /**
  * SlowLog params object that contains detailed info as params and region name : to be used for
  * filter purpose
@@ -32,15 +34,63 @@ public class SlowLogParams {
 
   private final String regionName;
   private final String params;
+  private final ClientProtos.Scan scan;
+  private final ClientProtos.MultiRequest multi;
+  private final ClientProtos.Get get;
+  private final ClientProtos.MutationProto mutate;
+
+  public SlowLogParams(String regionName, String params, ClientProtos.Scan scan) {
+    this.regionName = regionName;
+    this.params = params;
+    this.scan = scan;
+    this.multi = null;
+    this.get = null;
+    this.mutate = null;
+  }
+
+  public SlowLogParams(String regionName, String params, ClientProtos.MultiRequest multi) {
+    this.regionName = regionName;
+    this.params = params;
+    this.scan = null;
+    this.multi = multi;
+    this.get = null;
+    this.mutate = null;
+  }
+
+  public SlowLogParams(String regionName, String params, ClientProtos.Get get) {
+    this.regionName = regionName;
+    this.params = params;
+    this.scan = null;
+    this.multi = null;
+    this.get = get;
+    this.mutate = null;
+  }
+
+  public SlowLogParams(String regionName, String params, ClientProtos.MutationProto mutate) {
+    this.regionName = regionName;
+    this.params = params;
+    this.scan = null;
+    this.multi = null;
+    this.get = null;
+    this.mutate = mutate;
+  }
 
   public SlowLogParams(String regionName, String params) {
     this.regionName = regionName;
     this.params = params;
+    this.scan = null;
+    this.multi = null;
+    this.get = null;
+    this.mutate = null;
   }
 
   public SlowLogParams(String params) {
     this.regionName = StringUtils.EMPTY;
     this.params = params;
+    this.scan = null;
+    this.multi = null;
+    this.get = null;
+    this.mutate = null;
   }
 
   public String getRegionName() {
@@ -51,9 +101,26 @@ public String getParams() {
     return params;
   }
 
+  public ClientProtos.Scan getScan() {
+    return scan;
+  }
+
+  public ClientProtos.MultiRequest getMulti() {
+    return multi;
+  }
+
+  public ClientProtos.Get getGet() {
+    return get;
+  }
+
+  public ClientProtos.MutationProto getMutate() {
+    return mutate;
+  }
+
   @Override
   public String toString() {
     return new ToStringBuilder(this).append("regionName", regionName).append("params", params)
+      .append("scan", scan).append("multi", multi).append("get", get).append("mutate", mutate)
       .toString();
   }
 
@@ -67,11 +134,13 @@ public boolean equals(Object o) {
     }
     SlowLogParams that = (SlowLogParams) o;
     return new EqualsBuilder().append(regionName, that.regionName).append(params, that.params)
-      .isEquals();
+      .append(scan, that.scan).append(multi, that.multi).append(get, that.get)
+      .append(mutate, that.mutate).isEquals();
   }
 
   @Override
   public int hashCode() {
-    return new HashCodeBuilder(17, 37).append(regionName).append(params).toHashCode();
+    return new HashCodeBuilder(17, 37).append(regionName).append(params).append(scan).append(multi)
+      .append(get).append(mutate).toHashCode();
   }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java

@@ -87,6 +87,7 @@
 import org.apache.hadoop.hbase.client.LogEntry;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.OnlineLogRecord;
+import org.apache.hadoop.hbase.client.Operation;
 import org.apache.hadoop.hbase.client.PackagePrivateFieldAccessor;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.RegionInfoBuilder;
@@ -129,6 +130,8 @@
 import org.apache.hadoop.hbase.util.VersionInfo;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.hbase.thirdparty.com.google.common.io.ByteStreams;
 import org.apache.hbase.thirdparty.com.google.gson.JsonArray;
@@ -229,6 +232,8 @@
 @InterfaceAudience.Private // TODO: some clients (Hive, etc) use this class
 public final class ProtobufUtil {
 
+  private static final Logger LOG = LoggerFactory.getLogger(ProtobufUtil.class.getName());
+
   private ProtobufUtil() {
   }
 
@@ -1072,6 +1077,22 @@ public static ClientProtos.Scan toScan(final Scan scan) throws IOException {
     return scanBuilder.build();
   }
 
+  public static List<Operation> toMulti(MultiRequest multiRequest) {
+    return multiRequest.getRegionActionList().stream().map(RegionAction::getActionList)
+      .flatMap(Collection::stream).map(action -> {
+        try {
+          if (action.hasGet()) {
+            return ProtobufUtil.toGet(action.getGet());
+          } else if (action.hasMutation()) {
+            return ProtobufUtil.toMutation(action.getMutation());
+          }
+        } catch (IOException e) {
+          LOG.warn("Unable to convert RegionAction to client model: {}", action, e);
+        }
+        return null;
+      }).filter(Objects::nonNull).collect(Collectors.toList());
+  }
+
   /**
    * Convert a protocol buffer Scan to a client Scan
    * @param proto the protocol buffer Scan to convert
@@ -2139,41 +2160,62 @@ private static String getStringForByteString(ByteString bs) {
 
   /**
    * Return SlowLogParams to maintain recent online slowlog responses
-   * @param message Message object {@link Message}
+   * @param message                               Message object {@link Message}
+   * @param slowLogOperationMessagePayloadEnabled whether to include the {@link Message} in the
+   *                                              payload
    * @return SlowLogParams with regionName(for filter queries) and params
    */
-  public static SlowLogParams getSlowLogParams(Message message) {
+  public static SlowLogParams getSlowLogParams(Message message,
+    boolean slowLogOperationMessagePayloadEnabled) {
     if (message == null) {
       return null;
     }
     if (message instanceof ScanRequest) {
       ScanRequest scanRequest = (ScanRequest) message;
       String regionName = getStringForByteString(scanRequest.getRegion().getValue());
       String params = TextFormat.shortDebugString(message);
-      return new SlowLogParams(regionName, params);
+      if (slowLogOperationMessagePayloadEnabled) {
+        return new SlowLogParams(regionName, params, scanRequest.getScan());
+      } else {
+        return new SlowLogParams(regionName, params);
+      }
     } else if (message instanceof MutationProto) {
       MutationProto mutationProto = (MutationProto) message;
       String params = "type= " + mutationProto.getMutateType().toString();
-      return new SlowLogParams(params);
+      if (slowLogOperationMessagePayloadEnabled) {
+        return new SlowLogParams(null, params, mutationProto);
+      } else {
+        return new SlowLogParams(params);
+      }
     } else if (message instanceof GetRequest) {
       GetRequest getRequest = (GetRequest) message;
       String regionName = getStringForByteString(getRequest.getRegion().getValue());
       String params =
         "region= " + regionName + ", row= " + getStringForByteString(getRequest.getGet().getRow());
-      return new SlowLogParams(regionName, params);
+      if (slowLogOperationMessagePayloadEnabled) {
+        return new SlowLogParams(regionName, params, getRequest.getGet());
+      } else {
+        return new SlowLogParams(regionName, params);
+      }
     } else if (message instanceof MultiRequest) {
       MultiRequest multiRequest = (MultiRequest) message;
-      int actionsCount = multiRequest.getRegionActionList().stream()
-        .mapToInt(ClientProtos.RegionAction::getActionCount).sum();
       RegionAction actions = multiRequest.getRegionActionList().get(0);
       String regionName = getStringForByteString(actions.getRegion().getValue());
-      String params = "region= " + regionName + ", for " + actionsCount + " action(s)";
-      return new SlowLogParams(regionName, params);
+      String params = getShortTextFormat(multiRequest);
+      if (slowLogOperationMessagePayloadEnabled) {
+        return new SlowLogParams(regionName, params, multiRequest);
+      } else {
+        return new SlowLogParams(regionName, params);
+      }
     } else if (message instanceof MutateRequest) {
       MutateRequest mutateRequest = (MutateRequest) message;
       String regionName = getStringForByteString(mutateRequest.getRegion().getValue());
       String params = "region= " + regionName;
-      return new SlowLogParams(regionName, params);
+      if (slowLogOperationMessagePayloadEnabled) {
+        return new SlowLogParams(regionName, params, mutateRequest.getMutation());
+      } else {
+        return new SlowLogParams(regionName, params);
+      }
     } else if (message instanceof CoprocessorServiceRequest) {
       CoprocessorServiceRequest coprocessorServiceRequest = (CoprocessorServiceRequest) message;
       String params = "coprocessorService= " + coprocessorServiceRequest.getCall().getServiceName()
@@ -3376,10 +3418,23 @@ private static LogEntry getSlowLogRecord(final TooSlowLog.SlowLogPayload slowLog
         .setQueueTime(slowLogPayload.getQueueTime()).setRegionName(slowLogPayload.getRegionName())
         .setResponseSize(slowLogPayload.getResponseSize())
         .setServerClass(slowLogPayload.getServerClass()).setStartTime(slowLogPayload.getStartTime())
-        .setUserName(slowLogPayload.getUserName()).build();
+        .setUserName(slowLogPayload.getUserName())
+        .setScan(catchAll(() -> ProtobufUtil.toScan(slowLogPayload.getScan())))
+        .setMulti(catchAll(() -> ProtobufUtil.toMulti(slowLogPayload.getMulti())))
+        .setGet(catchAll(() -> ProtobufUtil.toGet(slowLogPayload.getGet())))
+        .setMutate(catchAll(() -> ProtobufUtil.toMutation(slowLogPayload.getMutate()))).build();
     return onlineLogRecord;
   }
 
+  private static <T> T catchAll(Callable<T> callable) {
+    try {
+      return callable.call();
+    } catch (Exception e) {
+      LOG.warn("");
+    }
+    return null;
+  }
+
   /**
    * Convert AdminProtos#SlowLogResponses to list of {@link OnlineLogRecord}
    * @param logEntry slowlog response protobuf instance