feat(client-v2): add span recorder SPI for operation and request tracing

CHANGELOG.md

@@ -4,6 +4,25 @@
 
 ### New Features
 
+- **[client-v2]** Added an observability SPI that lets an application observe client operations as spans.
+  `Client.Builder.setSpanRecorder(SpanRecorder)` registers a backend-agnostic recorder from the new
+  `com.clickhouse.client.api.observability` package: each operation (query, command, insert, `ping`,
+  `getTableSchema`) starts one operation span, and every transport request made for it - including each retry -
+  starts a child request span. `SpanRecorder` and `Span` are plain interfaces; an implementation extends the
+  `DefaultSpanRecorder` base class and overrides only the kinds of spans it cares about, so it keeps working when
+  the client starts a kind of span it does not know about. The operation's `QuerySettings`/`InsertSettings` is
+  passed to `startSpan(...)` so a recorder can take the properties it needs from it, and the reusable
+  `SpanSupport` class computes the attribute values. Span names and attribute keys follow the OpenTelemetry semantic conventions for
+  database and HTTP client spans; the keys are defined by the `SpanAttribute` enum and the values are computed by
+  the client, so all recorders report the same information (statement text, target database and table, query id,
+  statement parameters, batch size, the first configured endpoint on the operation span and the per-attempt
+  server address and port on the request spans, HTTP status, returned rows, and the error type and ClickHouse
+  error code on failure). An operation span is started on the calling thread, so it joins
+  the caller's ambient trace even when the operation runs on the client's executor, and it is ended exactly once. Previously the client exposed no hook for tracing, so an
+  application could not attribute a query or a retried request to its own trace. When no recorder is registered
+  nothing is recorded and no span-related work is done, so the default path is unchanged. An OpenTelemetry
+  implementation of the SPI follows in a separate module.
+  (https://github.com/ClickHouse/clickhouse-java/issues/2974)
 - **[client-v2, jdbc-v2]** Added support for the `BFloat16` data type (ClickHouse `24.11+`). `BFloat16` columns are read as
   Java `float` values (widening is lossless) and written from `float`/`Float` values, including through generic records, POJO
   binding, `Nullable(BFloat16)`, and `BFloat16` values held in `Dynamic`/`Variant` columns. On write the client keeps the
@@ -44,6 +63,10 @@
 
 ### Bug Fixes 
 
+- **[client-v2]** Fixed an insert unregistering its transport request after every attempt instead of once per
+  operation, so between two attempts of a retried insert `Client.cancelTransportRequest(String)` resolved the query
+  id to nothing at all. The registration now lives for the whole retry loop, as it already did for queries.
+  (https://github.com/ClickHouse/clickhouse-java/pull/2988)
 - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream
   returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial
   read. (https://github.com/ClickHouse/clickhouse-java/issues/2985)

client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java

@@ -14,6 +14,8 @@
 import com.clickhouse.client.api.enums.ProxyType;
 import com.clickhouse.client.api.enums.SSLMode;
 import com.clickhouse.client.api.http.ClickHouseHttpProto;
+import com.clickhouse.client.api.observability.Span;
+import com.clickhouse.client.api.observability.SpanSupport;
 import com.clickhouse.client.api.transport.Endpoint;
 import com.clickhouse.client.api.transport.internal.TransportRequest;
 import com.clickhouse.client.api.transport.internal.TransportResponse;
@@ -62,6 +64,7 @@
 import org.apache.hc.core5.http.protocol.HttpContext;
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.io.IOCallback;
+import org.apache.hc.core5.net.URIAuthority;
 import org.apache.hc.core5.net.URIBuilder;
 import org.apache.hc.core5.pool.ConnPoolControl;
 import org.apache.hc.core5.pool.PoolConcurrencyPolicy;
@@ -139,7 +142,15 @@ public class HttpAPIClientHelper {
 
     private final SslContextProvider sslContextProvider = new SslContextProvider();
 
-    public HttpAPIClientHelper(Map<String, Object> configuration, Object metricsRegistry, boolean initSslContext, LZ4Factory lz4Factory) {
+    /**
+     * Creates a span per transport request. Never {@code null} - disabled when observability is not
+     * configured.
+     */
+    private final SpanSupport spanSupport;
+
+    public HttpAPIClientHelper(Map<String, Object> configuration, Object metricsRegistry, boolean initSslContext,
+                               LZ4Factory lz4Factory, SpanSupport spanSupport) {
+        this.spanSupport = spanSupport == null ? SpanSupport.DISABLED : spanSupport;
         this.metricsRegistry = metricsRegistry;
         this.httpClient = createHttpClient(initSslContext, configuration);
         this.lz4Factory = lz4Factory;
@@ -686,6 +697,45 @@ public InputStream createDataInputStream() {
         }
     }
 
+    /**
+     * Executes a single transport request and records it as a child span of the given operation
+     * span, so a retried operation reports one request span per attempt. Recording happens around
+     * {@link #executeRequest(TransportRequest)}, which stays the single place where a request is
+     * actually executed.
+     *
+     * @param transportRequest - request to execute
+     * @param operationSpan - span of the operation this request is made for
+     * @return transport response
+     * @throws Exception when the request could not be completed
+     */
+    public TransportResponse executeRequest(TransportRequest transportRequest, Span operationSpan) throws Exception {
+        if (!spanSupport.isEnabled()) {
+            return executeRequest(transportRequest);
+        }
+
+        final Span requestSpan = startRequestSpan(operationSpan, transportRequest.getDelegate());
+        try {
+            TransportResponse response = executeRequest(transportRequest);
+            Object delegate = response.getDelegate();
+            if (delegate instanceof HttpResponse) {
+                spanSupport.recordHttpStatus(requestSpan, ((HttpResponse) delegate).getCode());
+            }
+            return response;
+        } catch (Exception e) {
+            spanSupport.recordRequestFailure(requestSpan, e);
+            throw e;
+        } finally {
+            requestSpan.end();
+        }
+    }
+
+    private Span startRequestSpan(Span operationSpan, HttpPost req) {
+        final URIAuthority authority = req.getAuthority();
+        return spanSupport.startRequestSpan(operationSpan,
+                authority == null ? null : authority.getHostName(),
+                authority == null ? -1 : authority.getPort());
+    }
+
     public TransportResponse executeRequest(TransportRequest transportRequest) throws Exception {
 
         final Map<String, Object> requestConfig = transportRequest.getConfig();

client-v2/src/main/java/com/clickhouse/client/api/observability/DefaultSpanRecorder.java

@@ -0,0 +1,67 @@
+package com.clickhouse.client.api.observability;
+
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.query.QuerySettings;
+
+/**
+ * Base class for {@link SpanRecorder} implementations. Every method returns {@link #NOOP_SPAN}, so
+ * a subclass overrides only the kinds of spans it wants to record and keeps working when the client
+ * starts a kind of span the subclass does not know about.
+ * <p>
+ * An instance of this class itself records nothing and is what the client uses when no recorder is
+ * registered.
+ */
+public class DefaultSpanRecorder implements SpanRecorder {
+
+    /**
+     * Span that records nothing. Returned by every method of this class and used by the client
+     * whenever there is nothing to record, so that a span reference is never {@code null}.
+     */
+    public static final Span NOOP_SPAN = new NoopSpan();
+
+    /**
+     * Shared instance that records nothing.
+     */
+    public static final DefaultSpanRecorder NOOP = new DefaultSpanRecorder();
+
+    @Override
+    public Span startSpan(String spanName, QuerySettings settings) {
+        return NOOP_SPAN;
+    }
+
+    @Override
+    public Span startSpan(String spanName, InsertSettings settings) {
+        return NOOP_SPAN;
+    }
+
+    @Override
+    public Span startRequestSpan(String spanName, Span operationSpan) {
+        return NOOP_SPAN;
+    }
+
+    /**
+     * Span implementation that discards everything reported to it.
+     */
+    private static final class NoopSpan implements Span {
+
+        @Override
+        public void setAttribute(String key, Object value) {
+            // records nothing
+        }
+
+        @Override
+        public void setError(String errorType) {
+            // records nothing
+        }
+
+        @Override
+        public void end() {
+            // records nothing
+        }
+
+        @Override
+        public String toString() {
+            return "NoopSpan";
+        }
+    }
+}

client-v2/src/main/java/com/clickhouse/client/api/observability/Span.java

@@ -0,0 +1,45 @@
+package com.clickhouse.client.api.observability;
+
+/**
+ * A single unit of work observed by a {@link SpanRecorder} - either a client operation or one
+ * transport request made for it.
+ * <p>
+ * An operation span covers sending the request and receiving the response head; it is ended when
+ * the operation hands its response to the caller, so it does not cover reading the response body
+ * (rows are streamed by the caller afterwards).
+ * <p>
+ * A recorder that does not record a given kind of span returns {@link DefaultSpanRecorder#NOOP_SPAN}
+ * instead, so the client never has to check for {@code null}.
+ * <p>
+ * A span is used by a single operation at a time, but the operation span and its request spans may
+ * be touched from different threads (an operation may run on the shared operation executor), so an
+ * implementation should not assume single-thread access.
+ */
+public interface Span {
+
+    /**
+     * Records an attribute. The keys the client uses are listed in {@link SpanAttribute}; there is a
+     * single attribute method so that an implementation cannot miss a value by overriding only one
+     * of several overloads.
+     *
+     * @param key - attribute key, see {@link SpanAttribute#getKey()}
+     * @param value - attribute value; a {@code String}, {@code Number} or {@code Boolean}, never
+     *              {@code null}
+     */
+    void setAttribute(String key, Object value);
+
+    /**
+     * Marks the span as failed and records {@link SpanAttribute#ERROR_TYPE} with the given value.
+     * An implementation should map this to its own failure status - for example an OpenTelemetry
+     * span status of {@code ERROR}.
+     *
+     * @param errorType - short, low-cardinality error identifier, usually an exception class name
+     */
+    void setError(String errorType);
+
+    /**
+     * Ends the span. Called exactly once by the client, also when the operation failed.
+     * An implementation should be idempotent, so that ending an already-ended span is harmless.
+     */
+    void end();
+}

client-v2/src/main/java/com/clickhouse/client/api/observability/SpanAttribute.java

@@ -0,0 +1,116 @@
+package com.clickhouse.client.api.observability;
+
+/**
+ * Attribute keys recorded on {@link Span}s by the client.
+ * <p>
+ * Keys follow the OpenTelemetry semantic conventions for database and HTTP client spans. They are
+ * defined here, on the SPI side, so that every {@link SpanRecorder} implementation reports the same
+ * key for the same piece of information.
+ */
+public enum SpanAttribute {
+
+    /**
+     * Database system name. Always {@code clickhouse}.
+     */
+    DB_SYSTEM_NAME("db.system.name"),
+
+    /**
+     * Target database name.
+     */
+    DB_NAMESPACE("db.namespace"),
+
+    /**
+     * Text of the statement sent to the server. Recorded for query and command operations.
+     */
+    DB_QUERY_TEXT("db.query.text"),
+
+    /**
+     * Table the operation targets. Recorded for insert and table-schema operations.
+     */
+    DB_COLLECTION_NAME("db.collection.name"),
+
+    /**
+     * Name of the client operation. Recorded when the operation is not a plain query - for example
+     * {@code insert}, {@code ping} or {@code getTableSchema}.
+     */
+    DB_OPERATION_NAME("db.operation.name"),
+
+    /**
+     * Number of items sent in a single batch. Recorded for POJO inserts.
+     */
+    DB_OPERATION_BATCH_SIZE("db.operation.batch.size"),
+
+    /**
+     * Prefix for statement parameter values. The full key is built with {@link #getKey(String)},
+     * for example {@code db.query.parameter.id}.
+     */
+    DB_QUERY_PARAMETER("db.query.parameter"),
+
+    /**
+     * ClickHouse error code returned by the server. Recorded when an operation fails.
+     */
+    DB_RESPONSE_STATUS_CODE("db.response.status_code"),
+
+    /**
+     * Number of rows returned by the server. Recorded when an operation succeeds and the server
+     * reported a progress summary.
+     */
+    DB_RESPONSE_RETURNED_ROWS("db.response.returned_rows"),
+
+    /**
+     * Query id of the operation, as assigned by the client or by the server.
+     */
+    CLICKHOUSE_QUERY_ID("clickhouse.query_id"),
+
+    /**
+     * Hostname of the server the request is sent to.
+     */
+    SERVER_ADDRESS("server.address"),
+
+    /**
+     * Port of the server the request is sent to.
+     */
+    SERVER_PORT("server.port"),
+
+    /**
+     * HTTP method of a transport request. Always {@code POST}.
+     */
+    HTTP_REQUEST_METHOD("http.request.method"),
+
+    /**
+     * HTTP status code returned for a transport request.
+     */
+    HTTP_RESPONSE_STATUS_CODE("http.response.status_code"),
+
+    /**
+     * Type of the error that made an operation or a request fail. Set with
+     * {@link Span#setError(String)}.
+     */
+    ERROR_TYPE("error.type");
+
+    private final String key;
+
+    SpanAttribute(String key) {
+        this.key = key;
+    }
+
+    /**
+     * Returns the attribute key.
+     *
+     * @return attribute key
+     */
+    public String getKey() {
+        return key;
+    }
+
+    /**
+     * Returns the attribute key for a named member of an attribute family - {@code key.suffix}.
+     * Used for {@link #DB_QUERY_PARAMETER}.
+     *
+     * @param suffix - name of the family member, for example a statement parameter name
+     * @return attribute key with the suffix appended
+     */
+    public String getKey(String suffix) {
+        return key + "." + suffix;
+    }
+}