[metrics] Add per-error TabletServer request metrics

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) {

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<Metrics> 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<Errors, Counter> 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;
         }

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();

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<RegisteredMetric> 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<RegisteredMetric> 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<RegisteredMetric> 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<Void> closeAsync() {
+            return CompletableFuture.completedFuture(null);
+        }
+
+        List<RegisteredMetric> metrics(String metricName) {
+            return registeredMetrics.stream()
+                    .filter(metric -> metric.metricName.equals(metricName))
+                    .collect(Collectors.toList());
+        }
+
+        List<RegisteredMetric> 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;
+        }
+    }
+}

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
       <td>Gauge</td>
     </tr>
     <tr>
-      <th rowspan="9">tabletserver</th>
+      <th rowspan="10">tabletserver</th>
       <td rowspan="1">request</td>
       <td>requestQueueSize</td>
       <td>The TabletServer node network waiting queue size.</td>
@@ -740,6 +740,12 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
       <td>responseSendTimeMs</td>
       <td>Time to send the response	for each request type.</td>
       <td>Histogram</td>
+    </tr>
+    <tr>
+      <td rowspan="1">request_error</td>
+      <td>errorsPerSecond</td>
+      <td>The number of failed RPC responses processed per second for each request type and <code>error</code> name. One event is recorded for each failed RPC response; <code>NONE</code> and errors in successful response buckets are excluded. A series appears only after its request/error first occurs.</td>
+      <td>Meter</td>
     </tr>
      <tr>
       <th rowspan="6">client</th>