feat(client-v2,jdbc-v2): add metrics recorder SPI

CHANGELOG.md

@@ -12,6 +12,23 @@
 
 ### New Features
 
+- **[client-v2, jdbc-v2]** Added a metrics SPI that lets an application export the metrics of client operations to any
+  metrics backend. `Client.Builder.setMetricsRecorder(MetricsRecorder)` registers a backend-agnostic recorder from the
+  `com.clickhouse.client.api.observability` package, and the jdbc-v2 property `jdbc_metrics_recorder` names the recorder
+  class a connection registers with its own client. Previously the client collected operation metrics but only returned
+  them to the caller, so exporting them was left to the application. Each completed operation reports exactly one
+  success or one failure event, and each retried attempt reports a retry event, which gives the operation duration, the
+  serialization duration, the number of operations by outcome and the number of retries. The SPI follows the pattern of
+  the span SPI: an implementation extends the `DefaultMetricsRecorder` base class and overrides only what it cares
+  about, so it keeps working when the client starts reporting an event it does not know about, and the reusable
+  `MetricsSupport` class derives the standard values from the same structures, so its logic is opt-in and overridable.
+  Metric names, units and attribute keys follow the OpenTelemetry semantic conventions for database clients where a
+  convention exists and are placed under `clickhouse.` where it does not; they are defined by the `MetricName` and
+  `MetricAttribute` enums, durations are reported in seconds, and a duration the client did not measure is reported as
+  `MetricsSupport.DURATION_UNKNOWN` instead of a made-up value. The metric attributes are deliberately a smaller set
+  than the span attributes, because an attribute of a metric becomes a time series: the statement text, the query id and
+  the statement parameters stay on spans. Nothing is recorded and no metrics-related work is done when no recorder is
+  registered. (https://github.com/ClickHouse/clickhouse-java/issues/2975)
 - **[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

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

@@ -31,7 +31,9 @@
 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.DefaultMetricsRecorder;
 import com.clickhouse.client.api.observability.DefaultSpanRecorder;
+import com.clickhouse.client.api.observability.MetricsRecorder;
 import com.clickhouse.client.api.observability.Span;
 import com.clickhouse.client.api.observability.SpanRecorder;
 import com.clickhouse.client.api.query.GenericRecord;
@@ -167,17 +169,27 @@ public class Client implements AutoCloseable {
      */
     private final SpanRecorder spanRecorder;
 
+    /**
+     * Recorder registered by an application; called once for every operation the client completes,
+     * with everything the client knows about it. Never {@code null} - it is
+     * {@link DefaultMetricsRecorder#NOOP} when observability is not configured, so no null check is
+     * needed on the operation paths.
+     */
+    private final MetricsRecorder metricsRecorder;
+
     private Client(Collection<Endpoint> endpoints, Map<String,String> configuration,
                    ExecutorService sharedOperationExecutor, ColumnToMethodMatchingStrategy columnToMethodMatchingStrategy,
                    Object metricsRegistry, Supplier<String> queryIdGenerator, CredentialsManager cManager,
-                   SSLContext sslContext, SpanRecorder spanRecorder) {
+                   SSLContext sslContext, SpanRecorder spanRecorder, MetricsRecorder metricsRecorder) {
         Map<String, Object> parsedConfiguration = new ConcurrentHashMap<>(ClientConfigProperties.parseConfigMap(configuration));
         if (sslContext != null) {
             parsedConfiguration.put(ClientConfigProperties.SSL_CONTEXT.getKey(), sslContext);
         }
         this.credentialsManager = cManager;
         this.spanRecorder = Objects.requireNonNull(spanRecorder,
                 "spanRecorder is required; use DefaultSpanRecorder.NOOP to record nothing");
+        this.metricsRecorder = Objects.requireNonNull(metricsRecorder,
+                "metricsRecorder is required; use DefaultMetricsRecorder.NOOP to record nothing");
         this.session = Session.extractFrom(parsedConfiguration);
         this.configuration = new ConcurrentHashMap<>(parsedConfiguration);
         this.readOnlyConfig = Collections.unmodifiableMap(configuration);
@@ -299,6 +311,7 @@ public static class Builder {
         private Supplier<String> queryIdGenerator;
         private SSLContext sslContext = null;
         private SpanRecorder spanRecorder = DefaultSpanRecorder.NOOP;
+        private MetricsRecorder metricsRecorder = DefaultMetricsRecorder.NOOP;
 
         // Trust/key material options that feed a context the client would otherwise build; none of them
         // may be combined with an application-supplied SSLContext (see build()).
@@ -1267,6 +1280,27 @@ public Builder setSpanRecorder(SpanRecorder spanRecorder) {
             return this;
         }
 
+        /**
+         * <p>Registers a {@link MetricsRecorder} that receives the metrics of client operations, so
+         * an application can export them to any metrics backend. Each completed operation (query,
+         * command, insert, ping, table-schema lookup) reports one success or one failure event, and
+         * every retried attempt reports a retry event.</p>
+         *
+         * <p>When no recorder is set nothing is recorded and no metrics-related work is done. The
+         * default is {@link DefaultMetricsRecorder#NOOP}, so registering that recorder is how an
+         * application asks for nothing to be recorded; {@code null} is rejected because it is a
+         * configuration error rather than a way to disable recording.</p>
+         *
+         * @param metricsRecorder - recorder to notify; must not be {@code null}
+         * @return same instance of the builder
+         * @throws NullPointerException when {@code metricsRecorder} is {@code null}
+         */
+        public Builder setMetricsRecorder(MetricsRecorder metricsRecorder) {
+            this.metricsRecorder = Objects.requireNonNull(metricsRecorder,
+                    "metricsRecorder is required; use DefaultMetricsRecorder.NOOP to record nothing");
+            return this;
+        }
+
         public Client build() {
             // check if endpoint are empty. so can not initiate client
             if (this.endpoints.isEmpty()) {
@@ -1354,7 +1388,7 @@ public Client build() {
 
             return new Client(this.endpoints, this.configuration, this.sharedOperationExecutor,
                 this.columnToMethodMatchingStrategy, this.metricRegistry, this.queryIdGenerator, cManager,
-                this.sslContext, this.spanRecorder);
+                this.sslContext, this.spanRecorder, this.metricsRecorder);
         }
     }
 
@@ -1463,6 +1497,9 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
 
         String operationId = registerOperationMetrics();
         requestSettings.setOperationId(operationId);
+        // Origin of the duration of a failed operation. Taken where the client starts OP_DURATION, which is
+        // the duration reported for a successful operation, so that both outcomes measure the same work.
+        final long operationStartNanos = System.nanoTime();
         globalClientStats.get(operationId).start(ClientMetrics.OP_DURATION);
         globalClientStats.get(operationId).start(ClientMetrics.OP_SERIALIZATION);
 
@@ -1537,12 +1574,14 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
                                 requestSettings.getQueryId(), OperationType.INSERT);
 
                         spanRecorder.recordInsertSuccess(operationSpan, metrics);
+                        metricsRecorder.recordInsertSuccess(requestSettings, tableName, metrics);
                         return new InsertResponse(transportResponse, metrics);
                     } catch (Exception e) {
                         String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
                         lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
                         if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(queryId)) {
                             if (i < maxAttempts) {
+                                metricsRecorder.recordInsertRetry(requestSettings, tableName, lastException);
                                 selectedEndpoint = logRetryAndSelectNextNode("Insert", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e);
                             } else {
                                 nodeSelector.getNextAliveNode(selectedEndpoint);
@@ -1557,7 +1596,11 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
                 LOG.warn(errMsg);
                 throw (lastException == null ? new ClientException(errMsg) : lastException);
             } catch (RuntimeException | Error e) {
+                // Taken before any recorder runs, like the duration of a successful operation, which the
+                // client stops in completeOperation.
+                final Duration failureDuration = durationSince(operationStartNanos);
                 spanRecorder.recordFailure(operationSpan, e);
+                metricsRecorder.recordInsertFailure(requestSettings, tableName, failureDuration, e);
                 throw e;
             } finally {
                 // The request of the last attempt stays registered until the operation is over, so a cancellation
@@ -1700,6 +1743,9 @@ public CompletableFuture<InsertResponse> insert(String tableName,
         if (clientStats == null) {
             clientStats = new ClientStatisticsHolder();
         }
+        // Origin of the duration of a failed operation. Taken where the client starts OP_DURATION, which is
+        // the duration reported for a successful operation, so that both outcomes measure the same work.
+        final long operationStartNanos = System.nanoTime();
         clientStats.start(ClientMetrics.OP_DURATION);
         final ClientStatisticsHolder finalClientStats = clientStats;
 
@@ -1754,12 +1800,14 @@ public CompletableFuture<InsertResponse> insert(String tableName,
                         OperationMetrics metrics = completeOperation(transportResponse, finalClientStats,
                                 requestSettings.getQueryId(), OperationType.INSERT);
                         spanRecorder.recordInsertSuccess(operationSpan, metrics);
+                        metricsRecorder.recordInsertSuccess(requestSettings, tableName, metrics);
                         return new InsertResponse(transportResponse, metrics);
                     } catch (Exception e) {
                         String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
                         lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
                         if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(requestSettings.getQueryId())) {
                             if (i < maxAttempts) {
+                                metricsRecorder.recordInsertRetry(requestSettings, tableName, lastException);
                                 selectedEndpoint = logRetryAndSelectNextNode("Insert (stream)", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e);
                             } else {
                                 nodeSelector.getNextAliveNode(selectedEndpoint);
@@ -1782,7 +1830,11 @@ public CompletableFuture<InsertResponse> insert(String tableName,
                 LOG.warn(errMsg);
                 throw (lastException == null ? new ClientException(errMsg) : lastException);
             } catch (RuntimeException | Error e) {
+                // Taken before any recorder runs, like the duration of a successful operation, which the
+                // client stops in completeOperation.
+                final Duration failureDuration = durationSince(operationStartNanos);
                 spanRecorder.recordFailure(operationSpan, e);
+                metricsRecorder.recordInsertFailure(requestSettings, tableName, failureDuration, e);
                 throw e;
             } finally {
                 // The request of the last attempt stays registered until the operation is over, so a cancellation
@@ -1861,6 +1913,9 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
         }
         applyFormatSpecificSettings(requestSettings);
         ClientStatisticsHolder clientStats = new ClientStatisticsHolder();
+        // Origin of the duration of a failed operation. Taken where the client starts OP_DURATION, which is
+        // the duration reported for a successful operation, so that both outcomes measure the same work.
+        final long operationStartNanos = System.nanoTime();
         clientStats.start(ClientMetrics.OP_DURATION);
 
         if (queryParams != null) {
@@ -1906,6 +1961,7 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
                             }
 
                             spanRecorder.recordQuerySuccess(operationSpan, metrics);
+                            metricsRecorder.recordQuerySuccess(requestSettings, metrics);
                             return new QueryResponse(transportResp, responseFormat, requestSettings, metrics);
 
                         } catch (Exception e) {
@@ -1914,6 +1970,7 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
                             lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
                             if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(requestSettings.getQueryId())) {
                                 if (i < maxAttempts) {
+                                    metricsRecorder.recordQueryRetry(requestSettings, lastException);
                                     selectedEndpoint = logRetryAndSelectNextNode("Query", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e);
                                 } else {
                                     nodeSelector.getNextAliveNode(selectedEndpoint);
@@ -1928,7 +1985,11 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
                     LOG.warn(errMsg);
                     throw (lastException == null ? new ClientException(errMsg) : lastException);
                 } catch (RuntimeException | Error e) {
+                    // Taken before any recorder runs, like the duration of a successful operation, which the
+                    // client stops in completeOperation.
+                    final Duration failureDuration = durationSince(operationStartNanos);
                     spanRecorder.recordFailure(operationSpan, e);
+                    metricsRecorder.recordQueryFailure(requestSettings, failureDuration, e);
                     throw e;
                 } finally {
                     // unregister transport request once we are done

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

@@ -0,0 +1,75 @@
+package com.clickhouse.client.api.observability;
+
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.metrics.OperationMetrics;
+import com.clickhouse.client.api.query.QuerySettings;
+
+import java.time.Duration;
+
+/**
+ * Base class for {@link MetricsRecorder} implementations. Every method records nothing, so a
+ * subclass overrides only what it wants to record and keeps working when the client starts reporting
+ * an event the subclass does not know about.
+ * <p>
+ * A subclass owns its instruments and decides what to report. To use the client's standard metric
+ * names, units and attributes it can hand the structures it is given to {@link #getMetricsSupport()}:
+ * <pre>{@code
+ * public void recordQuerySuccess(QuerySettings settings, OperationMetrics metrics) {
+ *     MetricsSupport support = getMetricsSupport();
+ *     myHistogram.record(support.operationDuration(metrics), support.queryAttributes(settings, null));
+ * }
+ * }</pre>
+ * Using it is optional - a recorder that reports something else, or in another form, ignores it, and
+ * one that wants other values overrides {@link #getMetricsSupport()} with its own subclass of
+ * {@link MetricsSupport}.
+ * <p>
+ * An instance of this class itself records nothing and is what the client uses when no recorder is
+ * registered.
+ */
+public class DefaultMetricsRecorder implements MetricsRecorder {
+
+    /**
+     * Shared instance that records nothing.
+     */
+    public static final DefaultMetricsRecorder NOOP = new DefaultMetricsRecorder();
+
+    /**
+     * Returns the helper a subclass can use to derive the client's standard metric names, units and
+     * attributes. Override to report other values.
+     *
+     * @return metrics support; never {@code null}
+     */
+    protected MetricsSupport getMetricsSupport() {
+        return MetricsSupport.DEFAULT;
+    }
+
+    @Override
+    public void recordQuerySuccess(QuerySettings settings, OperationMetrics metrics) {
+        // records nothing
+    }
+
+    @Override
+    public void recordInsertSuccess(InsertSettings settings, String tableName, OperationMetrics metrics) {
+        // records nothing
+    }
+
+    @Override
+    public void recordQueryFailure(QuerySettings settings, Duration duration, Throwable t) {
+        // records nothing
+    }
+
+    @Override
+    public void recordInsertFailure(InsertSettings settings, String tableName, Duration duration, Throwable t) {
+        // records nothing
+    }
+
+    @Override
+    public void recordQueryRetry(QuerySettings settings, Throwable cause) {
+        // records nothing
+    }
+
+    @Override
+    public void recordInsertRetry(InsertSettings settings, String tableName, Throwable cause) {
+        // records nothing
+    }
+}

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

@@ -0,0 +1,62 @@
+package com.clickhouse.client.api.observability;
+
+/**
+ * Attribute keys recorded on the metrics of {@link MetricName} by the client.
+ * <p>
+ * Keys follow the OpenTelemetry semantic conventions for database clients. They are defined here, on
+ * the SPI side, so that every {@link MetricsRecorder} implementation reports the same key for the
+ * same piece of information.
+ * <p>
+ * This is deliberately a smaller set than {@link SpanAttribute}: an attribute of a metric becomes a
+ * time series, so only low-cardinality values are reported. The statement text, the query id and the
+ * statement parameters are recorded on spans only.
+ */
+public enum MetricAttribute {
+
+    /**
+     * Database system name. Always {@code clickhouse}.
+     */
+    DB_SYSTEM_NAME("db.system.name"),
+
+    /**
+     * Target database name.
+     */
+    DB_NAMESPACE("db.namespace"),
+
+    /**
+     * Name of the client operation - {@code query} or {@code insert}.
+     */
+    DB_OPERATION_NAME("db.operation.name"),
+
+    /**
+     * Table the operation targets. Recorded for an insert.
+     */
+    DB_COLLECTION_NAME("db.collection.name"),
+
+    /**
+     * ClickHouse error code returned by the server. Recorded when an operation fails and the server
+     * reported one.
+     */
+    DB_RESPONSE_STATUS_CODE("db.response.status_code"),
+
+    /**
+     * Type of the error that made an operation fail, usually an exception class name. Recorded only
+     * on a failure, so a time series without it is the successful one.
+     */
+    ERROR_TYPE("error.type");
+
+    private final String key;
+
+    MetricAttribute(String key) {
+        this.key = key;
+    }
+
+    /**
+     * Returns the attribute key.
+     *
+     * @return attribute key
+     */
+    public String getKey() {
+        return key;
+    }
+}