Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ice-rest-catalog/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -598,6 +598,7 @@
<exclude>**/DockerScenarioBasedIT.java</exclude>
<exclude>**/DockerLocalFileIOClickHouseIT.java</exclude>
<exclude>**/DockerLocalFileIOClickHouseAllTypesIT.java</exclude>
<exclude>**/DockerSparkClientIdIT.java</exclude>
</excludes>
</configuration>
</plugin>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(),
Expand All @@ -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) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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"));
Expand All @@ -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(
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
/*
* 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} (grouping all calls from one Iceberg client) and a per-request
* {@code requestId} to logs (via MDC) and response headers, so a whole client session 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 a generated client id via the /v1/config response so Iceberg Java/PyIceberg clients echo it on every request (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;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* 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;

/**
* Advertises a per-client id to Iceberg clients via the {@code /v1/config} handshake.
*
* <p>The id 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 client instance share one {@code clientId}. Clients that ignore
* config overrides (e.g. ClickHouse) simply fall back to the server-side fingerprint.
*/
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) {
configResponse
.overrides()
.putIfAbsent(HEADER_PREFIX + clientIdHeader, RequestTracing.newClientId());
}
return restResponse;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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() + " ";
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
/*
* 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 Iceberg client instance; {@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 degrades gracefully:
*
* <ul>
* <li>If the request carries the client-id header (Iceberg Java / PyIceberg echo it once the
* server advertises it via the {@code /v1/config} response), that value is used.
* <li>Otherwise a stable fingerprint derived from the auth token, remote address and User-Agent
* is used. This covers clients (e.g. ClickHouse) that do not echo config headers.
* </ul>
*/
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;
}

/** A freshly generated client id, suitable for advertising via the config response. */
public static String newClientId() {
return UUID.randomUUID().toString();
}

/** 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());
}
}
Loading
Loading