[Draft] Improving slow query log payload

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

@@ -71,6 +71,7 @@ final public class OnlineLogRecord extends LogEntry {
   private final int multiGetsCount;
   private final int multiMutationsCount;
   private final int multiServiceCalls;
+  private final String operationJson;
 
   public long getStartTime() {
     return startTime;
@@ -128,11 +129,15 @@ public int getMultiServiceCalls() {
     return multiServiceCalls;
   }
 
+  public String getOperationJson() {
+    return operationJson;
+  }
+
   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 String operationJson) {
     this.startTime = startTime;
     this.processingTime = processingTime;
     this.queueTime = queueTime;
@@ -147,6 +152,7 @@ private OnlineLogRecord(final long startTime, final int processingTime, final in
     this.multiGetsCount = multiGetsCount;
     this.multiMutationsCount = multiMutationsCount;
     this.multiServiceCalls = multiServiceCalls;
+    this.operationJson = operationJson;
   }
 
   public static class OnlineLogRecordBuilder {
@@ -164,6 +170,7 @@ public static class OnlineLogRecordBuilder {
     private int multiGetsCount;
     private int multiMutationsCount;
     private int multiServiceCalls;
+    private String operationJson;
 
     public OnlineLogRecordBuilder setStartTime(long startTime) {
       this.startTime = startTime;
@@ -235,10 +242,15 @@ public OnlineLogRecordBuilder setMultiServiceCalls(int multiServiceCalls) {
       return this;
     }
 
+    public OnlineLogRecordBuilder setOperationJson(String operationJson) {
+      this.operationJson = operationJson;
+      return this;
+    }
+
     public OnlineLogRecord build() {
       return new OnlineLogRecord(startTime, processingTime, queueTime, responseSize, clientAddress,
         serverClass, methodName, callDetails, param, regionName, userName, multiGetsCount,
-        multiMutationsCount, multiServiceCalls);
+        multiMutationsCount, multiServiceCalls, operationJson);
     }
   }
 
@@ -261,15 +273,15 @@ 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(operationJson, that.operationJson).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(operationJson).toHashCode();
   }
 
   @Override
@@ -286,7 +298,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("operationJson", operationJson).toString();
   }
 
 }

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

@@ -32,15 +32,31 @@ public class SlowLogParams {
 
   private final String regionName;
   private final String params;
+  private final String operationJson;
 
-  public SlowLogParams(String regionName, String params) {
+  public SlowLogParams(
+    String regionName,
+    String params,
+    String operationJson
+  ) {
     this.regionName = regionName;
     this.params = params;
+    this.operationJson = operationJson;
+  }
+
+  public SlowLogParams(
+    String regionName,
+    String params
+  ) {
+    this.regionName = regionName;
+    this.params = params;
+    this.operationJson = StringUtils.EMPTY;
   }
 
   public SlowLogParams(String params) {
     this.regionName = StringUtils.EMPTY;
     this.params = params;
+    this.operationJson = StringUtils.EMPTY;
   }
 
   public String getRegionName() {
@@ -51,6 +67,10 @@ public String getParams() {
     return params;
   }
 
+  public String getOperationJson() {
+    return operationJson;
+  }
+
   @Override
   public String toString() {
     return new ToStringBuilder(this).append("regionName", regionName).append("params", params)

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

@@ -43,6 +43,7 @@
 import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.ByteBufferExtendedCell;
@@ -85,6 +86,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;
@@ -212,6 +214,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.RegionEventDescriptor.EventType;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.StoreDescriptor;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Protobufs utility. Be aware that a class named org.apache.hadoop.hbase.protobuf.ProtobufUtil
@@ -223,6 +227,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() {
   }
 
@@ -2094,38 +2100,55 @@ private static String getStringForByteString(ByteString bs) {
    * @param message Message object {@link Message}
    * @return SlowLogParams with regionName(for filter queries) and params
    */
-  public static SlowLogParams getSlowLogParams(Message message) {
+  public static SlowLogParams getSlowLogParams(
+    Message message,
+    boolean slowLogOperationJsonEnabled,
+    int maxCols
+  ) {
     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);
+      return new SlowLogParams(
+        regionName,
+        params,
+        toJson(slowLogOperationJsonEnabled, maxCols, () -> ProtobufUtil.toScan(scanRequest.getScan()))
+      );
     } else if (message instanceof MutationProto) {
       MutationProto mutationProto = (MutationProto) message;
       String params = "type= " + mutationProto.getMutateType().toString();
-      return new SlowLogParams(params);
+      return new SlowLogParams(
+        params,
+        toJson(slowLogOperationJsonEnabled, maxCols, () -> ProtobufUtil.toMutation(mutationProto))
+      );
     } 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);
+      return new SlowLogParams(
+        regionName,
+        params,
+        toJson(slowLogOperationJsonEnabled, maxCols, () -> ProtobufUtil.toGet(getRequest.getGet()))
+      );
     } 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)";
+      String params = getShortTextFormat(multiRequest);
       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);
+      return new SlowLogParams(
+        regionName,
+        params,
+        toJson(slowLogOperationJsonEnabled, maxCols, () -> ProtobufUtil.toMutation(mutateRequest.getMutation()))
+      );
     } else if (message instanceof CoprocessorServiceRequest) {
       CoprocessorServiceRequest coprocessorServiceRequest = (CoprocessorServiceRequest) message;
       String params = "coprocessorService= " + coprocessorServiceRequest.getCall().getServiceName()
@@ -2136,6 +2159,22 @@ public static SlowLogParams getSlowLogParams(Message message) {
     return new SlowLogParams(params);
   }
 
+  private static String toJson(
+    boolean slowLogOperationJsonEnabled,
+    int maxCols,
+    Callable<Operation> operation
+  ) {
+    if (!slowLogOperationJsonEnabled) {
+      return StringUtils.EMPTY;
+    }
+    try {
+      return operation.call().toJSON(maxCols);
+    } catch (Exception e) {
+      LOG.warn("Exception when deriving operation JSON", e);
+    }
+    return StringUtils.EMPTY;
+  }
+
   /**
    * Print out some subset of a MutationProto rather than all of it and its data
    * @param proto Protobuf to print out
@@ -3307,7 +3346,8 @@ 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())
+        .setOperationJson(slowLogPayload.getOperationJson()).build();
     return onlineLogRecord;
   }
 

hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java

@@ -1634,6 +1634,14 @@ public enum OperationStatusCode {
     "hbase.regionserver.slowlog.systable.enabled";
   public static final boolean DEFAULT_SLOW_LOG_SYS_TABLE_ENABLED_KEY = false;
 
+  public static final String SLOW_LOG_OPERATION_JSON_MAX_COLS =
+    "hbase.regionserver.slowlog.operation.json.max.cols";
+  public static final int SLOW_LOG_OPERATION_JSON_MAX_COLS_DEFAULT = 50;
+
+  public static final String SLOW_LOG_OPERATION_JSON_ENABLED =
+    "hbase.regionserver.slowlog.operation.json.enabled";
+  public static final boolean SLOW_LOG_OPERATION_JSON_ENABLED_DEFAULT = false;
+
   public static final String SHELL_TIMESTAMP_FORMAT_EPOCH_KEY =
     "hbase.shell.timestamp.format.epoch";
 

hbase-protocol-shaded/src/main/protobuf/TooSlowLog.proto

@@ -43,6 +43,7 @@ message SlowLogPayload {
   optional int32 multi_mutations = 13 [default = 0];
   optional int32 multi_service_calls = 14 [default = 0];
   required Type type = 15;
+  optional string operation_json = 16;
 
   // SLOW_LOG is RPC call slow in nature whereas LARGE_LOG is RPC call quite large.
   // Majority of times, slow logs are also large logs and hence, ALL is combination of

hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/impl/SlowLogQueueService.java

@@ -64,10 +64,16 @@ public class SlowLogQueueService implements NamedQueueService {
   private final boolean isSlowLogTableEnabled;
   private final SlowLogPersistentService slowLogPersistentService;
   private final Queue<TooSlowLog.SlowLogPayload> slowLogQueue;
+  private final int slowLogOperationJsonMaxCols;
+  private final boolean slowLogOperationJsonEnabled;
 
   public SlowLogQueueService(Configuration conf) {
     this.isOnlineLogProviderEnabled = conf.getBoolean(HConstants.SLOW_LOG_BUFFER_ENABLED_KEY,
       HConstants.DEFAULT_ONLINE_LOG_PROVIDER_ENABLED);
+    this.slowLogOperationJsonMaxCols = conf.getInt(HConstants.SLOW_LOG_OPERATION_JSON_MAX_COLS,
+      HConstants.SLOW_LOG_OPERATION_JSON_MAX_COLS_DEFAULT);
+    this.slowLogOperationJsonEnabled = conf.getBoolean(HConstants.SLOW_LOG_OPERATION_JSON_ENABLED,
+      HConstants.SLOW_LOG_OPERATION_JSON_ENABLED_DEFAULT);
 
     if (!isOnlineLogProviderEnabled) {
       this.isSlowLogTableEnabled = false;
@@ -127,7 +133,7 @@ public void consumeEventFromDisruptor(NamedQueuePayload namedQueuePayload) {
     long endTime = EnvironmentEdgeManager.currentTime();
     int processingTime = (int) (endTime - startTime);
     int qTime = (int) (startTime - receiveTime);
-    final SlowLogParams slowLogParams = ProtobufUtil.getSlowLogParams(param);
+    final SlowLogParams slowLogParams = ProtobufUtil.getSlowLogParams(param, slowLogOperationJsonEnabled, slowLogOperationJsonMaxCols);
     int numGets = 0;
     int numMutations = 0;
     int numServiceCalls = 0;
@@ -158,7 +164,8 @@ public void consumeEventFromDisruptor(NamedQueuePayload namedQueuePayload) {
       .setProcessingTime(processingTime).setQueueTime(qTime)
       .setRegionName(slowLogParams != null ? slowLogParams.getRegionName() : StringUtils.EMPTY)
       .setResponseSize(responseSize).setServerClass(className).setStartTime(startTime).setType(type)
-      .setUserName(userName).build();
+      .setUserName(userName)
+      .setOperationJson(slowLogParams != null ? slowLogParams.getOperationJson() : StringUtils.EMPTY).build();
     slowLogQueue.add(slowLogPayload);
     if (isSlowLogTableEnabled) {
       if (!slowLogPayload.getRegionName().startsWith("hbase:slowlog")) {