From 9f7baa8f4c1d8ccf0b589caa37cad15a29b8e695 Mon Sep 17 00:00:00 2001 From: Ray Mattingly Date: Mon, 19 Dec 2022 18:18:20 -0500 Subject: [PATCH] HBASE-27536: improve slowlog payload --- .../hadoop/hbase/client/OnlineLogRecord.java | 121 +++++++++++++-- .../hadoop/hbase/client/SlowLogParams.java | 73 ++++++++- .../hbase/shaded/protobuf/ProtobufUtil.java | 90 +++++++++-- .../hbase/client/TestOnlineLogRecord.java | 141 ++++++++++++++++++ .../org/apache/hadoop/hbase/HConstants.java | 4 + .../protobuf/server/region/TooSlowLog.proto | 6 + .../namequeues/impl/SlowLogQueueService.java | 26 +++- .../namequeues/TestNamedQueueRecorder.java | 114 +++++++++++++- 8 files changed, 546 insertions(+), 29 deletions(-) create mode 100644 hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOnlineLogRecord.java diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/OnlineLogRecord.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/OnlineLogRecord.java index 96182ca4b296..1abaa73b0a6d 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/OnlineLogRecord.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/OnlineLogRecord.java @@ -17,15 +17,22 @@ */ package org.apache.hadoop.hbase.client; +import java.util.List; +import java.util.Optional; import org.apache.commons.lang3.builder.EqualsBuilder; import org.apache.commons.lang3.builder.HashCodeBuilder; import org.apache.commons.lang3.builder.ToStringBuilder; +import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.util.GsonUtil; import org.apache.yetus.audience.InterfaceAudience; import org.apache.yetus.audience.InterfaceStability; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.hbase.thirdparty.com.google.gson.Gson; +import org.apache.hbase.thirdparty.com.google.gson.JsonElement; import org.apache.hbase.thirdparty.com.google.gson.JsonObject; +import org.apache.hbase.thirdparty.com.google.gson.JsonParser; import org.apache.hbase.thirdparty.com.google.gson.JsonSerializer; /** @@ -36,13 +43,31 @@ @InterfaceStability.Evolving final public class OnlineLogRecord extends LogEntry { - // used to convert object to pretty printed format - // used by toJsonPrettyPrint() + private static final Logger LOG = LoggerFactory.getLogger(OnlineLogRecord.class.getName()); + private static final JsonElement EXCLUDED_NODE = JsonParser.parseString(HConstants.EMPTY_STRING); + + private static JsonElement serializeCatchAll(Operation operation) { + try { + return JsonParser.parseString(operation.toJSON()); + } catch (Exception e) { + LOG.warn("Suppressing exception during OnlineLogRecord serialization with operation {}", + operation, e); + return EXCLUDED_NODE; + } + } + + private static final Gson INNER_GSON = GsonUtil.createGson().setPrettyPrinting() + .registerTypeAdapter(Operation.class, + (JsonSerializer< + Operation>) (operation, type, jsonSerializationContext) -> serializeCatchAll(operation)) + .registerTypeAdapter(Optional.class, + (JsonSerializer>) (optional, type, jsonSerializationContext) -> optional + .map(jsonSerializationContext::serialize).orElse(EXCLUDED_NODE)) + .create(); private static final Gson GSON = GsonUtil.createGson().setPrettyPrinting().registerTypeAdapter(OnlineLogRecord.class, (JsonSerializer) (slowLogPayload, type, jsonSerializationContext) -> { - Gson gson = new Gson(); - JsonObject jsonObj = (JsonObject) gson.toJsonTree(slowLogPayload); + JsonObject jsonObj = (JsonObject) INNER_GSON.toJsonTree(slowLogPayload); if (slowLogPayload.getMultiGetsCount() == 0) { jsonObj.remove("multiGetsCount"); } @@ -71,6 +96,10 @@ final public class OnlineLogRecord extends LogEntry { private final int multiGetsCount; private final int multiMutationsCount; private final int multiServiceCalls; + private final Optional scan; + private final Optional> multi; + private final Optional get; + private final Optional mutate; public long getStartTime() { return startTime; @@ -128,11 +157,52 @@ public int getMultiServiceCalls() { return multiServiceCalls; } - private OnlineLogRecord(final long startTime, final int processingTime, final int queueTime, + /** + * If {@value org.apache.hadoop.hbase.HConstants#SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED} is + * enabled then this value may be present and should represent the Scan that produced the given + * {@link OnlineLogRecord}. This value should only be present if {@link #getMulti()}, + * {@link #getGet()}, and {@link #getMutate()} are empty + */ + public Optional getScan() { + return scan; + } + + /** + * If {@value org.apache.hadoop.hbase.HConstants#SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED} is + * enabled then this value may be present and should represent the MultiRequest that produced the + * given {@link OnlineLogRecord}. This value should only be present if {@link #getScan}, + * {@link #getGet()}, and {@link #getMutate()} are empty + */ + public Optional> getMulti() { + return multi; + } + + /** + * If {@value org.apache.hadoop.hbase.HConstants#SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED} is + * enabled then this value may be present and should represent the Get that produced the given + * {@link OnlineLogRecord}. This value should only be present if {@link #getScan()}, + * {@link #getMulti()} ()}, and {@link #getMutate()} are empty + */ + public Optional getGet() { + return get; + } + + /** + * If {@value org.apache.hadoop.hbase.HConstants#SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED} is + * enabled then this value may be present and should represent the Mutation that produced the + * given {@link OnlineLogRecord}. This value should only be present if {@link #getScan}, + * {@link #getMulti()} ()}, and {@link #getGet()} ()} are empty + */ + public Optional getMutate() { + return mutate; + } + + 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 Optional scan, final Optional> multi, + final Optional get, final Optional mutate) { this.startTime = startTime; this.processingTime = processingTime; this.queueTime = queueTime; @@ -147,6 +217,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 +238,10 @@ public static class OnlineLogRecordBuilder { private int multiGetsCount; private int multiMutationsCount; private int multiServiceCalls; + private Optional scan = Optional.empty(); + private Optional> multi = Optional.empty(); + private Optional get = Optional.empty(); + private Optional mutate = Optional.empty(); public OnlineLogRecordBuilder setStartTime(long startTime) { this.startTime = startTime; @@ -235,10 +313,30 @@ public OnlineLogRecordBuilder setMultiServiceCalls(int multiServiceCalls) { return this; } + public OnlineLogRecordBuilder setScan(Scan scan) { + this.scan = Optional.of(scan); + return this; + } + + public OnlineLogRecordBuilder setMulti(List multi) { + this.multi = Optional.of(multi); + return this; + } + + public OnlineLogRecordBuilder setGet(Get get) { + this.get = Optional.of(get); + return this; + } + + public OnlineLogRecordBuilder setMutate(Mutation mutate) { + this.mutate = Optional.of(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,7 +359,8 @@ 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 @@ -269,7 +368,8 @@ 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 +386,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(); } } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SlowLogParams.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SlowLogParams.java index b1460c0b116c..e15be9b23581 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SlowLogParams.java +++ b/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(); } } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java index ca38a91e74a6..ad414718da9b 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java +++ b/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,21 @@ public static ClientProtos.Scan toScan(final Scan scan) throws IOException { return scanBuilder.build(); } + public static List toMulti(MultiRequest multiRequest) throws IOException { + List operations = new ArrayList<>(); + for (RegionAction regionAction : multiRequest.getRegionActionList()) { + List actions = regionAction.getActionList(); + for (ClientProtos.Action action : actions) { + if (action.hasGet()) { + operations.add(ProtobufUtil.toGet(action.getGet())); + } else if (action.hasMutation()) { + operations.add(ProtobufUtil.toMutation(action.getMutation())); + } + } + } + return operations; + } + /** * Convert a protocol buffer Scan to a client Scan * @param proto the protocol buffer Scan to convert @@ -2139,10 +2159,13 @@ 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; } @@ -2150,30 +2173,48 @@ public static SlowLogParams getSlowLogParams(Message message) { 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() @@ -3365,7 +3406,7 @@ public static Set toCompactedStoreFiles(byte[] bytes) throws IOException * @return SlowLog Payload for client usecase */ private static LogEntry getSlowLogRecord(final TooSlowLog.SlowLogPayload slowLogPayload) { - OnlineLogRecord onlineLogRecord = + OnlineLogRecord.OnlineLogRecordBuilder onlineLogRecordBuilder = new OnlineLogRecord.OnlineLogRecordBuilder().setCallDetails(slowLogPayload.getCallDetails()) .setClientAddress(slowLogPayload.getClientAddress()) .setMethodName(slowLogPayload.getMethodName()) @@ -3376,8 +3417,31 @@ 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(); - return onlineLogRecord; + .setUserName(slowLogPayload.getUserName()); + if (slowLogPayload.hasScan()) { + onlineLogRecordBuilder.setScan(catchAll(() -> ProtobufUtil.toScan(slowLogPayload.getScan()))); + } + if (slowLogPayload.hasMulti()) { + onlineLogRecordBuilder + .setMulti(catchAll(() -> ProtobufUtil.toMulti(slowLogPayload.getMulti()))); + } + if (slowLogPayload.hasGet()) { + onlineLogRecordBuilder.setGet(catchAll(() -> ProtobufUtil.toGet(slowLogPayload.getGet()))); + } + if (slowLogPayload.hasMutate()) { + onlineLogRecordBuilder + .setMutate(catchAll(() -> ProtobufUtil.toMutation(slowLogPayload.getMutate()))); + } + return onlineLogRecordBuilder.build(); + } + + private static T catchAll(Callable callable) { + try { + return callable.call(); + } catch (Exception e) { + LOG.warn("Suppressing exception", e); + } + return null; } /** diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOnlineLogRecord.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOnlineLogRecord.java new file mode 100644 index 000000000000..074f004d9a62 --- /dev/null +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestOnlineLogRecord.java @@ -0,0 +1,141 @@ +/* + * 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 com.google.common.collect.ImmutableList; +import java.util.Optional; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter; +import org.apache.hadoop.hbase.testclassification.ClientTests; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.Assert; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +@Category({ ClientTests.class, SmallTests.class }) +public class TestOnlineLogRecord { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestOnlineLogRecord.class); + + @Test + public void itSerializesScan() { + Scan scan = new Scan(); + scan.withStartRow(Bytes.toBytes(123)); + scan.withStopRow(Bytes.toBytes(456)); + String expectedOutput = "{\n" + " \"startTime\": \"1\",\n" + " \"processingTime\": 2,\n" + + " \"queueTime\": 3,\n" + " \"responseSize\": \"4\",\n" + " \"multiGetsCount\": 5,\n" + + " \"multiMutationsCount\": 6,\n" + " \"multiServiceCalls\": 7,\n" + " \"scan\": {\n" + + " \"startRow\": [\n" + " 0,\n" + " 0,\n" + " 0,\n" + " 123\n" + + " ],\n" + " \"includeStartRow\": true,\n" + " \"stopRow\": [\n" + " 0,\n" + + " 0,\n" + " 1,\n" + " -56\n" + " ],\n" + + " \"includeStopRow\": false,\n" + " \"maxVersions\": 1,\n" + " \"batch\": -1,\n" + + " \"allowPartialResults\": false,\n" + " \"storeLimit\": -1,\n" + + " \"storeOffset\": 0,\n" + " \"caching\": -1,\n" + " \"maxResultSize\": \"-1\",\n" + + " \"cacheBlocks\": true,\n" + " \"reversed\": false,\n" + " \"tr\": {\n" + + " \"minStamp\": \"0\",\n" + " \"maxStamp\": \"9223372036854775807\",\n" + + " \"allTime\": true\n" + " },\n" + " \"familyMap\": {},\n" + + " \"mvccReadPoint\": \"-1\",\n" + " \"limit\": -1,\n" + + " \"readType\": \"DEFAULT\",\n" + " \"needCursorResult\": false,\n" + + " \"targetReplicaId\": -1,\n" + " \"consistency\": \"STRONG\",\n" + + " \"colFamTimeRangeMap\": {},\n" + " \"priority\": -1\n" + " }\n" + "}"; + OnlineLogRecord o = new OnlineLogRecord(1, 2, 3, 4, null, null, null, null, null, null, null, 5, + 6, 7, Optional.of(scan), Optional.empty(), Optional.empty(), Optional.empty()); + Assert.assertEquals(o.toJsonPrettyPrint(), expectedOutput); + } + + @Test + public void itSerializesMulti() { + Get get1 = new Get(Bytes.toBytes(123)); + Get get2 = new Get(Bytes.toBytes(456)); + get2.setFilter(new FirstKeyOnlyFilter()); + String expectedOutput = "{\n" + " \"startTime\": \"1\",\n" + " \"processingTime\": 2,\n" + + " \"queueTime\": 3,\n" + " \"responseSize\": \"4\",\n" + " \"multiGetsCount\": 5,\n" + + " \"multiMutationsCount\": 6,\n" + " \"multiServiceCalls\": 7,\n" + " \"multi\": [\n" + + " {\n" + " \"row\": [\n" + " 0,\n" + " 0,\n" + " 0,\n" + + " 123\n" + " ],\n" + " \"maxVersions\": 1,\n" + + " \"cacheBlocks\": true,\n" + " \"storeLimit\": -1,\n" + + " \"storeOffset\": 0,\n" + " \"tr\": {\n" + " \"minStamp\": \"0\",\n" + + " \"maxStamp\": \"9223372036854775807\",\n" + " \"allTime\": true\n" + + " },\n" + " \"checkExistenceOnly\": false,\n" + " \"familyMap\": {},\n" + + " \"targetReplicaId\": -1,\n" + " \"consistency\": \"STRONG\",\n" + + " \"colFamTimeRangeMap\": {},\n" + " \"priority\": -1\n" + " },\n" + " {\n" + + " \"row\": [\n" + " 0,\n" + " 0,\n" + " 1,\n" + " -56\n" + + " ],\n" + " \"maxVersions\": 1,\n" + " \"cacheBlocks\": true,\n" + + " \"storeLimit\": -1,\n" + " \"storeOffset\": 0,\n" + " \"tr\": {\n" + + " \"minStamp\": \"0\",\n" + " \"maxStamp\": \"9223372036854775807\",\n" + + " \"allTime\": true\n" + " },\n" + " \"checkExistenceOnly\": false,\n" + + " \"familyMap\": {},\n" + " \"filter\": {\n" + " \"foundKV\": false\n" + + " },\n" + " \"targetReplicaId\": -1,\n" + " \"consistency\": \"STRONG\",\n" + + " \"colFamTimeRangeMap\": {},\n" + " \"priority\": -1\n" + " }\n" + " ]\n" + + "}"; + OnlineLogRecord o = new OnlineLogRecord(1, 2, 3, 4, null, null, null, null, null, null, null, 5, + 6, 7, Optional.empty(), + Optional.of(ImmutableList. builder().add(get1).add(get2).build()), + Optional.empty(), Optional.empty()); + Assert.assertEquals(o.toJsonPrettyPrint(), expectedOutput); + } + + @Test + public void itSerializesGet() { + Get get = new Get(Bytes.toBytes(123)); + String expectedOutput = "{\n" + " \"startTime\": \"1\",\n" + " \"processingTime\": 2,\n" + + " \"queueTime\": 3,\n" + " \"responseSize\": \"4\",\n" + " \"multiGetsCount\": 5,\n" + + " \"multiMutationsCount\": 6,\n" + " \"multiServiceCalls\": 7,\n" + " \"get\": {\n" + + " \"row\": [\n" + " 0,\n" + " 0,\n" + " 0,\n" + " 123\n" + " ],\n" + + " \"maxVersions\": 1,\n" + " \"cacheBlocks\": true,\n" + " \"storeLimit\": -1,\n" + + " \"storeOffset\": 0,\n" + " \"tr\": {\n" + " \"minStamp\": \"0\",\n" + + " \"maxStamp\": \"9223372036854775807\",\n" + " \"allTime\": true\n" + " },\n" + + " \"checkExistenceOnly\": false,\n" + " \"familyMap\": {},\n" + + " \"targetReplicaId\": -1,\n" + " \"consistency\": \"STRONG\",\n" + + " \"colFamTimeRangeMap\": {},\n" + " \"priority\": -1\n" + " }\n" + "}"; + OnlineLogRecord o = new OnlineLogRecord(1, 2, 3, 4, null, null, null, null, null, null, null, 5, + 6, 7, Optional.empty(), Optional.empty(), Optional.of(get), Optional.empty()); + Assert.assertEquals(o.toJsonPrettyPrint(), expectedOutput); + } + + @Test + public void itSerializesMutate() { + Mutation mutate = new Put(Bytes.toBytes(123)).addColumn(Bytes.toBytes(456), Bytes.toBytes(789), + Bytes.toBytes(0)); + String expectedOutput = "{\n" + " \"startTime\": \"1\",\n" + " \"processingTime\": 2,\n" + + " \"queueTime\": 3,\n" + " \"responseSize\": \"4\",\n" + " \"multiGetsCount\": 5,\n" + + " \"multiMutationsCount\": 6,\n" + " \"multiServiceCalls\": 7,\n" + " \"mutate\": {\n" + + " \"row\": [\n" + " 0,\n" + " 0,\n" + " 0,\n" + " 123\n" + " ],\n" + + " \"ts\": \"9223372036854775807\",\n" + " \"durability\": \"USE_DEFAULT\",\n" + + " \"familyMap\": {\n" + " \"[B@1a6e431\": [\n" + " {\n" + + " \"bytes\": [\n" + " 0,\n" + " 0,\n" + " 0,\n" + + " 24,\n" + " 0,\n" + " 0,\n" + " 0,\n" + + " 4,\n" + " 0,\n" + " 4,\n" + " 0,\n" + + " 0,\n" + " 0,\n" + " 123,\n" + " 4,\n" + + " 0,\n" + " 0,\n" + " 1,\n" + " -56,\n" + + " 0,\n" + " 0,\n" + " 3,\n" + " 21,\n" + + " 127,\n" + " -1,\n" + " -1,\n" + " -1,\n" + + " -1,\n" + " -1,\n" + " -1,\n" + " -1,\n" + + " 4,\n" + " 0,\n" + " 0,\n" + " 0,\n" + + " 0\n" + " ],\n" + " \"offset\": 0,\n" + + " \"length\": 36,\n" + " \"seqId\": \"0\"\n" + " }\n" + " ]\n" + + " },\n" + " \"priority\": -1\n" + " }\n" + "}"; + OnlineLogRecord o = new OnlineLogRecord(1, 2, 3, 4, null, null, null, null, null, null, null, 5, + 6, 7, Optional.empty(), Optional.empty(), Optional.empty(), Optional.of(mutate)); + Assert.assertEquals(o.toJsonPrettyPrint(), expectedOutput); + } +} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java index 26c105b40cb6..214b542c1453 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java @@ -1564,6 +1564,10 @@ public enum OperationStatusCode { // Default 10 mins. public static final int DEFAULT_SLOW_LOG_SYS_TABLE_CHORE_DURATION = 10 * 60 * 1000; + public static final String SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED = + "hbase.slowlog.operation.message.payload.enabled"; + public static final boolean SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED_DEFAULT = false; + public static final String SHELL_TIMESTAMP_FORMAT_EPOCH_KEY = "hbase.shell.timestamp.format.epoch"; diff --git a/hbase-protocol-shaded/src/main/protobuf/server/region/TooSlowLog.proto b/hbase-protocol-shaded/src/main/protobuf/server/region/TooSlowLog.proto index 36ed9d504a71..77fd7d23546d 100644 --- a/hbase-protocol-shaded/src/main/protobuf/server/region/TooSlowLog.proto +++ b/hbase-protocol-shaded/src/main/protobuf/server/region/TooSlowLog.proto @@ -27,6 +27,8 @@ option java_outer_classname = "TooSlowLog"; option java_generate_equals_and_hash = true; option optimize_for = SPEED; +import "client/Client.proto"; + message SlowLogPayload { required int64 start_time = 1; required int32 processing_time = 2; @@ -43,6 +45,10 @@ message SlowLogPayload { optional int32 multi_mutations = 13 [default = 0]; optional int32 multi_service_calls = 14 [default = 0]; required Type type = 15; + optional Scan scan = 16; + optional MultiRequest multi = 17; + optional Get get = 18; + optional MutationProto mutate = 19; // 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 diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/impl/SlowLogQueueService.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/impl/SlowLogQueueService.java index 86b24e9d975e..c80ebd4d4502 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/impl/SlowLogQueueService.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/impl/SlowLogQueueService.java @@ -65,10 +65,14 @@ public class SlowLogQueueService implements NamedQueueService { private final boolean isSlowLogTableEnabled; private final SlowLogPersistentService slowLogPersistentService; private final Queue slowLogQueue; + private final boolean slowLogOperationMessagePayloadEnabled; public SlowLogQueueService(Configuration conf) { this.isOnlineLogProviderEnabled = conf.getBoolean(HConstants.SLOW_LOG_BUFFER_ENABLED_KEY, HConstants.DEFAULT_ONLINE_LOG_PROVIDER_ENABLED); + this.slowLogOperationMessagePayloadEnabled = + conf.getBoolean(HConstants.SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED, + HConstants.SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED_DEFAULT); if (!isOnlineLogProviderEnabled) { this.isSlowLogTableEnabled = false; @@ -128,7 +132,8 @@ 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, slowLogOperationMessagePayloadEnabled); int numGets = 0; int numMutations = 0; int numServiceCalls = 0; @@ -151,7 +156,7 @@ public void consumeEventFromDisruptor(NamedQueuePayload namedQueuePayload) { final String userName = rpcCall.getRequestUserName().orElse(StringUtils.EMPTY); final String methodDescriptorName = methodDescriptor != null ? methodDescriptor.getName() : StringUtils.EMPTY; - TooSlowLog.SlowLogPayload slowLogPayload = TooSlowLog.SlowLogPayload.newBuilder() + TooSlowLog.SlowLogPayload.Builder slowLogPayloadBuilder = TooSlowLog.SlowLogPayload.newBuilder() .setCallDetails(methodDescriptorName + "(" + param.getClass().getName() + ")") .setClientAddress(clientAddress).setMethodName(methodDescriptorName).setMultiGets(numGets) .setMultiMutations(numMutations).setMultiServiceCalls(numServiceCalls) @@ -159,7 +164,22 @@ 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); + if (slowLogParams != null) { + if (slowLogParams.getScan() != null) { + slowLogPayloadBuilder.setScan(slowLogParams.getScan()); + } + if (slowLogParams.getMulti() != null) { + slowLogPayloadBuilder.setMulti(slowLogParams.getMulti()); + } + if (slowLogParams.getGet() != null) { + slowLogPayloadBuilder.setGet(slowLogParams.getGet()); + } + if (slowLogParams.getMutate() != null) { + slowLogPayloadBuilder.setMutate(slowLogParams.getMutate()); + } + } + TooSlowLog.SlowLogPayload slowLogPayload = slowLogPayloadBuilder.build(); slowLogQueue.add(slowLogPayload); if (isSlowLogTableEnabled) { if (!slowLogPayload.getRegionName().startsWith("hbase:slowlog")) { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestNamedQueueRecorder.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestNamedQueueRecorder.java index 406748fab630..89661416cace 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestNamedQueueRecorder.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestNamedQueueRecorder.java @@ -99,7 +99,7 @@ private boolean confirmPayloadParams(int i, int j, List slowLogP } @Test - public void testOnlieSlowLogConsumption() throws Exception { + public void testOnlineSlowLogConsumption() throws Exception { Configuration conf = applySlowLogRecorderConf(8); Constructor constructor = @@ -277,6 +277,118 @@ && confirmPayloadParams(12, 142, slowLogPayloads) Assert.assertEquals(slowLogPayloads.size(), 0); } + @Test + public void testOnlineSlowLogOperationMessagePayloadDefaultDisabled() throws Exception { + Configuration conf = applySlowLogRecorderConf(1); + conf.unset(HConstants.SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED); + Constructor constructor = + NamedQueueRecorder.class.getDeclaredConstructor(Configuration.class); + constructor.setAccessible(true); + namedQueueRecorder = constructor.newInstance(conf); + AdminProtos.SlowLogResponseRequest request = + AdminProtos.SlowLogResponseRequest.newBuilder().setLimit(1).build(); + + Assert.assertEquals(getSlowLogPayloads(request).size(), 0); + LOG.debug("Initially ringbuffer of Slow Log records is empty"); + RpcLogDetails rpcLogDetails = getRpcLogDetails("userName_1", "client_1", "class_1"); + namedQueueRecorder.addRecord(rpcLogDetails); + Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY.waitFor(3000, () -> { + Optional slowLogPayload = getSlowLogPayloads(request).stream().findAny(); + if (slowLogPayload.isPresent()) { + Message message = getMessage(slowLogPayload.get()); + return message == null; + } + return false; + })); + } + + @Test + public void testOnlineSlowLogOperationMessagePayloadExplicitlyDisabled() throws Exception { + Configuration conf = applySlowLogRecorderConf(1); + conf.setBoolean(HConstants.SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED, false); + Constructor constructor = + NamedQueueRecorder.class.getDeclaredConstructor(Configuration.class); + constructor.setAccessible(true); + namedQueueRecorder = constructor.newInstance(conf); + AdminProtos.SlowLogResponseRequest request = + AdminProtos.SlowLogResponseRequest.newBuilder().setLimit(1).build(); + + Assert.assertEquals(getSlowLogPayloads(request).size(), 0); + LOG.debug("Initially ringbuffer of Slow Log records is empty"); + RpcLogDetails rpcLogDetails = getRpcLogDetails("userName_1", "client_1", "class_1"); + namedQueueRecorder.addRecord(rpcLogDetails); + Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY.waitFor(3000, () -> { + Optional slowLogPayload = getSlowLogPayloads(request).stream().findAny(); + if (slowLogPayload.isPresent()) { + Message message = getMessage(slowLogPayload.get()); + return message == null; + } + return false; + })); + } + + @Test + public void testOnlineSlowLogOperationMessagePayloadEnabled() throws Exception { + i = 0; // set this so that the RpcCall and expected Message are deterministic + Configuration conf = applySlowLogRecorderConf(1); + conf.setBoolean(HConstants.SLOW_LOG_OPERATION_MESSAGE_PAYLOAD_ENABLED, true); + Constructor constructor = + NamedQueueRecorder.class.getDeclaredConstructor(Configuration.class); + constructor.setAccessible(true); + namedQueueRecorder = constructor.newInstance(conf); + AdminProtos.SlowLogResponseRequest request = + AdminProtos.SlowLogResponseRequest.newBuilder().setLimit(1).build(); + Message expectedMessage = + ClientProtos.MutationProto.newBuilder().setRow(ByteString.copyFromUtf8("row123")).build(); + + Assert.assertEquals(getSlowLogPayloads(request).size(), 0); + LOG.debug("Initially ringbuffer of Slow Log records is empty"); + RpcLogDetails rpcLogDetails = getRpcLogDetails("userName_1", "client_1", "class_1"); + namedQueueRecorder.addRecord(rpcLogDetails); + Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY.waitFor(3000, () -> { + Optional slowLogPayload = getSlowLogPayloads(request).stream().findAny(); + if (slowLogPayload.map(TestNamedQueueRecorder::hasMessage).orElse(false)) { + Message message = getMessage(slowLogPayload.get()); + return message.equals(expectedMessage); + } + return false; + })); + } + + private static boolean hasMessage(SlowLogPayload slowLogPayload) { + int trueCount = 0; + if (slowLogPayload.hasScan()) { + trueCount++; + } + if (slowLogPayload.hasMulti()) { + trueCount++; + } + if (slowLogPayload.hasGet()) { + trueCount++; + } + if (slowLogPayload.hasMutate()) { + trueCount++; + } + Assert.assertTrue(trueCount <= 1); + return trueCount > 0; + } + + private static Message getMessage(SlowLogPayload slowLogPayload) { + if (slowLogPayload.hasScan()) { + return slowLogPayload.getScan(); + } + if (slowLogPayload.hasMulti()) { + return slowLogPayload.getMulti(); + } + if (slowLogPayload.hasGet()) { + return slowLogPayload.getGet(); + } + if (slowLogPayload.hasMutate()) { + return slowLogPayload.getMutate(); + } + return null; + } + @Test public void testOnlineSlowLogWithDefaultDisableConfig() throws Exception { Configuration conf = HBASE_TESTING_UTILITY.getConfiguration();