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

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

@@ -76,17 +76,37 @@ public class RESTCatalogServlet extends HttpServlet {
 
   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 +130,6 @@ protected void handle(HttpServletRequest request, HttpServletResponse response)
 
     // 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() + " ";

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

@@ -0,0 +1,114 @@
+/*
+ * 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.internal.strings.Strings;
+import com.altinity.ice.rest.catalog.internal.auth.Session;
+import com.altinity.ice.rest.catalog.internal.config.TracingConfig;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.zip.CRC32;
+import org.slf4j.MDC;
+
+/**
+ * Resolves and propagates per-client / per-request correlation identifiers.
+ *
+ * <p>{@code clientId} groups every call from a single identity; {@code requestId} is unique per
+ * HTTP request. Both are placed in the SLF4J {@link MDC} (rendered by the logback pattern) and
+ * echoed back as response headers.
+ *
+ * <p>Client id resolution:
+ *
+ * <ul>
+ *   <li>If the request carries the client-id header, that value is used. Iceberg Java / PyIceberg
+ *       echo it once the server advertises it via the {@code /v1/config} response, so all calls
+ *       from that client re-send the same value.
+ *   <li>Otherwise a stable fingerprint derived from the auth token, remote address and User-Agent
+ *       is used. This covers the {@code /v1/config} request itself (before the header is echoed)
+ *       and clients (e.g. ClickHouse) that do not echo config headers.
+ * </ul>
+ *
+ * <p>The advertised id is this same fingerprint (see {@code RESTCatalogMiddlewareTracing}), so it
+ * is deterministic per identity and stable across separate client runs, giving identity-level
+ * grouping.
+ */
+public final class RequestTracing {
+
+  public static final String MDC_CLIENT_ID = "clientId";
+  public static final String MDC_REQUEST_ID = "requestId";
+
+  private final boolean enabled;
+  private final String clientIdHeader;
+  private final String requestIdHeader;
+
+  public RequestTracing(TracingConfig config) {
+    TracingConfig c = config != null ? config : TracingConfig.defaults();
+    this.enabled = c.enabledOrDefault();
+    this.clientIdHeader = c.clientIdHeaderOrDefault();
+    this.requestIdHeader = c.requestIdHeaderOrDefault();
+  }
+
+  public boolean enabled() {
+    return enabled;
+  }
+
+  public String clientIdHeader() {
+    return clientIdHeader;
+  }
+
+  /** Client id from the request header, falling back to a stable fingerprint. */
+  public String resolveClientId(HttpServletRequest request, Session session) {
+    String fromHeader = request.getHeader(clientIdHeader);
+    if (!Strings.isNullOrEmpty(fromHeader)) {
+      return fromHeader;
+    }
+    return fingerprint(request, session);
+  }
+
+  /** Request id from the request header, falling back to a freshly generated UUID. */
+  public String resolveRequestId(HttpServletRequest request) {
+    String fromHeader = request.getHeader(requestIdHeader);
+    if (!Strings.isNullOrEmpty(fromHeader)) {
+      return fromHeader;
+    }
+    return UUID.randomUUID().toString();
+  }
+
+  public void begin(HttpServletResponse response, String clientId, String requestId) {
+    MDC.put(MDC_CLIENT_ID, clientId);
+    MDC.put(MDC_REQUEST_ID, requestId);
+    if (response != null && !response.isCommitted()) {
+      response.setHeader(clientIdHeader, clientId);
+      response.setHeader(requestIdHeader, requestId);
+    }
+  }
+
+  public void end() {
+    MDC.remove(MDC_CLIENT_ID);
+    MDC.remove(MDC_REQUEST_ID);
+  }
+
+  private static String fingerprint(HttpServletRequest request, Session session) {
+    String uid = session != null ? session.uid() : "anonymous";
+    String remote = request.getRemoteAddr();
+    String userAgent = request.getHeader("User-Agent");
+    String raw =
+        (uid != null ? uid : "")
+            + "|"
+            + (remote != null ? remote : "")
+            + "|"
+            + (userAgent != null ? userAgent : "");
+    CRC32 crc = new CRC32();
+    crc.update(raw.getBytes(StandardCharsets.UTF_8));
+    return "fp-" + Long.toHexString(crc.getValue());
+  }
+}