diff --git a/fluss-common/src/main/java/org/apache/fluss/rpc/messages/ApiMessage.java b/fluss-common/src/main/java/org/apache/fluss/rpc/messages/ApiMessage.java index 55c000e89a1..c1c8870ac28 100644 --- a/fluss-common/src/main/java/org/apache/fluss/rpc/messages/ApiMessage.java +++ b/fluss-common/src/main/java/org/apache/fluss/rpc/messages/ApiMessage.java @@ -21,6 +21,8 @@ import org.apache.fluss.record.send.WritableOutput; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; +import java.util.Map; + /** * An object that can serialize itself. The serialization protocol is versioned. Messages also * implement toString, equals, and hashCode. @@ -28,6 +30,13 @@ @Internal public interface ApiMessage { + /** + * Accumulates this message's recursively set {@code error_code} values, including explicit zero + * values, into the supplied map without clearing its existing entries. Callers filter codes + * that are not errors. + */ + default void collectErrorCounts(Map errorCounts) {} + /** Gets the total serialized byte array size of the message. */ int totalSize(); diff --git a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ErrorCodeFieldFinder.java b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ErrorCodeFieldFinder.java new file mode 100644 index 00000000000..04eebf1b71c --- /dev/null +++ b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ErrorCodeFieldFinder.java @@ -0,0 +1,51 @@ +/* + * Copyright 2020 Splunk Inc. + * + * Licensed 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.fluss.protogen.generator.generator; + +import io.protostuff.parser.Field; +import io.protostuff.parser.Message; +import io.protostuff.parser.MessageField; + +import java.util.HashSet; +import java.util.Set; + +/** Finds protobuf messages with reachable error fields. */ +final class ErrorCodeFieldFinder { + + private ErrorCodeFieldFinder() {} + + static boolean hasErrorFields(Message message) { + return hasErrorFields(message, new HashSet<>()); + } + + private static boolean hasErrorFields(Message message, Set visited) { + if (!visited.add(message)) { + return false; + } + + if (ProtobufMessage.hasErrorFields(message)) { + return true; + } + for (Field field : message.getFields()) { + if (field instanceof MessageField + && hasErrorFields(((MessageField) field).getMessage(), visited)) { + return true; + } + } + return false; + } +} diff --git a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtoGenUtil.java b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtoGenUtil.java index 6058d1977a5..d26dfdcf684 100644 --- a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtoGenUtil.java +++ b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtoGenUtil.java @@ -29,6 +29,7 @@ import org.apache.fluss.utils.ProtoCodecUtils; import java.io.PrintWriter; +import java.util.Map; import static org.apache.fluss.shaded.guava32.com.google.common.base.CaseFormat.LOWER_CAMEL; import static org.apache.fluss.shaded.guava32.com.google.common.base.CaseFormat.LOWER_UNDERSCORE; @@ -67,6 +68,7 @@ public class ProtoGenUtil { ErrorMessage.class, FileRegionBytesView.class, MemorySegmentBytesView.class, + Map.class, ProtoCodecUtils.class, WritableOutput.class, // netty dependencies diff --git a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufMessage.java b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufMessage.java index 52a38053abf..980ad8cceb0 100644 --- a/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufMessage.java +++ b/fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufMessage.java @@ -20,6 +20,7 @@ import io.protostuff.parser.Field; import io.protostuff.parser.Message; +import io.protostuff.parser.MessageField; import java.io.PrintWriter; import java.util.ArrayList; @@ -39,6 +40,7 @@ public class ProtobufMessage { private final List fields; private final List nestedMessages; private final boolean hasErrorFields; + private final boolean hasReachableErrorFields; public ProtobufMessage(Message message, boolean isNested) { this.message = message; @@ -54,6 +56,7 @@ public ProtobufMessage(Message message, boolean isNested) { this.fields = new ArrayList<>(); this.hasErrorFields = hasErrorFields(message); + this.hasReachableErrorFields = ErrorCodeFieldFinder.hasErrorFields(message); for (int i = 0; i < message.getFields().size(); i++) { Field field = message.getFields().get(i); boolean isErrorField = @@ -96,6 +99,7 @@ public void generate(PrintWriter w) { generateZeroCopySerialize(w); generateTotalSize(w); generateZeroCopySize(w); + generateCollectErrorCounts(w); generateParseFrom(w); generateIsLazilyParsed(w); generateGetParsedByteBuf(w); @@ -125,6 +129,50 @@ public void generate(PrintWriter w) { w.println(); } + private void generateCollectErrorCounts(PrintWriter w) { + if (!hasReachableErrorFields) { + return; + } + + w.println(); + w.println(" @Override"); + w.println(" public void collectErrorCounts(Map errorCounts) {"); + if (hasErrorFields) { + w.println(" if (hasErrorCode()) {"); + w.println(" int errorCode = getErrorCode();"); + w.println(" Integer errorCount = errorCounts.get(errorCode);"); + w.println( + " errorCounts.put(errorCode, errorCount == null ? 1 : errorCount + 1);"); + w.println(" }"); + } + for (Field field : message.getFields()) { + if (!(field instanceof MessageField) + || !ErrorCodeFieldFinder.hasErrorFields(((MessageField) field).getMessage())) { + continue; + } + + String fieldName = ProtoGenUtil.camelCase(field.getName()); + if (field.isRepeated()) { + String pluralName = ProtoGenUtil.plural(fieldName); + String singularName = ProtoGenUtil.singular(fieldName); + w.format( + " for (int i = 0; i < %s(); i++) {\n", + ProtoGenUtil.camelCase("get", pluralName, "count")); + w.format( + " %s(i).collectErrorCounts(errorCounts);\n", + ProtoGenUtil.camelCase("get", singularName, "at")); + w.println(" }"); + } else { + w.format(" if (%s()) {\n", ProtoGenUtil.camelCase("has", fieldName)); + w.format( + " %s().collectErrorCounts(errorCounts);\n", + ProtoGenUtil.camelCase("get", field.getName())); + w.println(" }"); + } + } + w.println(" }"); + } + private void generateParseFrom(PrintWriter w) { w.println(); w.println(" @Override"); @@ -348,7 +396,7 @@ private boolean hasRequiredFields() { return false; } - private static boolean hasErrorFields(Message message) { + static boolean hasErrorFields(Message message) { boolean findErrorCode = false; boolean findErrorMessage = false; for (Field field : message.getFields()) { @@ -361,7 +409,7 @@ private static boolean hasErrorFields(Message message) { return findErrorCode && findErrorMessage; } - private static boolean isErrorCodeField(Field field) { + static boolean isErrorCodeField(Field field) { return field.getName().equals("error_code") && field.getJavaType().equals("int") && !field.isRepeated(); diff --git a/fluss-protogen/fluss-protogen-tests/src/main/proto/messages.proto b/fluss-protogen/fluss-protogen-tests/src/main/proto/messages.proto index df446f633d5..60c53e09d74 100644 --- a/fluss-protogen/fluss-protogen-tests/src/main/proto/messages.proto +++ b/fluss-protogen/fluss-protogen-tests/src/main/proto/messages.proto @@ -51,3 +51,13 @@ message E { optional NE nested_errors = 4; } +message ErrorContainer { + optional E optional_error = 1; + repeated E errors = 2; +} + +message RecursiveError { + optional int32 error_code = 1; + optional string error_message = 2; + optional RecursiveError child = 3; +} diff --git a/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/MessagesTest.java b/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/MessagesTest.java index a9704144e98..81e5c5f7f97 100644 --- a/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/MessagesTest.java +++ b/fluss-protogen/fluss-protogen-tests/src/test/java/org/apache/fluss/protogen/tests/MessagesTest.java @@ -29,7 +29,9 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -68,6 +70,32 @@ public void testErrorMessages() { .hasMessage("Field 'error_message' cannot be null"); } + @Test + public void testCollectErrorCounts() { + Map errorCounts = new HashMap<>(); + ErrorContainer errorContainer = new ErrorContainer(); + errorContainer.setOptionalError().setErrorCode(1); + errorContainer.addError().setErrorCode(1); + errorContainer.addError().setErrorCode(2); + + errorContainer.collectErrorCounts(errorCounts); + + assertThat(errorCounts).containsEntry(1, 2).containsEntry(2, 1); + + errorContainer.clear(); + errorContainer.addError().setErrorCode(3); + errorCounts.clear(); + errorContainer.collectErrorCounts(errorCounts); + + assertThat(errorCounts).containsOnlyKeys(3).containsEntry(3, 1); + + RecursiveError recursiveError = new RecursiveError().setErrorCode(4); + recursiveError.setChild().setErrorCode(5); + recursiveError.collectErrorCounts(errorCounts); + + assertThat(errorCounts).containsEntry(3, 1).containsEntry(4, 1).containsEntry(5, 1); + } + @Test public void testMessages() throws Exception { M lpm = new M(); diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java index ae6edf38cc4..ace786509bb 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java @@ -26,12 +26,14 @@ import org.apache.fluss.rpc.messages.AuthenticateRequest; import org.apache.fluss.rpc.messages.AuthenticateResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; +import org.apache.fluss.rpc.messages.ListOffsetsRequest; import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.protocol.ApiError; import org.apache.fluss.rpc.protocol.ApiKeys; import org.apache.fluss.rpc.protocol.ApiManager; import org.apache.fluss.rpc.protocol.ApiMethod; +import org.apache.fluss.rpc.protocol.Errors; import org.apache.fluss.rpc.protocol.MessageCodec; import org.apache.fluss.security.auth.ServerAuthenticator; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; @@ -51,6 +53,8 @@ import java.net.SocketAddress; import java.util.ArrayDeque; import java.util.Deque; +import java.util.HashMap; +import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -241,6 +245,7 @@ private void sendResponse(ChannelHandlerContext ctx, FlussRequest request) { CompletableFuture f = request.getResponseFuture(); sendSuccessResponse(ctx, request, f.get()); } catch (Throwable t) { + ExceptionUtils.rethrowIfFatalErrorOrOOM(t); sendError(ctx, request, t); } } else { @@ -265,8 +270,10 @@ private void sendSuccessResponse( request.getRequestCompletedTimeMs() - request.getRequestDequeTimeMs(), requestEndTimeMs - request.getRequestCompletedTimeMs(), request.getAddress()); - updateRequestMetrics(request, requestEndTimeMs); + tryUpdateRequestMetrics(request, requestEndTimeMs); + tryUpdateResponseErrorMetrics(request, responseMessage); } catch (Throwable t) { + ExceptionUtils.rethrowIfFatalErrorOrOOM(t); LOG.error("Failed to send response to client.", t); sendError(ctx, request, t); } @@ -279,7 +286,45 @@ private void sendError(ChannelHandlerContext ctx, FlussRequest request, Throwabl ByteBuf byteBuf = encodeErrorResponse(alloc, request.getRequestId(), error); ctx.writeAndFlush(byteBuf); - getMetrics(request).ifPresent(metrics -> metrics.markError(error.error())); + try { + getMetrics(request).ifPresent(metrics -> metrics.markError(error.error())); + } catch (Throwable metricFailure) { + logMetricsFailure("Failed to update request error metrics.", metricFailure); + } + } + + private void tryUpdateRequestMetrics(FlussRequest request, long requestEndTimeMs) { + try { + updateRequestMetrics(request, requestEndTimeMs); + } catch (Throwable t) { + logMetricsFailure("Failed to update request metrics.", t); + } + } + + private void tryUpdateResponseErrorMetrics(FlussRequest request, ApiMessage responseMessage) { + try { + Optional metrics = getMetrics(request); + if (!metrics.isPresent()) { + return; + } + + Map errorCounts = new HashMap<>(); + responseMessage.collectErrorCounts(errorCounts); + errorCounts.forEach( + (errorCode, count) -> { + Errors error = Errors.forCode(errorCode); + if (error != Errors.NONE) { + metrics.get().markErrorCount(error, count); + } + }); + } catch (Throwable t) { + logMetricsFailure("Failed to update response error metrics.", t); + } + } + + private void logMetricsFailure(String message, Throwable t) { + ExceptionUtils.rethrowIfFatalErrorOrOOM(t); + LOG.warn(message, t); } private void updateRequestMetrics(FlussRequest request, long requestEndTimeMs) { @@ -310,9 +355,13 @@ private Optional getMetrics(FlussRequest request) { boolean isHistorical = false; ApiMessage requestMessage = request.getMessage(); if (request.getApiKey() == ApiKeys.FETCH_LOG.id) { - // for fetch, we need to identify it's from client or follower + // For fetch, identify whether the request is from a client or follower. FetchLogRequest fetchLogRequest = (FetchLogRequest) requestMessage; isFromFollower = fetchLogRequest.getFollowerServerId() >= 0; + } else if (request.getApiKey() == ApiKeys.LIST_OFFSETS.id) { + // For list offsets, identify whether the request is from a client or follower. + ListOffsetsRequest listOffsetsRequest = (ListOffsetsRequest) requestMessage; + isFromFollower = listOffsetsRequest.getFollowerServerId() >= 0; } else if (request.getApiKey() == ApiKeys.LOOKUP.id) { isHistorical = hasHistoricalLookup((LookupRequest) requestMessage); } else if (request.getApiKey() == ApiKeys.PUT_KV.id) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java index 31c8b9ce3ad..73339d7209d 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java @@ -54,8 +54,9 @@ private RequestsMetrics(MetricGroup serverMetricsGroup, Collection apiK // we create a metrics group for each type of request, with the request type // as variable addMetrics(serverMetricsGroup, toRequestName(apiKey, false, false)); - if (apiKey == ApiKeys.FETCH_LOG) { - // For fetch, register separate metric groups for clients and followers. + if (apiKey == ApiKeys.FETCH_LOG || apiKey == ApiKeys.LIST_OFFSETS) { + // For fetch and list offsets, register separate metric groups for clients and + // followers. addMetrics(serverMetricsGroup, toRequestName(apiKey, true, false)); } if (apiKey == ApiKeys.LOOKUP) { @@ -76,7 +77,13 @@ public static RequestsMetrics createTabletServerRequestMetrics(MetricGroup serve ApiKeys.LOOKUP, ApiKeys.FETCH_LOG, ApiKeys.PREFIX_LOOKUP, - ApiKeys.GET_METADATA); + ApiKeys.GET_METADATA, + ApiKeys.LIST_OFFSETS, + ApiKeys.LIMIT_SCAN, + ApiKeys.SCAN_KV, + ApiKeys.GET_TABLE_STATS, + ApiKeys.NOTIFY_LEADER_AND_ISR, + ApiKeys.STOP_REPLICA); return new RequestsMetrics(serverMetricsGroup, apiKeys); } @@ -115,6 +122,18 @@ private static String toRequestName( return "prefixLookup"; case FETCH_LOG: return isFromFollower ? "fetchLogFollower" : "fetchLogClient"; + case LIST_OFFSETS: + return isFromFollower ? "listOffsetsFollower" : "listOffsetsClient"; + case LIMIT_SCAN: + return "limitScan"; + case SCAN_KV: + return "scanKv"; + case GET_TABLE_STATS: + return "tableStats"; + case NOTIFY_LEADER_AND_ISR: + return "notifyLeaderAndIsr"; + case STOP_REPLICA: + return "stopReplica"; case GET_METADATA: return "metadata"; default: @@ -182,7 +201,11 @@ public Counter getErrorsCount() { void markError(Errors error) { errorsCount.inc(); - errorsCountByError.computeIfAbsent(error, this::registerErrorMeter).inc(); + markErrorCount(error, 1); + } + + void markErrorCount(Errors error, int count) { + errorsCountByError.computeIfAbsent(error, this::registerErrorMeter).inc(count); } private Counter registerErrorMeter(Errors error) { diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerHandlerTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerHandlerTest.java index 833ae218540..b4eba690f98 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerHandlerTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/NettyServerHandlerTest.java @@ -20,7 +20,9 @@ import org.apache.fluss.cluster.ServerType; import org.apache.fluss.exception.TableNotExistException; import org.apache.fluss.metrics.Meter; +import org.apache.fluss.metrics.Metric; import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.AbstractMetricGroup; import org.apache.fluss.metrics.groups.GenericMetricGroup; import org.apache.fluss.metrics.groups.MetricGroup; import org.apache.fluss.metrics.util.NOPMetricsGroup; @@ -28,22 +30,42 @@ import org.apache.fluss.record.MemoryLogRecords; import org.apache.fluss.record.bytesview.MemorySegmentBytesView; import org.apache.fluss.rpc.TestingTabletGatewayService; +import org.apache.fluss.rpc.messages.ApiMessage; import org.apache.fluss.rpc.messages.ApiVersionsRequest; import org.apache.fluss.rpc.messages.ApiVersionsResponse; +import org.apache.fluss.rpc.messages.FetchLogRequest; +import org.apache.fluss.rpc.messages.FetchLogResponse; +import org.apache.fluss.rpc.messages.GetTableStatsRequest; +import org.apache.fluss.rpc.messages.GetTableStatsResponse; +import org.apache.fluss.rpc.messages.LimitScanRequest; +import org.apache.fluss.rpc.messages.LimitScanResponse; +import org.apache.fluss.rpc.messages.ListOffsetsRequest; +import org.apache.fluss.rpc.messages.ListOffsetsResponse; import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.LookupResponse; +import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest; +import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse; import org.apache.fluss.rpc.messages.PbApiVersion; +import org.apache.fluss.rpc.messages.PbFetchLogRespForTable; import org.apache.fluss.rpc.messages.PbLookupReqForBucket; import org.apache.fluss.rpc.messages.PbLookupRespForBucket; +import org.apache.fluss.rpc.messages.PbNotifyLeaderAndIsrRespForBucket; import org.apache.fluss.rpc.messages.PbProduceLogReqForBucket; +import org.apache.fluss.rpc.messages.PbStopReplicaRespForBucket; import org.apache.fluss.rpc.messages.PbValue; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.ProduceLogResponse; +import org.apache.fluss.rpc.messages.PutKvResponse; +import org.apache.fluss.rpc.messages.ScanKvRequest; +import org.apache.fluss.rpc.messages.ScanKvResponse; +import org.apache.fluss.rpc.messages.StopReplicaRequest; +import org.apache.fluss.rpc.messages.StopReplicaResponse; import org.apache.fluss.rpc.protocol.ApiKeys; import org.apache.fluss.rpc.protocol.ApiManager; import org.apache.fluss.rpc.protocol.Errors; import org.apache.fluss.rpc.protocol.MessageCodec; import org.apache.fluss.rpc.protocol.RequestType; +import org.apache.fluss.rpc.protocol.ResponseType; import org.apache.fluss.security.auth.PlainTextAuthenticationPlugin; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator; @@ -52,12 +74,13 @@ import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelId; import org.apache.fluss.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel; import org.apache.fluss.shaded.netty4.io.netty.util.concurrent.DefaultEventExecutor; -import org.apache.fluss.shaded.netty4.io.netty.util.concurrent.ImmediateEventExecutor; -import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -65,7 +88,9 @@ import java.time.Duration; import java.util.Collections; import java.util.Deque; +import java.util.HashMap; import java.util.HashSet; +import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; @@ -74,6 +99,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; import static org.apache.fluss.record.TestData.DATA1; import static org.apache.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject; @@ -85,24 +111,41 @@ /** Test for {@link NettyServerHandler}. */ final class NettyServerHandlerTest { + private static final InetSocketAddress REMOTE_ADDRESS = + new InetSocketAddress("127.0.0.1", 9092); + private NettyServerHandler serverHandler; private TestingRequestChannel requestChannel; private ChannelHandlerContext ctx; - @BeforeEach - void beforeEach() throws Exception { - this.requestChannel = new TestingRequestChannel(100); - MetricGroup metricGroup = NOPMetricsGroup.newInstance(); - this.serverHandler = - new NettyServerHandler( - requestChannel, - new ApiManager(ServerType.TABLET_SERVER), - "FLUSS", - true, - RequestsMetrics.createCoordinatorServerRequestMetrics(metricGroup), - new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); - this.ctx = mockChannelHandlerContext(); - serverHandler.channelActive(ctx); + @Test + void testFailedRequestMarksAggregateAndErrorMetrics() { + RequestsMetricsTest.RecordingMetricRegistry metricRegistry = + new RequestsMetricsTest.RecordingMetricRegistry(); + HandlerFixture fixture = createTabletServerHandler(metricRegistry); + try { + writeRequest(fixture, ApiKeys.LOOKUP, new LookupRequest().setTableId(1), 1001); + FlussRequest request = (FlussRequest) fixture.requestChannel.getAndRemoveRequest(0); + request.fail(new TableNotExistException("table does not exist")); + fixture.channel.runPendingTasks(); + + assertResponse(fixture.channel, ResponseType.ERROR_RESPONSE, 1001); + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) + .hasSize(2) + .allSatisfy( + registered -> + assertThat(((Meter) registered.metric).getCount()).isEqualTo(1)) + .anySatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .doesNotContainKey("error")) + .anySatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.TABLE_NOT_EXIST.name())); + } finally { + fixture.channel.finishAndReleaseAll(); + } } @Test @@ -240,57 +283,402 @@ public CompletableFuture produceLog( } @Test - void testFailedRequestMarksAggregateAndErrorMetrics() throws Exception { + void testSuccessfulResponseMarksBucketErrorsWithoutAggregate() { + RequestsMetricsTest.RecordingMetricRegistry metricRegistry = + new RequestsMetricsTest.RecordingMetricRegistry(); + HandlerFixture fixture = createTabletServerHandler(metricRegistry); + try { + writeRequest(fixture, ApiKeys.LOOKUP, new LookupRequest().setTableId(1), 1001); + + LookupResponse response = new LookupResponse(); + response.addBucketsResp().setBucketId(1).setErrorCode(Errors.TABLE_NOT_EXIST.code()); + response.addBucketsResp().setBucketId(2).setErrorCode(Errors.TABLE_NOT_EXIST.code()); + response.addBucketsResp().setBucketId(3).setErrorCode(Integer.MAX_VALUE); + response.addBucketsResp().setBucketId(4); + response.addBucketsResp().setBucketId(5).setErrorCode(Errors.NONE.code()); + completeRequest(fixture, response); + + assertResponse(fixture.channel, ResponseType.SUCCESS_RESPONSE, 1001); + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) + .hasSize(3) + .anySatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .doesNotContainKey("error") + .isEqualTo( + Collections.singletonMap("request", "lookup"))); + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) + .filteredOn( + registered -> registered.group.getAllVariables().containsKey("error")) + .anySatisfy( + registered -> { + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.TABLE_NOT_EXIST.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(2); + }) + .anySatisfy( + registered -> { + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.UNKNOWN_SERVER_ERROR.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(1); + }); + assertAggregateErrorsAreZero(metricRegistry, "lookup"); + } finally { + fixture.channel.finishAndReleaseAll(); + } + } + + @Test + void testEncodingFailureMarksOnlyWholeRpcError() { + RequestsMetricsTest.RecordingMetricRegistry metricRegistry = + new RequestsMetricsTest.RecordingMetricRegistry(); + HandlerFixture fixture = createTabletServerHandler(metricRegistry); + try { + writeRequest(fixture, ApiKeys.LOOKUP, new LookupRequest().setTableId(1), 1001); + + LookupResponse response = new LookupResponse(); + response.addBucketsResp().setErrorCode(Errors.TABLE_NOT_EXIST.code()); + completeRequest(fixture, response); + + assertResponse(fixture.channel, ResponseType.ERROR_RESPONSE, 1001); + assertThat(fixture.channel.outboundMessages()).isEmpty(); + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) + .hasSize(2) + .allSatisfy( + registered -> + assertThat(((Meter) registered.metric).getCount()).isEqualTo(1)) + .anySatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .doesNotContainKey("error")) + .anySatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .containsEntry( + "error", Errors.UNKNOWN_SERVER_ERROR.name())); + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) + .filteredOn( + registered -> registered.group.getAllVariables().containsKey("error")) + .noneSatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.TABLE_NOT_EXIST.name())); + } finally { + fixture.channel.finishAndReleaseAll(); + } + } + + @Test + void testBatchResponsesCollectErrorCounts() { + Map errorCounts = new HashMap<>(); + + ProduceLogResponse produceLogResponse = new ProduceLogResponse(); + produceLogResponse.addBucketsResp().setBucketId(1).setErrorCode(1); + PutKvResponse putKvResponse = new PutKvResponse(); + putKvResponse.addBucketsResp().setBucketId(1).setErrorCode(2); + LookupResponse lookupResponse = new LookupResponse(); + lookupResponse.addBucketsResp().setBucketId(1).setErrorCode(3); + FetchLogResponse fetchLogResponse = new FetchLogResponse(); + PbFetchLogRespForTable tableResponse = fetchLogResponse.addTablesResp().setTableId(1); + tableResponse.addBucketsResp().setBucketId(1).setErrorCode(4); + + produceLogResponse.collectErrorCounts(errorCounts); + putKvResponse.collectErrorCounts(errorCounts); + lookupResponse.collectErrorCounts(errorCounts); + fetchLogResponse.collectErrorCounts(errorCounts); + + assertThat(errorCounts) + .containsEntry(1, 1) + .containsEntry(2, 1) + .containsEntry(3, 1) + .containsEntry(4, 1); + } + + @Test + void testResponseMetricsFailureDoesNotSendErrorResponse() { + MetricGroup metricGroup = + new GenericMetricGroup( + new ErrorMetricRegistrationFailureRegistry(), null, "tabletserver"); + HandlerFixture fixture = createTabletServerHandler(metricGroup); + try { + writeRequest(fixture, ApiKeys.LOOKUP, new LookupRequest().setTableId(1), 1001); + + LookupResponse response = new LookupResponse(); + response.addBucketsResp().setBucketId(1).setErrorCode(Errors.TABLE_NOT_EXIST.code()); + completeRequest(fixture, response); + + assertResponse(fixture.channel, ResponseType.SUCCESS_RESPONSE, 1001); + assertThat(fixture.channel.outboundMessages()).isEmpty(); + } finally { + fixture.channel.finishAndReleaseAll(); + } + } + + private static class ErrorMetricRegistrationFailureRegistry + extends RequestsMetricsTest.RecordingMetricRegistry { + + @Override + public void register(Metric metric, String metricName, AbstractMetricGroup group) { + if (group.getAllVariables().containsKey("error")) { + throw new Error("metric registration failed"); + } + super.register(metric, metricName, group); + } + } + + @ParameterizedTest + @MethodSource("successfulResponsesWithErrors") + void successfulResponsesWithErrorsOnlyMarkPerErrorMetric( + ApiKeys apiKey, String requestName, ApiMessage request, ApiMessage response) { RequestsMetricsTest.RecordingMetricRegistry metricRegistry = new RequestsMetricsTest.RecordingMetricRegistry(); - MetricGroup metricGroup = new GenericMetricGroup(metricRegistry, null, "tabletserver"); - TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); - NettyServerHandler tabletServerHandler = + HandlerFixture fixture = createTabletServerHandler(metricRegistry); + try { + writeRequest(fixture, apiKey, request, 1001); + completeRequest(fixture, response); + + assertResponse(fixture.channel, ResponseType.SUCCESS_RESPONSE, 1001); + assertPerErrorMetric(metricRegistry, requestName); + assertAggregateErrorsAreZero(metricRegistry, requestName); + } finally { + fixture.channel.finishAndReleaseAll(); + } + } + + @ParameterizedTest + @MethodSource("requestsByOrigin") + void fetchLogAndListOffsetsMetricsAreIsolatedByRequestOrigin( + ApiKeys apiKey, int followerServerId, String requestName, String otherRequestName) { + RequestsMetricsTest.RecordingMetricRegistry metricRegistry = + new RequestsMetricsTest.RecordingMetricRegistry(); + HandlerFixture fixture = createTabletServerHandler(metricRegistry); + try { + writeRequest(fixture, apiKey, requestWithOrigin(apiKey, followerServerId), 1001); + completeRequest(fixture, responseWithError(apiKey)); + + assertResponse(fixture.channel, ResponseType.SUCCESS_RESPONSE, 1001); + assertRequestMetricCount(metricRegistry, requestName, 1); + assertRequestMetricCount(metricRegistry, otherRequestName, 0); + assertPerErrorMetric(metricRegistry, requestName); + assertNoPerErrorMetric(metricRegistry, otherRequestName); + assertAggregateErrorsAreZero(metricRegistry, requestName); + assertAggregateErrorsAreZero(metricRegistry, otherRequestName); + } finally { + fixture.channel.finishAndReleaseAll(); + } + } + + private static Stream successfulResponsesWithErrors() { + return Stream.of( + Arguments.of( + ApiKeys.LIST_OFFSETS, + "listOffsetsClient", + listOffsetsRequest(-1), + listOffsetsResponse()), + Arguments.of( + ApiKeys.LIMIT_SCAN, + "limitScan", + new LimitScanRequest().setTableId(1).setBucketId(1).setLimit(1), + new LimitScanResponse().setErrorCode(Errors.TABLE_NOT_EXIST.code())), + Arguments.of( + ApiKeys.SCAN_KV, + "scanKv", + new ScanKvRequest(), + new ScanKvResponse().setErrorCode(Errors.TABLE_NOT_EXIST.code())), + Arguments.of( + ApiKeys.GET_TABLE_STATS, + "tableStats", + new GetTableStatsRequest().setTableId(1), + tableStatsResponse()), + Arguments.of( + ApiKeys.NOTIFY_LEADER_AND_ISR, + "notifyLeaderAndIsr", + new NotifyLeaderAndIsrRequest().setCoordinatorEpoch(1), + notifyLeaderAndIsrResponse()), + Arguments.of( + ApiKeys.STOP_REPLICA, + "stopReplica", + new StopReplicaRequest().setCoordinatorEpoch(1), + stopReplicaResponse())); + } + + private static Stream requestsByOrigin() { + return Stream.of( + Arguments.of(ApiKeys.FETCH_LOG, -1, "fetchLogClient", "fetchLogFollower"), + Arguments.of(ApiKeys.FETCH_LOG, 0, "fetchLogFollower", "fetchLogClient"), + Arguments.of(ApiKeys.LIST_OFFSETS, -1, "listOffsetsClient", "listOffsetsFollower"), + Arguments.of(ApiKeys.LIST_OFFSETS, 0, "listOffsetsFollower", "listOffsetsClient")); + } + + private static ApiMessage requestWithOrigin(ApiKeys apiKey, int followerServerId) { + if (apiKey == ApiKeys.FETCH_LOG) { + return new FetchLogRequest().setFollowerServerId(followerServerId).setMaxBytes(1); + } + return listOffsetsRequest(followerServerId); + } + + private static ApiMessage responseWithError(ApiKeys apiKey) { + if (apiKey == ApiKeys.FETCH_LOG) { + FetchLogResponse response = new FetchLogResponse(); + PbFetchLogRespForTable tableResponse = response.addTablesResp().setTableId(1); + tableResponse + .addBucketsResp() + .setBucketId(1) + .setErrorCode(Errors.TABLE_NOT_EXIST.code()); + return response; + } + return listOffsetsResponse(); + } + + private static ListOffsetsRequest listOffsetsRequest(int followerServerId) { + return new ListOffsetsRequest() + .setFollowerServerId(followerServerId) + .setOffsetType(0) + .setTableId(1); + } + + private static ListOffsetsResponse listOffsetsResponse() { + ListOffsetsResponse response = new ListOffsetsResponse(); + response.addBucketsResp().setBucketId(1).setErrorCode(Errors.TABLE_NOT_EXIST.code()); + return response; + } + + private static GetTableStatsResponse tableStatsResponse() { + GetTableStatsResponse response = new GetTableStatsResponse(); + response.addBucketsResp().setBucketId(1).setErrorCode(Errors.TABLE_NOT_EXIST.code()); + return response; + } + + private static NotifyLeaderAndIsrResponse notifyLeaderAndIsrResponse() { + NotifyLeaderAndIsrResponse response = new NotifyLeaderAndIsrResponse(); + PbNotifyLeaderAndIsrRespForBucket bucketResponse = response.addNotifyBucketsLeaderResp(); + bucketResponse.setTableBucket().setTableId(1).setBucketId(1); + bucketResponse.setErrorCode(Errors.TABLE_NOT_EXIST.code()); + return response; + } + + private static StopReplicaResponse stopReplicaResponse() { + StopReplicaResponse response = new StopReplicaResponse(); + PbStopReplicaRespForBucket bucketResponse = response.addStopReplicasResp(); + bucketResponse.setTableBucket().setTableId(1).setBucketId(1); + bucketResponse.setErrorCode(Errors.TABLE_NOT_EXIST.code()); + return response; + } + + private static HandlerFixture createTabletServerHandler( + RequestsMetricsTest.RecordingMetricRegistry metricRegistry) { + return createTabletServerHandler( + new GenericMetricGroup(metricRegistry, null, "tabletserver")); + } + + private static HandlerFixture createTabletServerHandler(MetricGroup metricGroup) { + TestingRequestChannel requestChannel = new TestingRequestChannel(100); + NettyServerHandler handler = new NettyServerHandler( - tabletRequestChannel, + requestChannel, new ApiManager(ServerType.TABLET_SERVER), "FLUSS", true, RequestsMetrics.createTabletServerRequestMetrics(metricGroup), new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); - ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); - tabletServerHandler.channelActive(tabletContext); + EmbeddedChannel channel = + new EmbeddedChannel(handler) { + @Override + public SocketAddress remoteAddress() { + return REMOTE_ADDRESS; + } + }; + return new HandlerFixture(channel, requestChannel); + } - LookupRequest lookupRequest = new LookupRequest().setTableId(1); - PbLookupReqForBucket bucketRequest = - new PbLookupReqForBucket().setPartitionId(1).setBucketId(1); - bucketRequest.addKey("key".getBytes()); - lookupRequest.addAllBucketsReqs(Collections.singleton(bucketRequest)); - ByteBuf byteBuf = + private static void writeRequest( + HandlerFixture fixture, ApiKeys apiKey, ApiMessage request, int requestId) { + fixture.channel.writeInbound( MessageCodec.encodeRequest( ByteBufAllocator.DEFAULT, - ApiKeys.LOOKUP.id, - ApiKeys.LOOKUP.highestSupportedVersion, - 1001, - lookupRequest); - - tabletServerHandler.channelRead(tabletContext, byteBuf); - FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); - request.fail(new TableNotExistException("table does not exist")); - - assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) - .hasSize(2) - .allSatisfy( - registered -> - assertThat(((Meter) registered.metric).getCount()).isEqualTo(1)) - .anySatisfy( - registered -> - assertThat(registered.group.getAllVariables()) - .doesNotContainKey("error")) - .anySatisfy( + apiKey.id, + apiKey.highestSupportedVersion, + requestId, + request)); + } + + private static void completeRequest(HandlerFixture fixture, ApiMessage response) { + FlussRequest request = (FlussRequest) fixture.requestChannel.getAndRemoveRequest(0); + request.complete(response); + fixture.channel.runPendingTasks(); + } + + private static void assertResponse( + EmbeddedChannel channel, ResponseType expectedResponseType, int expectedRequestId) { + ByteBuf response = channel.readOutbound(); + try { + assertThat(response).isNotNull(); + int frameLength = response.readInt(); + assertThat(frameLength).isEqualTo(response.readableBytes()); + assertThat(ResponseType.forId(response.readByte())).isEqualTo(expectedResponseType); + assertThat(response.readInt()).isEqualTo(expectedRequestId); + } finally { + if (response != null) { + response.release(); + } + } + } + + private static void assertPerErrorMetric( + RequestsMetricsTest.RecordingMetricRegistry metricRegistry, String requestName) { + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, requestName)) + .filteredOn(registered -> registered.group.getAllVariables().containsKey("error")) + .singleElement() + .satisfies( + registered -> { + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.TABLE_NOT_EXIST.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(1); + }); + } + + private static void assertAggregateErrorsAreZero( + RequestsMetricsTest.RecordingMetricRegistry metricRegistry, String requestName) { + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, requestName)) + .filteredOn(registered -> !registered.group.getAllVariables().containsKey("error")) + .singleElement() + .satisfies( + registered -> assertThat(((Meter) registered.metric).getCount()).isZero()); + } + + private static void assertNoPerErrorMetric( + RequestsMetricsTest.RecordingMetricRegistry metricRegistry, String requestName) { + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, requestName)) + .filteredOn(registered -> registered.group.getAllVariables().containsKey("error")) + .isEmpty(); + } + + private static void assertRequestMetricCount( + RequestsMetricsTest.RecordingMetricRegistry metricRegistry, + String requestName, + int expectedCount) { + assertThat(metricRegistry.metrics(MetricNames.REQUESTS_RATE, requestName)) + .singleElement() + .satisfies( registered -> - assertThat(registered.group.getAllVariables()) - .containsEntry("error", Errors.TABLE_NOT_EXIST.name())); + assertThat(((Meter) registered.metric).getCount()) + .isEqualTo(expectedCount)); + } + + private static final class HandlerFixture { + private final EmbeddedChannel channel; + private final TestingRequestChannel requestChannel; + + private HandlerFixture(EmbeddedChannel channel, TestingRequestChannel requestChannel) { + this.channel = channel; + this.requestChannel = requestChannel; + } } @Test @Disabled("TODO: add back in https://github.com/apache/fluss/issues/771") void testResponseReturnInOrder() throws Exception { + initializeLegacyFixture(); // first write 10 requests to serverHandler. for (int i = 0; i < 10; i++) { ApiVersionsRequest request = new ApiVersionsRequest(); @@ -362,6 +750,7 @@ void testResponseReturnInOrder() throws Exception { @Test @Disabled("TODO: add back in https://github.com/apache/fluss/issues/771") void testDifferentResponseTypeReturnInSeparateOrder() throws Exception { + initializeLegacyFixture(); // 1. first write 5 requests with api as ApiKeys.API_VERSIONS to serverHandler. for (int i = 0; i < 5; i++) { ApiVersionsRequest request = new ApiVersionsRequest(); @@ -449,6 +838,21 @@ private static ByteBuf encodeProduceLogRequest() throws Exception { request); } + private void initializeLegacyFixture() throws Exception { + requestChannel = new TestingRequestChannel(100); + MetricGroup metricGroup = NOPMetricsGroup.newInstance(); + serverHandler = + new NettyServerHandler( + requestChannel, + new ApiManager(ServerType.TABLET_SERVER), + "FLUSS", + true, + RequestsMetrics.createCoordinatorServerRequestMetrics(metricGroup), + new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); + ctx = mockChannelHandlerContext(); + serverHandler.channelActive(ctx); + } + private static ChannelHandlerContext mockChannelHandlerContext() { ChannelId channelId = mock(ChannelId.class); when(channelId.asShortText()).thenReturn("short_text"); @@ -462,13 +866,6 @@ private static ChannelHandlerContext mockChannelHandlerContext() { return ctx; } - private static ChannelHandlerContext mockImmediateChannelHandlerContext() { - ChannelHandlerContext ctx = mockChannelHandlerContext(); - when(ctx.channel().remoteAddress()).thenReturn(new InetSocketAddress("127.0.0.1", 9092)); - when(ctx.executor()).thenReturn(ImmediateEventExecutor.INSTANCE); - return ctx; - } - private ApiVersionsResponse makeApiVersionResponse() { ApiVersionsResponse response = new ApiVersionsResponse(); PbApiVersion apiVersion = new PbApiVersion(); diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/RequestsMetricsTest.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/RequestsMetricsTest.java index b6f6e735726..7ab96c5289b 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/RequestsMetricsTest.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/RequestsMetricsTest.java @@ -106,6 +106,24 @@ void errorsKeepAggregateAndRegisterBreakdownByError() { }); } + @Test + void registersMetricsForAdditionalTabletServerRequests() { + RecordingMetricRegistry registry = new RecordingMetricRegistry(); + MetricGroup serverMetricGroup = new GenericMetricGroup(registry, null, "tabletserver"); + RequestsMetrics requestsMetrics = + RequestsMetrics.createTabletServerRequestMetrics(serverMetricGroup); + + assertThat(requestsMetrics.getMetrics(ApiKeys.LIST_OFFSETS.id, false, false)).isPresent(); + assertThat(requestsMetrics.getMetrics(ApiKeys.LIST_OFFSETS.id, true, false)).isPresent(); + assertThat(requestsMetrics.getMetrics(ApiKeys.LIMIT_SCAN.id, false, false)).isPresent(); + assertThat(requestsMetrics.getMetrics(ApiKeys.SCAN_KV.id, false, false)).isPresent(); + assertThat(requestsMetrics.getMetrics(ApiKeys.GET_TABLE_STATS.id, false, false)) + .isPresent(); + assertThat(requestsMetrics.getMetrics(ApiKeys.NOTIFY_LEADER_AND_ISR.id, false, false)) + .isPresent(); + assertThat(requestsMetrics.getMetrics(ApiKeys.STOP_REPLICA.id, false, false)).isPresent(); + } + static class RecordingMetricRegistry implements MetricRegistry { private final List registeredMetrics = new ArrayList<>(); diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index ee57382f523..d3f8d951fbc 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/website/docs/maintenance/observability/monitor-metrics.md @@ -716,6 +716,13 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM request_metadata request_fetchLogClient request_fetchLogFollower + request_listOffsetsClient + request_listOffsetsFollower + request_limitScan + request_scanKv + request_tableStats + request_notifyLeaderAndIsr + request_stopReplica requestsPerSecond The total number of requests processed per second for each request type. @@ -723,7 +730,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM errorsPerSecond - The total number of error requests processed per second for each request type. + The total number of failed RPC responses processed per second for each request type. This original aggregate excludes errors carried by successful response buckets or scan responses. Meter @@ -754,7 +761,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM request_error errorsPerSecond - The number of failed RPC responses processed per second for each request type and error name. One event is recorded for each failed RPC response; NONE and errors in successful response buckets are excluded. A series appears only after its request/error first occurs. + The number of errors processed per second for each request type and error name. It includes failed RPC responses and non-NONE error codes in successful response buckets or scan responses. Because one successful response can contain multiple bucket or scan errors, this metric does not necessarily sum to the request type's original aggregate errorsPerSecond. A series appears only after its request/error first occurs. Meter