[metrics] Count per-bucket errors in TabletServer response metrics

fluss-common/src/main/java/org/apache/fluss/rpc/messages/ApiMessage.java

@@ -21,13 +21,22 @@
 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.
  */
 @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<Integer, Integer> errorCounts) {}
+
     /** Gets the total serialized byte array size of the message. */
     int totalSize();
 

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<Message> 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;
+    }
+}

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

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<ProtobufField> fields;
     private final List<ProtobufMessage> 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<Integer, Integer> 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();

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;
+}

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

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java

@@ -26,12 +26,14 @@
 import org.apache.fluss.rpc.messages.AuthenticateRequest;
 import org.apache.fluss.rpc.messages.AuthenticateResponse;
 import org.apache.fluss.rpc.messages.FetchLogRequest;
+import org.apache.fluss.rpc.messages.ListOffsetsRequest;
 import org.apache.fluss.rpc.messages.LookupRequest;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.protocol.ApiError;
 import org.apache.fluss.rpc.protocol.ApiKeys;
 import org.apache.fluss.rpc.protocol.ApiManager;
 import org.apache.fluss.rpc.protocol.ApiMethod;
+import org.apache.fluss.rpc.protocol.Errors;
 import org.apache.fluss.rpc.protocol.MessageCodec;
 import org.apache.fluss.security.auth.ServerAuthenticator;
 import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf;
@@ -51,6 +53,8 @@
 import java.net.SocketAddress;
 import java.util.ArrayDeque;
 import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -241,6 +245,7 @@ private void sendResponse(ChannelHandlerContext ctx, FlussRequest request) {
                 CompletableFuture<ApiMessage> f = request.getResponseFuture();
                 sendSuccessResponse(ctx, request, f.get());
             } catch (Throwable t) {
+                ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
                 sendError(ctx, request, t);
             }
         } else {
@@ -265,8 +270,10 @@ private void sendSuccessResponse(
                     request.getRequestCompletedTimeMs() - request.getRequestDequeTimeMs(),
                     requestEndTimeMs - request.getRequestCompletedTimeMs(),
                     request.getAddress());
-            updateRequestMetrics(request, requestEndTimeMs);
+            tryUpdateRequestMetrics(request, requestEndTimeMs);
+            tryUpdateResponseErrorMetrics(request, responseMessage);
         } catch (Throwable t) {
+            ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
             LOG.error("Failed to send response to client.", t);
             sendError(ctx, request, t);
         }
@@ -279,7 +286,45 @@ private void sendError(ChannelHandlerContext ctx, FlussRequest request, Throwabl
         ByteBuf byteBuf = encodeErrorResponse(alloc, request.getRequestId(), error);
         ctx.writeAndFlush(byteBuf);
 
-        getMetrics(request).ifPresent(metrics -> metrics.markError(error.error()));
+        try {
+            getMetrics(request).ifPresent(metrics -> metrics.markError(error.error()));
+        } catch (Throwable metricFailure) {
+            logMetricsFailure("Failed to update request error metrics.", metricFailure);
+        }
+    }
+
+    private void tryUpdateRequestMetrics(FlussRequest request, long requestEndTimeMs) {
+        try {
+            updateRequestMetrics(request, requestEndTimeMs);
+        } catch (Throwable t) {
+            logMetricsFailure("Failed to update request metrics.", t);
+        }
+    }
+
+    private void tryUpdateResponseErrorMetrics(FlussRequest request, ApiMessage responseMessage) {
+        try {
+            Optional<RequestsMetrics.Metrics> metrics = getMetrics(request);
+            if (!metrics.isPresent()) {
+                return;
+            }
+
+            Map<Integer, Integer> errorCounts = new HashMap<>();
+            responseMessage.collectErrorCounts(errorCounts);
+            errorCounts.forEach(
+                    (errorCode, count) -> {
+                        Errors error = Errors.forCode(errorCode);
+                        if (error != Errors.NONE) {
+                            metrics.get().markErrorCount(error, count);
+                        }
+                    });
+        } catch (Throwable t) {
+            logMetricsFailure("Failed to update response error metrics.", t);
+        }
+    }
+
+    private void logMetricsFailure(String message, Throwable t) {
+        ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
+        LOG.warn(message, t);
     }
 
     private void updateRequestMetrics(FlussRequest request, long requestEndTimeMs) {
@@ -310,9 +355,13 @@ private Optional<RequestsMetrics.Metrics> 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) {