From 1daf6af4925c934a1304d42b817d2c6465fa87e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Mon, 14 Sep 2026 21:27:36 +0800 Subject: [PATCH 1/2] [metrics] Count errors in successful TabletServer responses Generate recursive raw error-code collection for protobuf messages. Count bucket failures without changing aggregate RPC error semantics. Cover response errors, message reuse, and encoding and metrics failures. --- .../apache/fluss/rpc/messages/ApiMessage.java | 9 + .../generator/ErrorCodeFieldFinder.java | 51 +++++ .../generator/generator/ProtoGenUtil.java | 2 + .../generator/generator/ProtobufMessage.java | 52 ++++- .../src/main/proto/messages.proto | 10 + .../fluss/protogen/tests/MessagesTest.java | 28 +++ .../rpc/netty/server/NettyServerHandler.java | 48 +++- .../rpc/netty/server/RequestsMetrics.java | 6 +- .../netty/server/NettyServerHandlerTest.java | 214 ++++++++++++++++++ .../observability/monitor-metrics.md | 4 +- 10 files changed, 417 insertions(+), 7 deletions(-) create mode 100644 fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ErrorCodeFieldFinder.java 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..9817da1a035 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 @@ -32,6 +32,7 @@ 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 +52,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 +244,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 +269,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 +285,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) { 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..f8860567555 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 @@ -182,7 +182,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..1a81a4049a7 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; @@ -30,15 +32,18 @@ import org.apache.fluss.rpc.TestingTabletGatewayService; import org.apache.fluss.rpc.messages.ApiVersionsRequest; import org.apache.fluss.rpc.messages.ApiVersionsResponse; +import org.apache.fluss.rpc.messages.FetchLogResponse; import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.LookupResponse; 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.PbProduceLogReqForBucket; 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.protocol.ApiKeys; import org.apache.fluss.rpc.protocol.ApiManager; import org.apache.fluss.rpc.protocol.Errors; @@ -65,7 +70,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; @@ -79,7 +86,10 @@ import static org.apache.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** Test for {@link NettyServerHandler}. */ @@ -288,6 +298,210 @@ void testFailedRequestMarksAggregateAndErrorMetrics() throws Exception { .containsEntry("error", Errors.TABLE_NOT_EXIST.name())); } + @Test + void testSuccessfulResponseMarksBucketErrorsWithoutAggregate() throws Exception { + RequestsMetricsTest.RecordingMetricRegistry metricRegistry = + new RequestsMetricsTest.RecordingMetricRegistry(); + MetricGroup metricGroup = new GenericMetricGroup(metricRegistry, null, "tabletserver"); + TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); + NettyServerHandler tabletServerHandler = + new NettyServerHandler( + tabletRequestChannel, + new ApiManager(ServerType.TABLET_SERVER), + "FLUSS", + true, + RequestsMetrics.createTabletServerRequestMetrics(metricGroup), + new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); + ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); + tabletServerHandler.channelActive(tabletContext); + + LookupRequest lookupRequest = new LookupRequest().setTableId(1); + PbLookupReqForBucket bucketRequest = + new PbLookupReqForBucket().setPartitionId(1).setBucketId(1); + bucketRequest.addKey("key".getBytes()); + lookupRequest.addAllBucketsReqs(Collections.singleton(bucketRequest)); + tabletServerHandler.channelRead( + tabletContext, + MessageCodec.encodeRequest( + ByteBufAllocator.DEFAULT, + ApiKeys.LOOKUP.id, + ApiKeys.LOOKUP.highestSupportedVersion, + 1001, + lookupRequest)); + + 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()); + FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); + request.complete(response); + + 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); + }); + assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) + .filteredOn(registered -> !registered.group.getAllVariables().containsKey("error")) + .allSatisfy( + registered -> assertThat(((Meter) registered.metric).getCount()).isZero()); + } + + @Test + void testEncodingFailureMarksOnlyWholeRpcError() throws Exception { + RequestsMetricsTest.RecordingMetricRegistry metricRegistry = + new RequestsMetricsTest.RecordingMetricRegistry(); + MetricGroup metricGroup = new GenericMetricGroup(metricRegistry, null, "tabletserver"); + TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); + NettyServerHandler tabletServerHandler = + new NettyServerHandler( + tabletRequestChannel, + new ApiManager(ServerType.TABLET_SERVER), + "FLUSS", + true, + RequestsMetrics.createTabletServerRequestMetrics(metricGroup), + new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); + ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); + tabletServerHandler.channelActive(tabletContext); + + LookupRequest lookupRequest = new LookupRequest().setTableId(1); + PbLookupReqForBucket bucketRequest = + new PbLookupReqForBucket().setPartitionId(1).setBucketId(1); + bucketRequest.addKey("key".getBytes()); + lookupRequest.addAllBucketsReqs(Collections.singleton(bucketRequest)); + tabletServerHandler.channelRead( + tabletContext, + MessageCodec.encodeRequest( + ByteBufAllocator.DEFAULT, + ApiKeys.LOOKUP.id, + ApiKeys.LOOKUP.highestSupportedVersion, + 1001, + lookupRequest)); + + LookupResponse response = new LookupResponse(); + response.addBucketsResp().setErrorCode(Errors.TABLE_NOT_EXIST.code()); + FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); + request.complete(response); + + verify(tabletContext).writeAndFlush(any(ByteBuf.class)); + 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())); + } + + @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() throws Exception { + MetricGroup metricGroup = + new GenericMetricGroup( + new ErrorMetricRegistrationFailureRegistry(), null, "tabletserver"); + TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); + NettyServerHandler tabletServerHandler = + new NettyServerHandler( + tabletRequestChannel, + new ApiManager(ServerType.TABLET_SERVER), + "FLUSS", + true, + RequestsMetrics.createTabletServerRequestMetrics(metricGroup), + new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); + ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); + tabletServerHandler.channelActive(tabletContext); + + LookupRequest lookupRequest = new LookupRequest().setTableId(1); + PbLookupReqForBucket bucketRequest = + new PbLookupReqForBucket().setPartitionId(1).setBucketId(1); + bucketRequest.addKey("key".getBytes()); + lookupRequest.addAllBucketsReqs(Collections.singleton(bucketRequest)); + tabletServerHandler.channelRead( + tabletContext, + MessageCodec.encodeRequest( + ByteBufAllocator.DEFAULT, + ApiKeys.LOOKUP.id, + ApiKeys.LOOKUP.highestSupportedVersion, + 1001, + lookupRequest)); + + LookupResponse response = new LookupResponse(); + response.addBucketsResp().setBucketId(1).setErrorCode(Errors.TABLE_NOT_EXIST.code()); + FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); + request.complete(response); + + verify(tabletContext).flush(); + verify(tabletContext, never()).writeAndFlush(any(ByteBuf.class)); + } + + 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); + } + } + @Test @Disabled("TODO: add back in https://github.com/apache/fluss/issues/771") void testResponseReturnInOrder() throws Exception { diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index ee57382f523..b87c74d8939 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/website/docs/maintenance/observability/monitor-metrics.md @@ -723,7 +723,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. Errors carried by successful response buckets are excluded. Meter @@ -754,7 +754,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. Because one successful response can contain multiple bucket errors, this metric does not necessarily sum to the request type's total errorsPerSecond. A series appears only after its request/error first occurs. Meter From efd8c7922364f984532e3d9c18d49bba2e03a974 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Tue, 22 Sep 2026 15:59:27 +0800 Subject: [PATCH 2/2] [rpc] Complete TabletServer response error metrics coverage Register missing TabletServer APIs and split ListOffsets client/follower metrics. Exercise response errors with EmbeddedChannel tests and document the expanded metrics scope. --- .../rpc/netty/server/NettyServerHandler.java | 7 +- .../rpc/netty/server/RequestsMetrics.java | 25 +- .../netty/server/NettyServerHandlerTest.java | 621 ++++++++++++------ .../rpc/netty/server/RequestsMetricsTest.java | 18 + .../observability/monitor-metrics.md | 11 +- 5 files changed, 457 insertions(+), 225 deletions(-) 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 9817da1a035..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,6 +26,7 @@ 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; @@ -354,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 f8860567555..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: 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 1a81a4049a7..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 @@ -30,25 +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; @@ -57,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; @@ -81,38 +99,53 @@ 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; import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** 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 @@ -250,178 +283,89 @@ public CompletableFuture produceLog( } @Test - void testFailedRequestMarksAggregateAndErrorMetrics() throws Exception { + void testSuccessfulResponseMarksBucketErrorsWithoutAggregate() { RequestsMetricsTest.RecordingMetricRegistry metricRegistry = new RequestsMetricsTest.RecordingMetricRegistry(); - MetricGroup metricGroup = new GenericMetricGroup(metricRegistry, null, "tabletserver"); - TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); - NettyServerHandler tabletServerHandler = - new NettyServerHandler( - tabletRequestChannel, - new ApiManager(ServerType.TABLET_SERVER), - "FLUSS", - true, - RequestsMetrics.createTabletServerRequestMetrics(metricGroup), - new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); - ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); - tabletServerHandler.channelActive(tabletContext); - - 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 = - 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 -> + 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()) - .doesNotContainKey("error")) - .anySatisfy( - registered -> + .containsEntry("error", Errors.TABLE_NOT_EXIST.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(2); + }) + .anySatisfy( + registered -> { assertThat(registered.group.getAllVariables()) - .containsEntry("error", Errors.TABLE_NOT_EXIST.name())); - } - - @Test - void testSuccessfulResponseMarksBucketErrorsWithoutAggregate() throws Exception { - RequestsMetricsTest.RecordingMetricRegistry metricRegistry = - new RequestsMetricsTest.RecordingMetricRegistry(); - MetricGroup metricGroup = new GenericMetricGroup(metricRegistry, null, "tabletserver"); - TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); - NettyServerHandler tabletServerHandler = - new NettyServerHandler( - tabletRequestChannel, - new ApiManager(ServerType.TABLET_SERVER), - "FLUSS", - true, - RequestsMetrics.createTabletServerRequestMetrics(metricGroup), - new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); - ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); - tabletServerHandler.channelActive(tabletContext); - - LookupRequest lookupRequest = new LookupRequest().setTableId(1); - PbLookupReqForBucket bucketRequest = - new PbLookupReqForBucket().setPartitionId(1).setBucketId(1); - bucketRequest.addKey("key".getBytes()); - lookupRequest.addAllBucketsReqs(Collections.singleton(bucketRequest)); - tabletServerHandler.channelRead( - tabletContext, - MessageCodec.encodeRequest( - ByteBufAllocator.DEFAULT, - ApiKeys.LOOKUP.id, - ApiKeys.LOOKUP.highestSupportedVersion, - 1001, - lookupRequest)); - - 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()); - FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); - request.complete(response); - - 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); - }); - assertThat(metricRegistry.metrics(MetricNames.ERRORS_RATE, "lookup")) - .filteredOn(registered -> !registered.group.getAllVariables().containsKey("error")) - .allSatisfy( - registered -> assertThat(((Meter) registered.metric).getCount()).isZero()); + .containsEntry("error", Errors.UNKNOWN_SERVER_ERROR.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(1); + }); + assertAggregateErrorsAreZero(metricRegistry, "lookup"); + } finally { + fixture.channel.finishAndReleaseAll(); + } } @Test - void testEncodingFailureMarksOnlyWholeRpcError() throws Exception { + void testEncodingFailureMarksOnlyWholeRpcError() { RequestsMetricsTest.RecordingMetricRegistry metricRegistry = new RequestsMetricsTest.RecordingMetricRegistry(); - MetricGroup metricGroup = new GenericMetricGroup(metricRegistry, null, "tabletserver"); - TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); - NettyServerHandler tabletServerHandler = - new NettyServerHandler( - tabletRequestChannel, - new ApiManager(ServerType.TABLET_SERVER), - "FLUSS", - true, - RequestsMetrics.createTabletServerRequestMetrics(metricGroup), - new PlainTextAuthenticationPlugin.PlainTextServerAuthenticator()); - ChannelHandlerContext tabletContext = mockImmediateChannelHandlerContext(); - tabletServerHandler.channelActive(tabletContext); - - LookupRequest lookupRequest = new LookupRequest().setTableId(1); - PbLookupReqForBucket bucketRequest = - new PbLookupReqForBucket().setPartitionId(1).setBucketId(1); - bucketRequest.addKey("key".getBytes()); - lookupRequest.addAllBucketsReqs(Collections.singleton(bucketRequest)); - tabletServerHandler.channelRead( - tabletContext, - MessageCodec.encodeRequest( - ByteBufAllocator.DEFAULT, - ApiKeys.LOOKUP.id, - ApiKeys.LOOKUP.highestSupportedVersion, - 1001, - lookupRequest)); - - LookupResponse response = new LookupResponse(); - response.addBucketsResp().setErrorCode(Errors.TABLE_NOT_EXIST.code()); - FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); - request.complete(response); - - verify(tabletContext).writeAndFlush(any(ByteBuf.class)); - 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())); + 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 @@ -451,60 +395,290 @@ void testBatchResponsesCollectErrorCounts() { } @Test - void testResponseMetricsFailureDoesNotSendErrorResponse() throws Exception { + void testResponseMetricsFailureDoesNotSendErrorResponse() { MetricGroup metricGroup = new GenericMetricGroup( new ErrorMetricRegistrationFailureRegistry(), null, "tabletserver"); - TestingRequestChannel tabletRequestChannel = new TestingRequestChannel(100); - NettyServerHandler tabletServerHandler = + 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(); + 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)); - tabletServerHandler.channelRead( - tabletContext, + 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)); + apiKey.id, + apiKey.highestSupportedVersion, + requestId, + request)); + } - LookupResponse response = new LookupResponse(); - response.addBucketsResp().setBucketId(1).setErrorCode(Errors.TABLE_NOT_EXIST.code()); - FlussRequest request = (FlussRequest) tabletRequestChannel.getAndRemoveRequest(0); + private static void completeRequest(HandlerFixture fixture, ApiMessage response) { + FlussRequest request = (FlussRequest) fixture.requestChannel.getAndRemoveRequest(0); request.complete(response); + fixture.channel.runPendingTasks(); + } - verify(tabletContext).flush(); - verify(tabletContext, never()).writeAndFlush(any(ByteBuf.class)); + 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 class ErrorMetricRegistrationFailureRegistry - extends RequestsMetricsTest.RecordingMetricRegistry { + 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); + }); + } - @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); + 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(((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(); @@ -576,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(); @@ -663,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"); @@ -676,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 b87c74d8939..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 failed RPC responses processed per second for each request type. Errors carried by successful response buckets are excluded. + 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 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. Because one successful response can contain multiple bucket errors, this metric does not necessarily sum to the request type's total errorsPerSecond. 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