fix(server): record bounded request body and client IP in slow query log

hugegraph-cluster-test/hugegraph-clustertest-dist/src/assembly/static/conf/rest-server.properties.template

@@ -70,3 +70,5 @@ server.role=$ROLE$
 
 # slow query log
 log.slow_query_threshold=1000
+# bytes of request body recorded as-is (may contain sensitive literals), 0 to disable
+log.slow_query_body_limit=512

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/AccessLogFilter.java

@@ -23,30 +23,44 @@
 import static org.apache.hugegraph.metrics.MetricsUtil.METRICS_PATH_SUCCESS_COUNTER;
 import static org.apache.hugegraph.metrics.MetricsUtil.METRICS_PATH_TOTAL_COUNTER;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.lang.reflect.Method;
 import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.CodingErrorAction;
 import java.util.Map;
 
+import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.DecompressInterceptor.Decompress;
 import org.apache.hugegraph.config.HugeConfig;
 import org.apache.hugegraph.config.ServerOptions;
 import org.apache.hugegraph.core.GraphManager;
 import org.apache.hugegraph.metrics.MetricsUtil;
 import org.apache.hugegraph.util.Log;
+import org.glassfish.grizzly.http.server.Request;
 import org.slf4j.Logger;
 
 import jakarta.inject.Singleton;
 import jakarta.ws.rs.HttpMethod;
 import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerRequestFilter;
 import jakarta.ws.rs.container.ContainerResponseContext;
 import jakarta.ws.rs.container.ContainerResponseFilter;
+import jakarta.ws.rs.container.ResourceInfo;
 import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.MultivaluedMap;
 import jakarta.ws.rs.ext.Provider;
 
-// TODO: should add test for this class
 @Provider
 @Singleton
-public class AccessLogFilter implements ContainerResponseFilter {
+public class AccessLogFilter implements ContainerRequestFilter, ContainerResponseFilter {
 
     private static final Logger LOG = Log.logger(AccessLogFilter.class);
 
@@ -55,14 +69,27 @@ public class AccessLogFilter implements ContainerResponseFilter {
     private static final String GREMLIN = "gremlin";
     private static final String CYPHER = "cypher";
 
+    // Request property holding the bounded request body preview for the slow query log
+    public static final String REQUEST_BODY = "request_body";
+    public static final String UNKNOWN_IP = "<unknown_ip>";
+
+    private static final String TRUNCATED_MARK = "...";
+    private static final String ENCODED_BODY = "<encoded>";
+    private static final Charset DEFAULT_CHARSET = Charset.forName(API.CHARSET);
+
     @Context
     private jakarta.inject.Provider<HugeConfig> configProvider;
 
     @Context
     private jakarta.inject.Provider<GraphManager> managerProvider;
 
+    @Context
+    private jakarta.inject.Provider<Request> requestProvider;
+
+    @Context
+    private ResourceInfo resourceInfo;
+
     public static boolean needRecordLog(ContainerRequestContext context) {
-        // TODO: add test for 'path' result ('/gremlin' or 'gremlin')
         String path = context.getUriInfo().getPath();
 
         // GraphsAPI/CypherAPI/Job GremlinAPI
@@ -83,8 +110,7 @@ private static String normalizePath(ContainerRequestContext requestContext) {
         // Replace variable parts of the path with placeholders
         String requestPath = requestContext.getUriInfo().getPath();
         // get uri params
-        MultivaluedMap<String, String> pathParameters = requestContext.getUriInfo()
-                                                                      .getPathParameters();
+        MultivaluedMap<String, String> pathParameters = requestContext.getUriInfo().getPathParameters();
 
         String newPath = requestPath;
         for (Map.Entry<String, java.util.List<String>> entry : pathParameters.entrySet()) {
@@ -100,15 +126,50 @@ private static String normalizePath(ContainerRequestContext requestContext) {
         return newPath;
     }
 
+    /**
+     * Keep a bounded preview of the request body for the slow query log.
+     * Only the first {@link ServerOptions#SLOW_QUERY_LOG_BODY_LIMIT} bytes are read, and they are replayed in front
+     * of the untouched remainder of the entity stream, so the resource method still receives the whole body.
+     *
+     * @param requestContext requestContext
+     */
+    @Override
+    public void filter(ContainerRequestContext requestContext) throws IOException {
+        if (!mayHaveBody(requestContext.getMethod()) || !needRecordLog(requestContext)) {
+            return;
+        }
+
+        HugeConfig config = this.configProvider.get();
+        long timeThreshold = config.get(ServerOptions.SLOW_QUERY_LOG_TIME_THRESHOLD);
+        int bodyLimit = config.get(ServerOptions.SLOW_QUERY_LOG_BODY_LIMIT);
+        if (timeThreshold <= 0 || bodyLimit <= 0) {
+            return;
+        }
+
+        if (this.decodesEntity()) {
+            // The resource decodes its entity later (DecompressInterceptor), so the raw bytes here are not readable
+            requestContext.setProperty(REQUEST_BODY, ENCODED_BODY);
+            return;
+        }
+
+        InputStream entity = requestContext.getEntityStream();
+        // Read one byte past the limit to know whether the preview is truncated
+        byte[] prefix = new byte[bodyLimit + 1];
+        int length = entity.readNBytes(prefix, 0, prefix.length);
+        requestContext.setEntityStream(new SequenceInputStream(new ByteArrayInputStream(prefix, 0, length), entity));
+        Charset charset = requestCharset(requestContext);
+        requestContext.setProperty(REQUEST_BODY, preview(prefix, length, bodyLimit, charset));
+    }
+
     /**
      * Use filter to log request info
      *
      * @param requestContext  requestContext
      * @param responseContext responseContext
      */
     @Override
-    public void filter(ContainerRequestContext requestContext,
-                       ContainerResponseContext responseContext) throws IOException {
+    public void filter(ContainerRequestContext requestContext, ContainerResponseContext responseContext)
+            throws IOException {
         // Grab corresponding request / response info from context;
         URI uri = requestContext.getUriInfo().getRequestUri();
         String method = requestContext.getMethod();
@@ -145,11 +206,12 @@ public void filter(ContainerRequestContext requestContext,
             HugeConfig config = configProvider.get();
             long timeThreshold = config.get(ServerOptions.SLOW_QUERY_LOG_TIME_THRESHOLD);
             // Record slow query if meet needs, watch out the perf
-            if (timeThreshold > 0 && executeTime > timeThreshold &&
-                needRecordLog(requestContext)) {
-                // TODO: set RequestBody null, handle it later & should record "client IP"
-                LOG.info("[Slow Query] execTime={}ms, body={}, method={}, path={}, query={}",
-                         executeTime, null, method, path, uri.getQuery());
+            if (timeThreshold > 0 && executeTime > timeThreshold && needRecordLog(requestContext)) {
+                String clientIp = this.clientIp();
+                Object body = requestContext.getProperty(REQUEST_BODY);
+                LOG.info("[Slow Query] ip={}, execTime={}ms, method={}, path={}, query={}, " +
+                         "body={}", clientIp, executeTime, method, singleLine(path),
+                         singleLine(uri.getQuery()), singleLine(body));
             }
         }
 
@@ -161,6 +223,55 @@ public void filter(ContainerRequestContext requestContext,
         }
     }
 
+    private static boolean mayHaveBody(String method) {
+        // DELETE endpoints take path/query params only, so there is no body to record
+        return HttpMethod.POST.equals(method) || HttpMethod.PUT.equals(method);
+    }
+
+    private boolean decodesEntity() {
+        Method method = this.resourceInfo == null ? null : this.resourceInfo.getResourceMethod();
+        return method != null && method.isAnnotationPresent(Decompress.class);
+    }
+
+    private static Charset requestCharset(ContainerRequestContext requestContext) {
+        MediaType mediaType = requestContext.getMediaType();
+        if (mediaType == null) {
+            return DEFAULT_CHARSET;
+        }
+        String charset = mediaType.getParameters().get(MediaType.CHARSET_PARAMETER);
+        return charset == null ? DEFAULT_CHARSET : Charset.forName(charset);
+    }
+
+    private static String preview(byte[] bytes, int length, int limit, Charset charset) {
+        boolean truncated = length > limit;
+        int size = Math.min(length, limit);
+        CharsetDecoder decoder = charset.newDecoder().onMalformedInput(CodingErrorAction.REPLACE)
+                                        .onUnmappableCharacter(CodingErrorAction.REPLACE);
+        CharBuffer chars = CharBuffer.allocate((int) (size * decoder.maxCharsPerByte()) + 1);
+        /*
+         * A multi-byte character cut by the limit is dropped rather than turned
+         * into a replacement character: endOfInput=false keeps the incomplete
+         * tail undecoded
+         */
+        decoder.decode(ByteBuffer.wrap(bytes, 0, size), chars, !truncated);
+        if (!truncated) {
+            decoder.flush(chars);
+        }
+        String body = chars.flip().toString();
+        return truncated ? body + TRUNCATED_MARK : body;
+    }
+
+    private static String singleLine(Object value) {
+        // Path, query and body are client controlled, keep the log entry on a single line
+        return value == null ? null : value.toString().replace("\r", "\\r").replace("\n", "\\n");
+    }
+
+    private String clientIp() {
+        Request request = this.requestProvider.get();
+        String address = request == null ? null : request.getRemoteAddr();
+        return address == null || address.isEmpty() ? UNKNOWN_IP : address;
+    }
+
     private boolean statusOk(int status) {
         return status >= 200 && status < 300;
     }

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/PathFilter.java

@@ -50,7 +50,6 @@ public class PathFilter implements ContainerRequestFilter {
     private static final String ARTHAS_START = "arthas";
 
     public static final String REQUEST_TIME = "request_time";
-    public static final String REQUEST_PARAMS_JSON = "request_params_json";
 
     private static final String DELIMITER = "/";
     private static final Set<String> WHITE_API_LIST = ImmutableSet.of(

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilterDynamicFeature.java

@@ -17,6 +17,7 @@
 
 package org.apache.hugegraph.api.filter;
 
+import jakarta.ws.rs.Priorities;
 import jakarta.ws.rs.container.DynamicFeature;
 import jakarta.ws.rs.container.ResourceInfo;
 import jakarta.ws.rs.core.FeatureContext;
@@ -27,9 +28,8 @@ public class RedirectFilterDynamicFeature implements DynamicFeature {
 
     @Override
     public void configure(ResourceInfo resourceInfo, FeatureContext context) {
-        if (resourceInfo.getResourceMethod()
-                        .isAnnotationPresent(RedirectFilter.RedirectMasterRole.class)) {
-            context.register(RedirectFilter.class);
+        if (resourceInfo.getResourceMethod().isAnnotationPresent(RedirectFilter.RedirectMasterRole.class)) {
+            context.register(RedirectFilter.class, Priorities.USER + 1);
         }
     }
 }

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java

@@ -657,6 +657,17 @@ public class ServerOptions extends OptionHolder {
                     nonNegativeInt(),
                     1000L
             );
+
+    public static final ConfigOption<Integer> SLOW_QUERY_LOG_BODY_LIMIT =
+            new ConfigOption<>(
+                    "log.slow_query_body_limit",
+                    "The max bytes of request body recorded in the slow query log, " +
+                    "the recorded prefix is written as-is and may contain sensitive " +
+                    "literals of gremlin/cypher scripts, 0 means the body is not recorded.",
+                    rangeInt(0, 1024 * 1024),
+                    512
+            );
+
     public static final ConfigOption<Double> JVM_MEMORY_MONITOR_THRESHOLD =
             new ConfigOption<>(
                     "memory_monitor.threshold",

hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties

@@ -28,6 +28,8 @@ arthas.disabled_commands=jad
 
 # slow query log
 log.slow_query_threshold=1000
+# bytes of request body recorded as-is (may contain sensitive literals), 0 to disable
+log.slow_query_body_limit=512
 
 # jvm(in-heap) memory usage monitor, set 1 to disable it
 memory_monitor.threshold=0.85

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java

@@ -28,6 +28,7 @@
 import org.apache.hugegraph.meta.managers.AuthMetaManagerTest;
 import org.apache.hugegraph.traversal.optimize.TraversalUtilOptimizeTest;
 import org.apache.hugegraph.unit.api.auth.LoginAPITest;
+import org.apache.hugegraph.unit.api.filter.AccessLogFilterTest;
 import org.apache.hugegraph.unit.api.filter.LoadDetectFilterTest;
 import org.apache.hugegraph.unit.api.filter.PathFilterTest;
 import org.apache.hugegraph.unit.api.gremlin.GremlinQueryAPITest;
@@ -103,6 +104,7 @@
 @RunWith(Suite.class)
 @Suite.SuiteClasses({
         /* api filter */
+        AccessLogFilterTest.class,
         LoadDetectFilterTest.class,
         LoginAPITest.class,
         PathFilterTest.class,