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 b04722d5c01..f158afa3bb4 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 @@ -274,7 +274,7 @@ private void sendError(ChannelHandlerContext ctx, FlussRequest request, Throwabl ByteBuf byteBuf = encodeErrorResponse(alloc, request.getRequestId(), error); ctx.writeAndFlush(byteBuf); - getMetrics(request).ifPresent(metrics -> metrics.getErrorsCount().inc()); + getMetrics(request).ifPresent(metrics -> metrics.markError(error.error())); } 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 d3aa345ca99..31c8b9ce3ad 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 @@ -26,6 +26,7 @@ import org.apache.fluss.metrics.ThreadSafeSimpleCounter; import org.apache.fluss.metrics.groups.MetricGroup; import org.apache.fluss.rpc.protocol.ApiKeys; +import org.apache.fluss.rpc.protocol.Errors; import java.util.Arrays; import java.util.Collection; @@ -34,6 +35,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; /** * A class wrapping the metrics registered for different request types. It's mainly used to simplify @@ -129,8 +131,10 @@ public Optional getMetrics( /** A class wrapping all registered metrics for a given request type. */ public static final class Metrics { private static final int WINDOW_SIZE = 1024; + private final MetricGroup metricGroup; private final Counter requestsCount; private final Counter errorsCount; + private final Map errorsCountByError = new ConcurrentHashMap<>(); private final Histogram requestBytes; @@ -140,6 +144,7 @@ public static final class Metrics { private final Histogram totalTimeMs; private Metrics(MetricGroup metricGroup) { + this.metricGroup = metricGroup; requestsCount = new ThreadSafeSimpleCounter(); metricGroup.meter(MetricNames.REQUESTS_RATE, new MeterView(requestsCount)); errorsCount = new ThreadSafeSimpleCounter(); @@ -175,6 +180,19 @@ public Counter getErrorsCount() { return errorsCount; } + void markError(Errors error) { + errorsCount.inc(); + errorsCountByError.computeIfAbsent(error, this::registerErrorMeter).inc(); + } + + private Counter registerErrorMeter(Errors error) { + Counter perErrorCount = new ThreadSafeSimpleCounter(); + metricGroup + .addGroup("error", error.name()) + .meter(MetricNames.ERRORS_RATE, new MeterView(perErrorCount)); + return perErrorCount; + } + public Histogram getRequestBytes() { return requestBytes; } 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 113bf3f7ace..13c131a5c9f 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 @@ -18,6 +18,10 @@ package org.apache.fluss.rpc.netty.server; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.metrics.Meter; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.GenericMetricGroup; import org.apache.fluss.metrics.groups.MetricGroup; import org.apache.fluss.metrics.util.NOPMetricsGroup; import org.apache.fluss.rpc.messages.ApiVersionsRequest; @@ -30,6 +34,7 @@ import org.apache.fluss.rpc.messages.PbValue; 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.security.auth.PlainTextAuthenticationPlugin; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; @@ -38,11 +43,13 @@ import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandlerContext; import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelId; 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 java.net.InetSocketAddress; import java.time.Duration; import java.util.Collections; import java.util.Deque; @@ -77,6 +84,55 @@ void beforeEach() throws Exception { serverHandler.channelActive(ctx); } + @Test + void testFailedRequestMarksAggregateAndErrorMetrics() 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)); + 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 -> + assertThat(registered.group.getAllVariables()) + .doesNotContainKey("error")) + .anySatisfy( + registered -> + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.TABLE_NOT_EXIST.name())); + } + @Test @Disabled("TODO: add back in https://github.com/apache/fluss/issues/771") void testResponseReturnInOrder() throws Exception { @@ -216,6 +272,13 @@ 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 new file mode 100644 index 00000000000..b6f6e735726 --- /dev/null +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/server/RequestsMetricsTest.java @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.rpc.netty.server; + +import org.apache.fluss.metrics.CharacterFilter; +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.registry.MetricRegistry; +import org.apache.fluss.rpc.protocol.ApiKeys; +import org.apache.fluss.rpc.protocol.Errors; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link RequestsMetrics}. */ +class RequestsMetricsTest { + + @Test + void errorsKeepAggregateAndRegisterBreakdownByError() { + RecordingMetricRegistry registry = new RecordingMetricRegistry(); + MetricGroup serverMetricGroup = new GenericMetricGroup(registry, null, "tabletserver"); + RequestsMetrics requestsMetrics = + RequestsMetrics.createTabletServerRequestMetrics(serverMetricGroup); + + List initialErrorMeters = + registry.metrics(MetricNames.ERRORS_RATE, "produceLog"); + assertThat(initialErrorMeters).hasSize(1); + assertThat(initialErrorMeters.get(0).group.getAllVariables()).doesNotContainKey("error"); + assertThat(((Meter) initialErrorMeters.get(0).metric).getCount()).isZero(); + + RequestsMetrics.Metrics metrics = + requestsMetrics.getMetrics(ApiKeys.PRODUCE_LOG.id, false, false).get(); + metrics.markError(Errors.TABLE_NOT_EXIST); + metrics.markError(Errors.TABLE_NOT_EXIST); + metrics.markError(Errors.UNKNOWN_SERVER_ERROR); + + List errorMeters = + registry.metrics(MetricNames.ERRORS_RATE, "produceLog"); + assertThat(errorMeters).hasSize(3); + assertThat(errorMeters) + .allSatisfy( + registered -> { + assertThat(registered.metricName).isEqualTo("errorsPerSecond"); + assertThat(registered.group.getAllVariables()) + .containsEntry("request", "produceLog"); + assertThat(registered.metric).isInstanceOf(Meter.class); + }); + assertThat(errorMeters) + .anySatisfy( + registered -> { + assertThat( + registered.group.getLogicalScope( + CharacterFilter.NO_OP_FILTER, '_')) + .isEqualTo("tabletserver_request"); + assertThat(registered.group.getAllVariables()) + .doesNotContainKey("error"); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(3); + }); + assertThat(errorMeters) + .filteredOn(registered -> registered.group.getAllVariables().containsKey("error")) + .allSatisfy( + registered -> { + assertThat( + registered.group.getLogicalScope( + CharacterFilter.NO_OP_FILTER, '_')) + .isEqualTo("tabletserver_request_error"); + }); + assertThat(errorMeters) + .anySatisfy( + registered -> { + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.TABLE_NOT_EXIST.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(2); + }); + assertThat(errorMeters) + .anySatisfy( + registered -> { + assertThat(registered.group.getAllVariables()) + .containsEntry("error", Errors.UNKNOWN_SERVER_ERROR.name()); + assertThat(((Meter) registered.metric).getCount()).isEqualTo(1); + }); + } + + static class RecordingMetricRegistry implements MetricRegistry { + + private final List registeredMetrics = new ArrayList<>(); + + @Override + public int getNumberReporters() { + return 0; + } + + @Override + public void register(Metric metric, String metricName, AbstractMetricGroup group) { + registeredMetrics.add(new RegisteredMetric(metric, metricName, group)); + } + + @Override + public void unregister(Metric metric, String metricName, AbstractMetricGroup group) {} + + @Override + public CompletableFuture closeAsync() { + return CompletableFuture.completedFuture(null); + } + + List metrics(String metricName) { + return registeredMetrics.stream() + .filter(metric -> metric.metricName.equals(metricName)) + .collect(Collectors.toList()); + } + + List metrics(String metricName, String request) { + return metrics(metricName).stream() + .filter(metric -> request.equals(metric.group.getAllVariables().get("request"))) + .collect(Collectors.toList()); + } + } + + static class RegisteredMetric { + + final Metric metric; + final String metricName; + final MetricGroup group; + + private RegisteredMetric(Metric metric, String metricName, MetricGroup group) { + this.metric = metric; + this.metricName = metricName; + this.group = group; + } + } +} diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 06d7ca43394..fdcfac1d2a8 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/website/docs/maintenance/observability/monitor-metrics.md @@ -683,7 +683,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM Gauge - tabletserver + tabletserver request requestQueueSize The TabletServer node network waiting queue size. @@ -740,6 +740,12 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM responseSendTimeMs Time to send the response for each request type. Histogram + + + 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. + Meter client