feat(client-v2-otel): add OpenTelemetry span recorder module

CHANGELOG.md

@@ -4,6 +4,20 @@
 
 ### New Features
 
+- **[client-v2]** Added an OpenTelemetry implementation of the observability SPI.
+  `Client.Builder.setSpanRecorder(new OpenTelemetrySpanRecorder(openTelemetry))`
+  reports every client operation and every transport request as an OpenTelemetry `CLIENT` span: an operation span is
+  started as a child of the current OpenTelemetry context, so it joins the application's own trace, and each request
+  span - including one per retry - is a child of its operation span. Span names and attribute keys are the standard
+  ones of the SPI (the recorder derives them through `SpanSupport`), every value is recorded with the OpenTelemetry
+  attribute type that matches it, and a failure sets the span status to `ERROR` and is recorded as an OpenTelemetry
+  exception event next to the `error.type` and `db.response.status_code` attributes. The recorder reports to a
+  supplied `OpenTelemetry` instance, to a `Tracer` given to `new OpenTelemetrySpanRecorder(Tracer)`, or to
+  `GlobalOpenTelemetry` - read when a span is started - when constructed without arguments. Previously an application that wanted
+  OpenTelemetry spans had to write that mapping itself. The OpenTelemetry API is a compile-only dependency of
+  `client-v2`: the recorder is used only by an application that already provides `opentelemetry-api` at runtime, so
+  nothing is added to the classpath of a client that does not use it.
+  (https://github.com/ClickHouse/clickhouse-java/issues/2974)
 - **[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 (a query, a command or an insert - including
@@ -20,12 +34,19 @@
   `SpanSupport`, so all recorders that use it 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
+  error code on failure). The outcome of a completed operation is reported per operation kind - `recordQuerySuccess`
+  for a read and `recordInsertSuccess` for an insert - because the metrics that describe a read are not the ones that
+  describe a write: a query reports `db.response.returned_rows`, `clickhouse.response.read_rows` and
+  `clickhouse.response.read_bytes`, an insert reports `clickhouse.response.written_rows` and
+  `clickhouse.response.written_bytes`. The same distinction is available on the metrics themselves through the new
+  `OperationMetrics#getOperationType()`, which returns the new `com.clickhouse.client.api.metrics.OperationType` -
+  the kind of the call the application made, so a command that writes is reported as a query.
+  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
   for every operation that starts. 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.
+  implementation of the SPI is available as `OpenTelemetrySpanRecorder`.
   (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

client-v2/pom.xml

@@ -84,7 +84,28 @@
             <version>${guava.version}</version>
         </dependency>
 
+        <!-- Compile-only: OpenTelemetrySpanRecorder is used only when the
+             application already provides the OpenTelemetry API at runtime. -->
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+            <version>${opentelemetry.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- Test Dependencies -->
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+            <version>${opentelemetry.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk-testing</artifactId>
+            <version>${opentelemetry.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>

client-v2/src/main/java/com/clickhouse/client/api/Client.java

@@ -30,6 +30,7 @@
 import com.clickhouse.client.api.metadata.TableSchema;
 import com.clickhouse.client.api.metrics.ClientMetrics;
 import com.clickhouse.client.api.metrics.OperationMetrics;
+import com.clickhouse.client.api.metrics.OperationType;
 import com.clickhouse.client.api.observability.DefaultSpanRecorder;
 import com.clickhouse.client.api.observability.Span;
 import com.clickhouse.client.api.observability.SpanRecorder;
@@ -1531,9 +1532,10 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
 
                     try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest, operationSpan)) {
                         ClientStatisticsHolder clientStats = globalClientStats.remove(operationId);
-                        OperationMetrics metrics = completeOperation(transportResponse, clientStats, requestSettings.getQueryId());
+                        OperationMetrics metrics = completeOperation(transportResponse, clientStats,
+                                requestSettings.getQueryId(), OperationType.INSERT);
 
-                        spanRecorder.recordSuccess(operationSpan, metrics);
+                        spanRecorder.recordInsertSuccess(operationSpan, metrics);
                         return new InsertResponse(transportResponse, metrics);
                     } catch (Exception e) {
                         String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
@@ -1748,8 +1750,9 @@ public CompletableFuture<InsertResponse> insert(String tableName,
                     registerTransportReq(queryId, transportRequest);
 
                     try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest, operationSpan)) {
-                        OperationMetrics metrics = completeOperation(transportResponse, finalClientStats, requestSettings.getQueryId());
-                        spanRecorder.recordSuccess(operationSpan, metrics);
+                        OperationMetrics metrics = completeOperation(transportResponse, finalClientStats,
+                                requestSettings.getQueryId(), OperationType.INSERT);
+                        spanRecorder.recordInsertSuccess(operationSpan, metrics);
                         return new InsertResponse(transportResponse, metrics);
                     } catch (Exception e) {
                         String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
@@ -1891,13 +1894,14 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
                         TransportResponse transportResp = null;
                         try {
                             transportResp = httpClientHelper.executeRequest(request, operationSpan);
-                            OperationMetrics metrics = completeOperation(transportResp, clientStats, requestSettings.getQueryId());
+                            OperationMetrics metrics = completeOperation(transportResp, clientStats,
+                                    requestSettings.getQueryId(), OperationType.QUERY);
                             ClickHouseFormat responseFormat = transportResp.getDataFormat();
                             if (responseFormat == null) {
                                 responseFormat = requestSettings.getFormat();
                             }
 
-                            spanRecorder.recordSuccess(operationSpan, metrics);
+                            spanRecorder.recordQuerySuccess(operationSpan, metrics);
                             return new QueryResponse(transportResp, responseFormat, requestSettings, metrics);
 
                         } catch (Exception e) {
@@ -1995,8 +1999,9 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
         return query(sqlQuery, queryParams, null);
     }
 
-    private OperationMetrics completeOperation(TransportResponse transportResponse, ClientStatisticsHolder clientStats, String originalQueryId) {
-        OperationMetrics metrics = new OperationMetrics(clientStats);
+    private OperationMetrics completeOperation(TransportResponse transportResponse, ClientStatisticsHolder clientStats,
+                                               String originalQueryId, OperationType operationType) {
+        OperationMetrics metrics = new OperationMetrics(clientStats, operationType);
         String summary = transportResponse.getSummaryJson();
         ProcessParser.parseSummary(summary, metrics);
         String queryId = transportResponse.getQueryId();

client-v2/src/main/java/com/clickhouse/client/api/metrics/OperationMetrics.java

@@ -19,8 +19,39 @@ public class OperationMetrics {
 
     private final ClientStatisticsHolder clientStatistics;
 
+    private final OperationType operationType;
+
+    /**
+     * Creates metrics of an operation whose kind is not known - their
+     * {@link #getOperationType()} is {@link OperationType#UNKNOWN}. The client always reports the
+     * kind of the operation, so this constructor is only for metrics created by user code.
+     *
+     * @param clientStatisticsHolder - holder of the client-side statistics of the operation
+     */
     public OperationMetrics(ClientStatisticsHolder clientStatisticsHolder) {
+        this(clientStatisticsHolder, OperationType.UNKNOWN);
+    }
+
+    /**
+     * Creates metrics of an operation of the given kind.
+     *
+     * @param clientStatisticsHolder - holder of the client-side statistics of the operation
+     * @param operationType - kind of the operation; {@code null} is read as {@link OperationType#UNKNOWN}
+     */
+    public OperationMetrics(ClientStatisticsHolder clientStatisticsHolder, OperationType operationType) {
         this.clientStatistics = clientStatisticsHolder;
+        this.operationType = operationType == null ? OperationType.UNKNOWN : operationType;
+    }
+
+    /**
+     * Returns the kind of the operation these metrics were collected for. It tells which of the
+     * metrics are meaningful - a read operation reports what the server read and returned, an insert
+     * reports what it wrote.
+     *
+     * @return kind of the operation; never {@code null}
+     */
+    public OperationType getOperationType() {
+        return operationType;
     }
 
     public Metric getMetric(ServerMetrics metric) {
@@ -59,6 +90,7 @@ public void setQueryId(String queryId) {
     public String toString() {
         return "OperationStatistics{" +
                 "\"queryId\"=\"" + queryId + "\", " +
+                "\"operationType\"=\"" + operationType + "\", " +
                 "\"metrics\"=" + metrics +
                 '}';
     }

client-v2/src/main/java/com/clickhouse/client/api/metrics/OperationType.java

@@ -0,0 +1,28 @@
+package com.clickhouse.client.api.metrics;
+
+/**
+ * Kind of client operation a set of {@link OperationMetrics} was collected for.
+ * <p>
+ * The kind decides which metrics of the operation are meaningful: a read operation reports how much
+ * the server read and returned, an insert reports how much the server wrote.
+ */
+public enum OperationType {
+
+    /**
+     * Operation the client ran as a statement - a query, a command, a ping or a table-schema lookup.
+     * It is the kind of the call the application made, not of the work the server did: a command that
+     * writes, such as {@code INSERT INTO ... SELECT}, is run as a statement and is reported here.
+     */
+    QUERY,
+
+    /**
+     * Operation the client ran as an insert, through one of the {@code insert} methods.
+     */
+    INSERT,
+
+    /**
+     * Kind of the operation is not known. Reported for metrics that were created without one, which
+     * the client itself never does.
+     */
+    UNKNOWN
+}

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

@@ -71,7 +71,12 @@ public void recordHttpStatus(Span requestSpan, int statusCode) {
     }
 
     @Override
-    public void recordSuccess(Span operationSpan, OperationMetrics metrics) {
+    public void recordQuerySuccess(Span operationSpan, OperationMetrics metrics) {
+        // records nothing
+    }
+
+    @Override
+    public void recordInsertSuccess(Span operationSpan, OperationMetrics metrics) {
         // records nothing
     }
 

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

@@ -52,7 +52,7 @@ public enum SpanAttribute {
     DB_RESPONSE_STATUS_CODE("db.response.status_code"),
 
     /**
-     * Number of rows returned by the server. Recorded when an operation succeeds and the server
+     * Number of rows returned by the server. Recorded when a read operation succeeds and the server
      * reported a progress summary.
      */
     DB_RESPONSE_RETURNED_ROWS("db.response.returned_rows"),
@@ -62,6 +62,30 @@ public enum SpanAttribute {
      */
     CLICKHOUSE_QUERY_ID("clickhouse.query_id"),
 
+    /**
+     * Number of rows the server read from the storage. Recorded when a read operation succeeds and
+     * the server reported a progress summary.
+     */
+    CLICKHOUSE_RESPONSE_READ_ROWS("clickhouse.response.read_rows"),
+
+    /**
+     * Number of bytes the server read from the storage. Recorded when a read operation succeeds and
+     * the server reported a progress summary.
+     */
+    CLICKHOUSE_RESPONSE_READ_BYTES("clickhouse.response.read_bytes"),
+
+    /**
+     * Number of rows the server wrote to the storage. Recorded when an insert succeeds and the server
+     * reported a progress summary.
+     */
+    CLICKHOUSE_RESPONSE_WRITTEN_ROWS("clickhouse.response.written_rows"),
+
+    /**
+     * Number of bytes the server wrote to the storage. Recorded when an insert succeeds and the
+     * server reported a progress summary.
+     */
+    CLICKHOUSE_RESPONSE_WRITTEN_BYTES("clickhouse.response.written_bytes"),
+
     /**
      * Hostname of the server the request is sent to.
      */

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

@@ -2,6 +2,7 @@
 
 import com.clickhouse.client.api.insert.InsertSettings;
 import com.clickhouse.client.api.metrics.OperationMetrics;
+import com.clickhouse.client.api.metrics.OperationType;
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.client.api.transport.Endpoint;
 
@@ -93,13 +94,27 @@ public interface SpanRecorder {
     void recordHttpStatus(Span requestSpan, int statusCode);
 
     /**
-     * Reports that an operation completed successfully.
+     * Reports that a read operation completed successfully. It is the counterpart of
+     * {@link #startQuerySpan(QuerySettings, String, Endpoint)}.
      *
      * @param operationSpan - span of the operation
-     * @param metrics - metrics of the completed operation; source of the query id and of the number
-     *                of returned rows. May be {@code null}
+     * @param metrics - metrics of the completed operation, whose
+     *                {@link OperationMetrics#getOperationType()} is {@link OperationType#QUERY};
+     *                source of the query id and of what the server read and returned. May be
+     *                {@code null}
      */
-    void recordSuccess(Span operationSpan, OperationMetrics metrics);
+    void recordQuerySuccess(Span operationSpan, OperationMetrics metrics);
+
+    /**
+     * Reports that an insert operation completed successfully. It is the counterpart of
+     * {@link #startInsertSpan(InsertSettings, String, int, Endpoint)}.
+     *
+     * @param operationSpan - span of the operation
+     * @param metrics - metrics of the completed operation, whose
+     *                {@link OperationMetrics#getOperationType()} is {@link OperationType#INSERT};
+     *                source of the query id and of what the server wrote. May be {@code null}
+     */
+    void recordInsertSuccess(Span operationSpan, OperationMetrics metrics);
 
     /**
      * Reports that an operation failed.

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

@@ -156,23 +156,67 @@ public void recordEndpoint(Span span, String host, int port) {
     }
 
     /**
-     * Records the outcome of a successfully completed operation.
+     * Records the outcome of a successfully completed read operation - what the server read and what
+     * it returned.
      *
      * @param span - span of the operation
      * @param metrics - metrics of the completed operation, may be {@code null}
      */
-    public void recordSuccess(Span span, OperationMetrics metrics) {
+    public void recordQuerySuccess(Span span, OperationMetrics metrics) {
         if (metrics == null) {
             return;
         }
 
+        recordQueryId(span, metrics);
+        recordServerMetric(span, metrics, ServerMetrics.RESULT_ROWS, SpanAttribute.DB_RESPONSE_RETURNED_ROWS);
+        recordServerMetric(span, metrics, ServerMetrics.NUM_ROWS_READ, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS);
+        recordServerMetric(span, metrics, ServerMetrics.NUM_BYTES_READ, SpanAttribute.CLICKHOUSE_RESPONSE_READ_BYTES);
+    }
+
+    /**
+     * Records the outcome of a successfully completed insert operation - what the server wrote.
+     *
+     * @param span - span of the operation
+     * @param metrics - metrics of the completed operation, may be {@code null}
+     */
+    public void recordInsertSuccess(Span span, OperationMetrics metrics) {
+        if (metrics == null) {
+            return;
+        }
+
+        recordQueryId(span, metrics);
+        recordServerMetric(span, metrics, ServerMetrics.NUM_ROWS_WRITTEN,
+                SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS);
+        recordServerMetric(span, metrics, ServerMetrics.NUM_BYTES_WRITTEN,
+                SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES);
+    }
+
+    /**
+     * Records the query id of a completed operation, which the server may have assigned itself.
+     *
+     * @param span - span of the operation
+     * @param metrics - metrics of the completed operation
+     */
+    protected void recordQueryId(Span span, OperationMetrics metrics) {
         if (metrics.getQueryId() != null) {
             span.setAttribute(SpanAttribute.CLICKHOUSE_QUERY_ID.getKey(), metrics.getQueryId());
         }
-        // the row count comes from the server's progress summary, which is not always available
-        Metric returnedRows = metrics.getMetric(ServerMetrics.RESULT_ROWS);
-        if (returnedRows != null && returnedRows.getLong() >= 0) {
-            span.setAttribute(SpanAttribute.DB_RESPONSE_RETURNED_ROWS.getKey(), returnedRows.getLong());
+    }
+
+    /**
+     * Records one server metric of a completed operation. The value comes from the server's progress
+     * summary, which is not always available, so a metric the server did not report is left out.
+     *
+     * @param span - span of the operation
+     * @param metrics - metrics of the completed operation
+     * @param metric - server metric to read
+     * @param attribute - attribute to record it under
+     */
+    protected void recordServerMetric(Span span, OperationMetrics metrics, ServerMetrics metric,
+                                      SpanAttribute attribute) {
+        Metric value = metrics.getMetric(metric);
+        if (value != null && value.getLong() >= 0) {
+            span.setAttribute(attribute.getKey(), value.getLong());
         }
     }