feat: Add logic to include client-id to group requests from same Iceberg client

ice-rest-catalog/README.md

@@ -24,6 +24,23 @@ commitLock:
 
 If `enabled` is true but the catalog backend is not etcd, the lock is ignored (warning in logs). When lock acquisition exceeds `acquireTimeoutMs`, the server responds with HTTP **503** so clients can retry.
 
+## Debugging REST API traffic
+
+Set `ICE_REST_CATALOG_LOG_LEVEL=DEBUG` to log the full JSON request and response bodies for every REST call. This is useful for inspecting exactly what an Iceberg client sends and what the server returns.
+
+**Local:**
+```bash
+ICE_REST_CATALOG_LOG_LEVEL=DEBUG java -jar ice-rest-catalog/target/ice-rest-catalog-jar
+```
+
+**Docker / docker-compose:**
+```yaml
+environment:
+  ICE_REST_CATALOG_LOG_LEVEL: DEBUG
+```
+
+The default level is `INFO`, which logs one line per request (`@uid METHOD path`) without bodies.
+
 ## Documentation
 
 - [Architecture](../docs/architecture.md) -- components, design principles, HA, backup/recovery

ice-rest-catalog/pom.xml

@@ -598,6 +598,7 @@
             <exclude>**/DockerScenarioBasedIT.java</exclude>
             <exclude>**/DockerLocalFileIOClickHouseIT.java</exclude>
             <exclude>**/DockerLocalFileIOClickHouseAllTypesIT.java</exclude>
+            <exclude>**/DockerSparkClientIdIT.java</exclude>
           </excludes>
         </configuration>
       </plugin>

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/Main.java

@@ -38,7 +38,9 @@
 import com.altinity.ice.rest.catalog.internal.rest.RESTCatalogMiddlewareCredentials;
 import com.altinity.ice.rest.catalog.internal.rest.RESTCatalogMiddlewareTableConfig;
 import com.altinity.ice.rest.catalog.internal.rest.RESTCatalogMiddlewareTableCredentials;
+import com.altinity.ice.rest.catalog.internal.rest.RESTCatalogMiddlewareTracing;
 import com.altinity.ice.rest.catalog.internal.rest.RESTCatalogServlet;
+import com.altinity.ice.rest.catalog.internal.rest.RequestTracing;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.net.HostAndPort;
 import io.prometheus.metrics.instrumentation.jvm.JvmMetrics;
@@ -324,6 +326,14 @@ private static Server createBaseServer(
       }
     }
 
+    var tracingConfig = config.tracing();
+    if (tracingConfig.enabledOrDefault() && tracingConfig.advertiseClientIdOrDefault()) {
+      restCatalogAdapter =
+          new RESTCatalogMiddlewareTracing(
+              restCatalogAdapter, tracingConfig.clientIdHeaderOrDefault());
+    }
+    RequestTracing tracing = new RequestTracing(tracingConfig);
+
     logger.info(
         "Commit retry config: numRetries={} minWaitMs={} maxWaitMs={} totalTimeoutMs={}",
         config.commitRetry().numRetries(),
@@ -336,7 +346,7 @@ private static Server createBaseServer(
         config.commitLock().leaseTtlSeconds(),
         config.commitLock().acquireTimeoutMs());
 
-    var h = new ServletHolder(new RESTCatalogServlet(restCatalogAdapter));
+    var h = new ServletHolder(new RESTCatalogServlet(restCatalogAdapter, tracing));
     mux.addServlet(h, "/*");
 
     if (registerAdminServlet) {

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/Config.java

@@ -68,7 +68,10 @@ public record Config(
     @JsonPropertyDescription(
             "(experimental) Iceberg properties (see https://iceberg.apache.org/javadoc/1.8.1/"
                 + "; e.g. https://iceberg.apache.org/javadoc/1.8.1/org/apache/iceberg/aws/s3/S3FileIOProperties.html)")
-        Map<String, String> icebergProperties) {
+        Map<String, String> icebergProperties,
+    @JsonPropertyDescription(
+            "OpenTelemetry tracing and request/client correlation config (disabled by default)")
+        TracingConfig tracing) {
 
   private static final String DEFAULT_ADDR = "0.0.0.0:5000";
   private static final String DEFAULT_DEBUG_ADDR = "0.0.0.0:5001";
@@ -90,7 +93,8 @@ public Config(
       CommitRetryConfig commitRetry,
       CommitLockConfig commitLock,
       Map<String, String> loadTableProperties,
-      @JsonProperty("iceberg") Map<String, String> icebergProperties) {
+      @JsonProperty("iceberg") Map<String, String> icebergProperties,
+      TracingConfig tracing) {
     this.addr = Strings.orDefault(addr, DEFAULT_ADDR);
     this.debugAddr = Strings.orDefault(debugAddr, DEFAULT_DEBUG_ADDR);
     this.adminAddr = Strings.orDefault(adminAddr, System.getenv("ICE_REST_CATALOG_ADMIN_ADDR"));
@@ -110,6 +114,7 @@ public Config(
     this.commitLock = Objects.requireNonNullElse(commitLock, CommitLockConfig.defaults());
     this.loadTableProperties = Objects.requireNonNullElse(loadTableProperties, Map.of());
     this.icebergProperties = Objects.requireNonNullElse(icebergProperties, Map.of());
+    this.tracing = Objects.requireNonNullElse(tracing, TracingConfig.defaults());
   }
 
   public record S3(

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/TracingConfig.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * 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
+ */
+package com.altinity.ice.rest.catalog.internal.config;
+
+import com.altinity.ice.internal.strings.Strings;
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+
+/**
+ * Per-request / per-client correlation configuration.
+ *
+ * <p>Adds a stable {@code clientId} (a deterministic per-identity fingerprint grouping all calls
+ * from the same uid + remote address + user-agent, including across separate client runs) and a
+ * per-request {@code requestId} to logs (via MDC) and response headers, so a whole client's
+ * activity can be grouped from the logs alone. All settings are optional; the feature is on by
+ * default and has no external dependencies.
+ */
+public record TracingConfig(
+    @JsonPropertyDescription(
+            "Enable request/client correlation ids in logs and response headers (true by default)")
+        Boolean enabled,
+    @JsonPropertyDescription(
+            "Header carrying the per-client id echoed by Iceberg clients (X-Ice-Client-Id)")
+        String clientIdHeader,
+    @JsonPropertyDescription("Header carrying the per-request id (X-Request-Id)")
+        String requestIdHeader,
+    @JsonPropertyDescription(
+            "Advertise the resolved (deterministic fingerprint) client id via the /v1/config response so Iceberg Java/PyIceberg clients echo the same id on every request, keeping it stable across runs (true by default)")
+        Boolean advertiseClientId) {
+
+  public static final String DEFAULT_CLIENT_ID_HEADER = "X-Ice-Client-Id";
+  public static final String DEFAULT_REQUEST_ID_HEADER = "X-Request-Id";
+
+  public static TracingConfig defaults() {
+    return new TracingConfig(null, null, null, null);
+  }
+
+  public boolean enabledOrDefault() {
+    return enabled == null || enabled;
+  }
+
+  public String clientIdHeaderOrDefault() {
+    return !Strings.isNullOrEmpty(clientIdHeader) ? clientIdHeader : DEFAULT_CLIENT_ID_HEADER;
+  }
+
+  public String requestIdHeaderOrDefault() {
+    return !Strings.isNullOrEmpty(requestIdHeader) ? requestIdHeader : DEFAULT_REQUEST_ID_HEADER;
+  }
+
+  public boolean advertiseClientIdOrDefault() {
+    return advertiseClientId == null || advertiseClientId;
+  }
+}

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/rest/RESTCatalogMiddlewareTracing.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * 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
+ */
+package com.altinity.ice.rest.catalog.internal.rest;
+
+import com.altinity.ice.rest.catalog.internal.auth.Session;
+import java.util.Map;
+import org.apache.iceberg.rest.RESTResponse;
+import org.apache.iceberg.rest.responses.ConfigResponse;
+import org.slf4j.MDC;
+
+/**
+ * Advertises the client id to Iceberg clients via the {@code /v1/config} handshake.
+ *
+ * <p>The advertised value is the id already resolved for the config request (read from the {@link
+ * MDC}) - a deterministic per-identity fingerprint ({@code fp-...}) derived from uid + remote
+ * address + user-agent. It is returned as a {@code header.<clientIdHeader>} override. Iceberg Java
+ * and PyIceberg merge config overrides and re-send matching {@code header.*} properties on every
+ * subsequent request, so all calls from that identity share one {@code clientId} - and because the
+ * value is deterministic (not a random per-instance id), it stays stable across separate
+ * short-lived client runs. Clients that ignore config overrides (e.g. ClickHouse) fall back to the
+ * same server-side fingerprint anyway.
+ */
+public class RESTCatalogMiddlewareTracing extends RESTCatalogMiddleware {
+
+  private static final String HEADER_PREFIX = "header.";
+
+  private final String clientIdHeader;
+
+  public RESTCatalogMiddlewareTracing(RESTCatalogHandler next, String clientIdHeader) {
+    super(next);
+    this.clientIdHeader = clientIdHeader;
+  }
+
+  @Override
+  public <T extends RESTResponse> T handle(
+      Session session,
+      Route route,
+      Map<String, String> vars,
+      Object requestBody,
+      Class<T> responseType) {
+    T restResponse = this.next.handle(session, route, vars, requestBody, responseType);
+    if (restResponse instanceof ConfigResponse configResponse) {
+      // Advertise the id already resolved for this config request (the stable fingerprint), so
+      // echoing clients keep using the same deterministic id across runs.
+      String clientId = MDC.get(RequestTracing.MDC_CLIENT_ID);
+      if (clientId != null && !clientId.isEmpty()) {
+        configResponse.overrides().putIfAbsent(HEADER_PREFIX + clientIdHeader, clientId);
+      }
+    }
+    return restResponse;
+  }
+}

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/rest/RESTCatalogServlet.java

@@ -25,6 +25,7 @@
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.hc.core5.http.ContentType;
@@ -76,17 +77,37 @@
 
   private final RESTCatalogHandler restCatalogAdapter;
   private final HttpMetrics httpMetrics;
+  private final RequestTracing tracing;
 
-  public RESTCatalogServlet(RESTCatalogHandler restCatalogAdapter) {
+  public RESTCatalogServlet(RESTCatalogHandler restCatalogAdapter, RequestTracing tracing) {
     this.restCatalogAdapter = restCatalogAdapter;
     this.httpMetrics = HttpMetrics.getInstance();
+    this.tracing = tracing;
   }
 
   protected void handle(HttpServletRequest request, HttpServletResponse response)
       throws IOException {
     HTTPRequest.HTTPMethod method = HTTPRequest.HTTPMethod.valueOf(request.getMethod());
     String path = request.getRequestURI().substring(1);
 
+    Session session = Session.from(request);
+    String clientId = tracing.resolveClientId(request, session);
+    String requestId = tracing.resolveRequestId(request);
+    tracing.begin(response, clientId, requestId);
+    try {
+      handleTraced(method, path, session, request, response);
+    } finally {
+      tracing.end();
+    }
+  }
+
+  private void handleTraced(
+      HTTPRequest.HTTPMethod method,
+      String path,
+      Session session,
+      HttpServletRequest request,
+      HttpServletResponse response)
+      throws IOException {
     Pair<Route, Map<String, String>> routeContext = Route.from(method, path);
     if (routeContext == null) {
       // Track unknown route requests
@@ -110,7 +131,6 @@
 
     // Track request with metrics
     try (var timer = httpMetrics.startRequest(method.name(), route.name())) {
-      Session session = Session.from(request);
       String userToLog = "";
       if (session != null) {
         userToLog = "@" + session.uid() + " ";
@@ -122,10 +142,13 @@
       // FIXME: this should be in RESTCatalogAdapter, not here
       Object requestBody = null;
       if (route.requestClass() != null) {
-        requestBody =
-            RESTObjectMapper.mapper().readValue(request.getReader(), route.requestClass());
+        String rawBody = CharStreams.toString(request.getReader());
+        logger.debug("{}{} {} request body: {}", userToLog, method, path, rawBody);
+        requestBody = RESTObjectMapper.mapper().readValue(rawBody, route.requestClass());
       } else if (route == Route.TOKENS) {
-        requestBody = RESTUtil.decodeFormData(CharStreams.toString(request.getReader()));
+        String rawBody = CharStreams.toString(request.getReader());
+        logger.debug("{}{} {} request body: <redacted>", userToLog, method, path);
+        requestBody = RESTUtil.decodeFormData(rawBody);
       }
 
       Map<String, String> queryParams =
@@ -157,6 +180,12 @@
         response.setStatus(error.code());
         response.setHeader(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType());
         byte[] errorBytes = RESTObjectMapper.mapper().writeValueAsBytes(error);
+        logger.debug(
+            "{}{} {} error response body: {}",
+            userToLog,
+            method,
+            path,
+            new String(errorBytes, StandardCharsets.UTF_8));
         timer.setResponseSize(errorBytes.length);
         response.getOutputStream().write(errorBytes);
         return;
@@ -167,6 +196,12 @@
       response.setHeader(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType());
       if (responseBody != null) {
         byte[] responseBytes = RESTObjectMapper.mapper().writeValueAsBytes(responseBody);
+        logger.debug(
+            "{}{} {} response body: {}",
+            userToLog,
+            method,
+            path,
+            new String(responseBytes, StandardCharsets.UTF_8));
         timer.setResponseSize(responseBytes.length);
         response.getOutputStream().write(responseBytes);
       }