diff --git a/CHANGELOG.md b/CHANGELOG.md index 6efa8ac4c..cab483129 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/client-v2/src/main/java/com/clickhouse/client/api/Client.java b/client-v2/src/main/java/com/clickhouse/client/api/Client.java index 894f12e91..dbe29a268 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/Client.java +++ b/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,10 +169,18 @@ 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 endpoints, Map configuration, ExecutorService sharedOperationExecutor, ColumnToMethodMatchingStrategy columnToMethodMatchingStrategy, Object metricsRegistry, Supplier queryIdGenerator, CredentialsManager cManager, - SSLContext sslContext, SpanRecorder spanRecorder) { + SSLContext sslContext, SpanRecorder spanRecorder, MetricsRecorder metricsRecorder) { Map parsedConfiguration = new ConcurrentHashMap<>(ClientConfigProperties.parseConfigMap(configuration)); if (sslContext != null) { parsedConfiguration.put(ClientConfigProperties.SSL_CONTEXT.getKey(), sslContext); @@ -178,6 +188,8 @@ private Client(Collection endpoints, Map configuration, 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 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; } + /** + *

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.

+ * + *

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.

+ * + * @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 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 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 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 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 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 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 query(String sqlQuery, Map query(String sqlQuery, Map query(String sqlQuery, Map query(String sqlQuery, Map + * 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()}: + * 
{@code
+ * public void recordQuerySuccess(QuerySettings settings, OperationMetrics metrics) {
+ *     MetricsSupport support = getMetricsSupport();
+ *     myHistogram.record(support.operationDuration(metrics), support.queryAttributes(settings, null));
+ * }
+ * }
+ * 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}. + *

+ * 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 + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricAttribute.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricAttribute.java new file mode 100644 index 000000000..c1ca3041f --- /dev/null +++ b/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. + *

+ * 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. + *

+ * 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; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricName.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricName.java new file mode 100644 index 000000000..c4b58ac28 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricName.java @@ -0,0 +1,83 @@ +package com.clickhouse.client.api.observability; + +/** + * Metrics reported by the client, with the name, the unit and the description an exporter should + * register its instrument with. + *

+ * Names and units follow the OpenTelemetry semantic conventions for database clients where a + * convention exists, and are placed under {@code clickhouse.} where it does not - the same rule + * {@link SpanAttribute} follows. Units are the UCUM codes the conventions use, so a duration is + * reported in seconds (the values of {@link com.clickhouse.client.api.metrics.ClientMetrics} + * are milliseconds; {@link MetricsSupport} converts them). + *

+ * The names are defined here, on the SPI side, so that every {@link MetricsRecorder} implementation + * reports the same metric under the same name. + */ +public enum MetricName { + + /** + * Duration of a client operation. Reported for a successful and for a failed operation alike, so + * the number of operations by outcome is the count of this metric grouped by + * {@link MetricAttribute#ERROR_TYPE}. + */ + OPERATION_DURATION("db.client.operation.duration", "s", "Duration of a ClickHouse client operation."), + + /** + * Duration of the serialization step of a client operation. Reported when the client measured it, + * which it does for an insert of POJOs. + */ + OPERATION_SERIALIZATION_DURATION("clickhouse.client.operation.serialization.duration", "s", + "Duration of the serialization step of a ClickHouse client operation."), + + /** + * Number of completed client operations. Counted by outcome - a successful operation has no + * {@link MetricAttribute#ERROR_TYPE} attribute, a failed one carries it. + */ + OPERATION_COUNT("clickhouse.client.operation.count", "{operation}", + "Number of completed ClickHouse client operations, by outcome."), + + /** + * Number of retried attempts of client operations. An operation that succeeds on its third + * attempt contributes two retries. + */ + OPERATION_RETRIES("clickhouse.client.operation.retries", "{retry}", + "Number of retried attempts of ClickHouse client operations."); + + private final String key; + private final String unit; + private final String description; + + MetricName(String key, String unit, String description) { + this.key = key; + this.unit = unit; + this.description = description; + } + + /** + * Returns the name of the metric. + * + * @return metric name + */ + public String getKey() { + return key; + } + + /** + * Returns the unit of the metric as a UCUM code - {@code s} for a duration, an annotation like + * {@code {operation}} for a count. + * + * @return metric unit + */ + public String getUnit() { + return unit; + } + + /** + * Returns the description of the metric. + * + * @return metric description + */ + public String getDescription() { + return description; + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricsRecorder.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricsRecorder.java new file mode 100644 index 000000000..12bb4c950 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricsRecorder.java @@ -0,0 +1,110 @@ +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; + +/** + * Backend-agnostic hook that lets an application export the metrics of client operations. + *

+ * A recorder is registered with + * {@link com.clickhouse.client.api.Client.Builder#setMetricsRecorder(MetricsRecorder)}. It is called + * once per completed operation, with everything the client knows about it, so an implementation is + * free to report whatever it needs and however it wants - including nothing. + *

+ * Three kinds of events are reported: + *

    + *
  • Success - {@link #recordQuerySuccess(QuerySettings, OperationMetrics)} for a read + * operation and {@link #recordInsertSuccess(InsertSettings, String, OperationMetrics)} for an + * insert. The durations of the operation are read from the metrics of the completed + * operation.
    • + *
  • Failure - {@link #recordQueryFailure(QuerySettings, Duration, Throwable)} and + * {@link #recordInsertFailure(InsertSettings, String, Duration, Throwable)}. An operation that + * failed has no metrics, so the client measures its duration itself and passes it in. It measures + * the same work a successful operation reports - from where it starts the operation until the + * outcome is known, before any recorder runs - so that both outcomes form one latency series.
    • + *
  • Retry - {@link #recordQueryRetry(QuerySettings, Throwable)} and + * {@link #recordInsertRetry(InsertSettings, String, Throwable)}, called once per retried + * attempt. The operation itself may still succeed.
    • + *
+ * Exactly one success or one failure event is reported per operation the client started, so counting + * those events gives the number of operations by outcome. + *

+ * An implementation does not have to derive the standard metric names, units and attributes itself: + * it may call {@link MetricsSupport}, which computes them - the names listed in {@link MetricName} + * and the keys listed in {@link MetricAttribute} - from the same structures. That is opt-in; a + * recorder that wants to report something else, or in another form, simply does not use it. + *

+ * Implementations should extend {@link DefaultMetricsRecorder} and override only what they care + * about; the inherited methods record nothing, so a recorder keeps working when the client starts + * reporting an event it does not know about. + *

+ * A recorder is shared by all operations of a client instance and must be thread-safe. It is called on + * the thread that runs the operation and must not throw and not block, because a failure of a recorder + * is a failure of the operation it reports. + */ +public interface MetricsRecorder { + + /** + * Reports that a read operation - a query, a command, a ping or a table-schema lookup - + * completed successfully. + * + * @param settings - resolved settings of the operation; source of the target database + * @param metrics - metrics of the completed operation; source of the operation duration and of + * what the server read and returned. May be {@code null} + */ + void recordQuerySuccess(QuerySettings settings, OperationMetrics metrics); + + /** + * Reports that an insert operation completed successfully. + * + * @param settings - resolved settings of the operation; source of the target database + * @param tableName - target table + * @param metrics - metrics of the completed operation; source of the operation and serialization + * durations and of what the server wrote. May be {@code null} + */ + void recordInsertSuccess(InsertSettings settings, String tableName, OperationMetrics metrics); + + /** + * Reports that a read operation failed. It is the counterpart of + * {@link #recordQuerySuccess(QuerySettings, OperationMetrics)}. + * + * @param settings - resolved settings of the operation + * @param duration - time the operation took before it failed, measured by the client + * @param t - failure the caller receives + */ + void recordQueryFailure(QuerySettings settings, Duration duration, Throwable t); + + /** + * Reports that an insert operation failed. It is the counterpart of + * {@link #recordInsertSuccess(InsertSettings, String, OperationMetrics)}. + * + * @param settings - resolved settings of the operation + * @param tableName - target table + * @param duration - time the operation took before it failed, measured by the client + * @param t - failure the caller receives + */ + void recordInsertFailure(InsertSettings settings, String tableName, Duration duration, Throwable t); + + /** + * Reports that an attempt of a read operation failed and the client retries it. Called once per + * retried attempt, so an operation that succeeds on its third attempt reports two retries. + * + * @param settings - resolved settings of the operation + * @param cause - failure of the attempt that is retried, reported the way the operation would + * report it if no further attempt succeeded + */ + void recordQueryRetry(QuerySettings settings, Throwable cause); + + /** + * Reports that an attempt of an insert operation failed and the client retries it. Called once + * per retried attempt. + * + * @param settings - resolved settings of the operation + * @param tableName - target table + * @param cause - failure of the attempt that is retried + */ + void recordInsertRetry(InsertSettings settings, String tableName, Throwable cause); +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricsSupport.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricsSupport.java new file mode 100644 index 000000000..89c6779a5 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/MetricsSupport.java @@ -0,0 +1,196 @@ +package com.clickhouse.client.api.observability; + +import com.clickhouse.client.api.ServerException; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.internal.StopWatch; +import com.clickhouse.client.api.metrics.ClientMetrics; +import com.clickhouse.client.api.metrics.Metric; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.query.QuerySettings; + +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Derives the client's standard metric attributes and values from the structures a + * {@link MetricsRecorder} is called with. + *

+ * This class is a helper for a recorder, not a layer in front of one: the client always calls + * the registered recorder first, and an implementation decides whether to use this class. Using it is + * how a recorder reports the same values under the same {@link MetricName} and the same + * {@link MetricAttribute} keys as every other recorder; a recorder that wants other values either + * overrides the method that computes them, or does not use this class at all. + *

+ * Every method may be overridden. {@link #DEFAULT} is a shared instance for implementations that keep + * the standard behaviour - the class holds no state. + */ +public class MetricsSupport { + + /** + * Shared instance with the standard behaviour. + */ + public static final MetricsSupport DEFAULT = new MetricsSupport(); + + /** + * Value of {@link MetricAttribute#DB_SYSTEM_NAME}. + */ + public static final String DB_SYSTEM_NAME = "clickhouse"; + + public static final String OPERATION_QUERY = "query"; + + public static final String OPERATION_INSERT = "insert"; + + /** + * Returned by the duration methods when the client did not measure that duration, so that a + * recorder does not report a made-up value. + */ + public static final double DURATION_UNKNOWN = -1d; + + private static final double NANOS_PER_SECOND = 1_000_000_000d; + + private static final double MILLIS_PER_SECOND = 1_000d; + + /** + * Returns the attributes of a read operation - a query, a command, a ping or a table-schema + * lookup. + * + * @param settings - resolved settings of the operation + * @param failure - failure of the operation, or {@code null} when it succeeded + * @return attributes, keyed by {@link MetricAttribute#getKey()} + */ + public Map queryAttributes(QuerySettings settings, Throwable failure) { + return attributes(settings == null ? null : settings.getDatabase(), OPERATION_QUERY, null, failure); + } + + /** + * Returns the attributes of an insert operation. + * + * @param settings - resolved settings of the operation + * @param tableName - target table + * @param failure - failure of the operation, or {@code null} when it succeeded + * @return attributes, keyed by {@link MetricAttribute#getKey()} + */ + public Map insertAttributes(InsertSettings settings, String tableName, Throwable failure) { + return attributes(settings == null ? null : settings.getDatabase(), OPERATION_INSERT, tableName, failure); + } + + /** + * Returns the duration of a completed operation in seconds, the unit of + * {@link MetricName#OPERATION_DURATION}. + * + * @param metrics - metrics of the completed operation, may be {@code null} + * @return duration in seconds, or {@link #DURATION_UNKNOWN} when the client did not measure it + */ + public double operationDuration(OperationMetrics metrics) { + return durationOf(metrics, ClientMetrics.OP_DURATION); + } + + /** + * Returns the duration of the serialization step of a completed operation in seconds, the unit of + * {@link MetricName#OPERATION_SERIALIZATION_DURATION}. + * + * @param metrics - metrics of the completed operation, may be {@code null} + * @return duration in seconds, or {@link #DURATION_UNKNOWN} when the client did not measure it + */ + public double serializationDuration(OperationMetrics metrics) { + return durationOf(metrics, ClientMetrics.OP_SERIALIZATION); + } + + /** + * Returns a duration the client measured itself - the duration of a failed operation - in + * seconds, the unit of {@link MetricName#OPERATION_DURATION}. + * + * @param duration - measured duration, may be {@code null} + * @return duration in seconds, or {@link #DURATION_UNKNOWN} when there is none + */ + public double duration(Duration duration) { + return duration == null ? DURATION_UNKNOWN : duration.toNanos() / NANOS_PER_SECOND; + } + + /** + * Returns the value of {@link MetricAttribute#ERROR_TYPE} for a failure - a short, + * low-cardinality identifier of what went wrong. A failure the server reported is identified by + * the server error it carries, the same way {@link SpanSupport} identifies it, so a span and a + * metric of the same failure report the same error type. + * + * @param t - failure, may be {@code null} + * @return error type, or {@code null} when there is no failure + */ + public String errorType(Throwable t) { + if (t == null) { + return null; + } + + ServerException serverException = findServerException(t); + return serverException == null ? t.getClass().getName() : serverException.getClass().getName(); + } + + /** + * Reads one duration of a completed operation and converts it to seconds. A duration the client + * did not measure is not reported. + * + * @param metrics - metrics of the completed operation, may be {@code null} + * @param metric - duration to read + * @return duration in seconds, or {@link #DURATION_UNKNOWN} + */ + protected double durationOf(OperationMetrics metrics, ClientMetrics metric) { + if (metrics == null) { + return DURATION_UNKNOWN; + } + + Metric value = metrics.getMetric(metric); + if (value == null) { + return DURATION_UNKNOWN; + } + // A stopwatch keeps nanoseconds, while the Metric contract exposes whole milliseconds, which + // would round a fast operation down to zero. + if (value instanceof StopWatch) { + return ((StopWatch) value).getElapsedNanos() / NANOS_PER_SECOND; + } + return value.getLong() / MILLIS_PER_SECOND; + } + + /** + * Returns the attributes that describe an operation. A value the client does not know is left + * out, so a recorder never reports an attribute with a placeholder value. + */ + protected Map attributes(String namespace, String operationName, String collectionName, + Throwable failure) { + Map attributes = new LinkedHashMap<>(); + attributes.put(MetricAttribute.DB_SYSTEM_NAME.getKey(), DB_SYSTEM_NAME); + if (namespace != null) { + attributes.put(MetricAttribute.DB_NAMESPACE.getKey(), namespace); + } + if (operationName != null) { + attributes.put(MetricAttribute.DB_OPERATION_NAME.getKey(), operationName); + } + if (collectionName != null) { + attributes.put(MetricAttribute.DB_COLLECTION_NAME.getKey(), collectionName); + } + if (failure != null) { + ServerException serverException = findServerException(failure); + if (serverException != null) { + attributes.put(MetricAttribute.DB_RESPONSE_STATUS_CODE.getKey(), serverException.getCode()); + } + attributes.put(MetricAttribute.ERROR_TYPE.getKey(), errorType(failure)); + } + return Collections.unmodifiableMap(attributes); + } + + /** + * Finds the server error in a failure, if the server reported one. + */ + protected ServerException findServerException(Throwable t) { + for (Throwable cause = t; cause != null; cause = cause.getCause()) { + if (cause instanceof ServerException) { + return (ServerException) cause; + } + if (cause.getCause() == cause) { + break; + } + } + return null; + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/observability/CapturingMetricsRecorder.java b/client-v2/src/test/java/com/clickhouse/client/api/observability/CapturingMetricsRecorder.java new file mode 100644 index 000000000..235f565e5 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/observability/CapturingMetricsRecorder.java @@ -0,0 +1,140 @@ +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; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Recorder that keeps every metric it reports, so tests can assert what the client reported. It takes + * the values and the attributes from {@link MetricsSupport}, which is how a recorder opts in to the + * client's standard values. + */ +public class CapturingMetricsRecorder extends DefaultMetricsRecorder { + + private final List metrics = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void recordQuerySuccess(QuerySettings settings, OperationMetrics metrics) { + MetricsSupport support = getMetricsSupport(); + recordCompletion(support.operationDuration(metrics), support.serializationDuration(metrics), + support.queryAttributes(settings, null)); + } + + @Override + public void recordInsertSuccess(InsertSettings settings, String tableName, OperationMetrics metrics) { + MetricsSupport support = getMetricsSupport(); + recordCompletion(support.operationDuration(metrics), support.serializationDuration(metrics), + support.insertAttributes(settings, tableName, null)); + } + + @Override + public void recordQueryFailure(QuerySettings settings, Duration duration, Throwable t) { + MetricsSupport support = getMetricsSupport(); + recordCompletion(support.duration(duration), MetricsSupport.DURATION_UNKNOWN, + support.queryAttributes(settings, t)); + } + + @Override + public void recordInsertFailure(InsertSettings settings, String tableName, Duration duration, Throwable t) { + MetricsSupport support = getMetricsSupport(); + recordCompletion(support.duration(duration), MetricsSupport.DURATION_UNKNOWN, + support.insertAttributes(settings, tableName, t)); + } + + @Override + public void recordQueryRetry(QuerySettings settings, Throwable cause) { + add(MetricName.OPERATION_RETRIES, 1, getMetricsSupport().queryAttributes(settings, cause)); + } + + @Override + public void recordInsertRetry(InsertSettings settings, String tableName, Throwable cause) { + add(MetricName.OPERATION_RETRIES, 1, getMetricsSupport().insertAttributes(settings, tableName, cause)); + } + + private void recordCompletion(double duration, double serializationDuration, Map attributes) { + add(MetricName.OPERATION_COUNT, 1, attributes); + add(MetricName.OPERATION_DURATION, duration, attributes); + if (serializationDuration != MetricsSupport.DURATION_UNKNOWN) { + add(MetricName.OPERATION_SERIALIZATION_DURATION, serializationDuration, attributes); + } + } + + private void add(MetricName name, double value, Map attributes) { + metrics.add(new RecordedMetric(name, value, attributes)); + } + + public List getMetrics() { + synchronized (metrics) { + return new ArrayList<>(metrics); + } + } + + /** + * Returns the values reported for the given metric, in the order they were reported. + */ + public List getMetrics(MetricName name) { + List selected = new ArrayList<>(); + for (RecordedMetric metric : getMetrics()) { + if (metric.getName() == name) { + selected.add(metric); + } + } + return selected; + } + + /** + * Returns the only value reported for the given metric. + */ + public RecordedMetric getOnlyMetric(MetricName name) { + List selected = getMetrics(name); + if (selected.size() != 1) { + throw new AssertionError("Expected exactly one " + name + " but got " + selected); + } + return selected.get(0); + } + + public void clear() { + metrics.clear(); + } + + public static final class RecordedMetric { + + private final MetricName name; + private final double value; + private final Map attributes; + + RecordedMetric(MetricName name, double value, Map attributes) { + this.name = name; + this.value = value; + this.attributes = new LinkedHashMap<>(attributes); + } + + public MetricName getName() { + return name; + } + + public double getValue() { + return value; + } + + public Map getAttributes() { + return Collections.unmodifiableMap(attributes); + } + + public Object getAttribute(MetricAttribute attribute) { + return attributes.get(attribute.getKey()); + } + + @Override + public String toString() { + return "RecordedMetric{name=" + name + ", value=" + value + ", attributes=" + attributes + '}'; + } + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/observability/MetricsRecorderUnitTest.java b/client-v2/src/test/java/com/clickhouse/client/api/observability/MetricsRecorderUnitTest.java new file mode 100644 index 000000000..34ff47163 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/observability/MetricsRecorderUnitTest.java @@ -0,0 +1,365 @@ +package com.clickhouse.client.api.observability; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.observability.CapturingMetricsRecorder.RecordedMetric; +import com.clickhouse.client.api.query.QueryResponse; +import com.clickhouse.client.api.query.QuerySettings; +import com.clickhouse.client.api.transport.Endpoint; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseFormat; +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +public class MetricsRecorderUnitTest { + + private static final String DEAD_ENDPOINT = "http://127.0.0.1:1"; // nothing listens here + + // Work a span recorder does before the operation runs and after it failed. The two delays differ so that + // each end of the measured interval is pinned on its own, and both are long enough to stand out from the + // operation itself, which fails as soon as the connection to the dead endpoint is refused. + private static final Duration SPAN_START_DELAY = Duration.ofMillis(800); + private static final Duration RECORD_FAILURE_DELAY = Duration.ofMillis(500); + + private CapturingMetricsRecorder recorder; + private WireMockServer mockServer; + + @BeforeMethod + void setUp() { + recorder = new CapturingMetricsRecorder(); + mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort()); + mockServer.start(); + mockServer.stubFor(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse().withStatus(200) + .withHeader("Content-Type", "text/plain") + .withBody(""))); + } + + @AfterMethod + void tearDown() { + mockServer.stop(); + } + + @Test + public void testQueryReportsDurationAndOperationAttributes() throws Exception { + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + try (QueryResponse response = client.query("SELECT 1", new QuerySettings().setQueryId("query-id-1")) + .get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + RecordedMetric duration = recorder.getOnlyMetric(MetricName.OPERATION_DURATION); + Assert.assertTrue(duration.getValue() > 0, "Unexpected duration: " + duration.getValue()); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_SYSTEM_NAME), "clickhouse"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_NAMESPACE), "test_db"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_OPERATION_NAME), "query"); + Assert.assertNull(duration.getAttribute(MetricAttribute.ERROR_TYPE), "the operation succeeded"); + Assert.assertNull(duration.getAttribute(MetricAttribute.DB_COLLECTION_NAME), "a query has no table"); + Assert.assertEquals(recorder.getOnlyMetric(MetricName.OPERATION_COUNT).getValue(), 1d); + Assert.assertTrue(recorder.getMetrics(MetricName.OPERATION_RETRIES).isEmpty(), "no attempt failed"); + } + + @Test + public void testInsertReportsTableAndSerializationDuration() throws Exception { + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + client.register(ValuePojo.class, new TableSchema("target_table", null, "", + Collections.singletonList(ClickHouseColumn.of("value", "String")))); + client.insert("target_table", Collections.singletonList(new ValuePojo("a"))).get(10, TimeUnit.SECONDS) + .close(); + } + + RecordedMetric duration = recorder.getOnlyMetric(MetricName.OPERATION_DURATION); + Assert.assertTrue(duration.getValue() > 0, "Unexpected duration: " + duration.getValue()); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_OPERATION_NAME), "insert"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_COLLECTION_NAME), "target_table"); + Assert.assertNull(duration.getAttribute(MetricAttribute.ERROR_TYPE)); + + RecordedMetric serialization = recorder.getOnlyMetric(MetricName.OPERATION_SERIALIZATION_DURATION); + Assert.assertTrue(serialization.getValue() > 0, "Unexpected duration: " + serialization.getValue()); + Assert.assertEquals(serialization.getAttribute(MetricAttribute.DB_COLLECTION_NAME), "target_table"); + } + + @Test + public void testQueryWithoutSerializationStepReportsNoSerializationDuration() throws Exception { + // contrast case: the client measures the serialization step of an insert only + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + client.query("SELECT 1").get(10, TimeUnit.SECONDS).close(); + } + + Assert.assertTrue(recorder.getMetrics(MetricName.OPERATION_SERIALIZATION_DURATION).isEmpty(), + "a duration the client did not measure must not be reported"); + } + + @Test + public void testFailedQueryIsCountedWithErrorType() throws Exception { + String expectedErrorType = null; + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(0).build()) { + try { + client.query("SELECT 1").get(30, TimeUnit.SECONDS).close(); + Assert.fail("a query against a dead endpoint must fail"); + } catch (ExecutionException e) { + expectedErrorType = e.getCause().getClass().getName(); + } catch (RuntimeException e) { + // with synchronous operations the failure is thrown by the operation itself + expectedErrorType = e.getClass().getName(); + } + } + + RecordedMetric count = recorder.getOnlyMetric(MetricName.OPERATION_COUNT); + Assert.assertEquals(count.getValue(), 1d, "a failed operation is counted too"); + Assert.assertEquals(count.getAttribute(MetricAttribute.ERROR_TYPE), expectedErrorType); + Assert.assertEquals(count.getAttribute(MetricAttribute.DB_OPERATION_NAME), "query"); + + RecordedMetric duration = recorder.getOnlyMetric(MetricName.OPERATION_DURATION); + Assert.assertTrue(duration.getValue() > 0, "the client measures the duration of a failed operation"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.ERROR_TYPE), expectedErrorType); + } + + @Test + public void testFailedInsertIsCountedWithTableAndErrorType() throws Exception { + String expectedErrorType = null; + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(0).build()) { + client.register(ValuePojo.class, new TableSchema("target_table", null, "", + Collections.singletonList(ClickHouseColumn.of("value", "String")))); + try { + client.insert("target_table", Collections.singletonList(new ValuePojo("a"))).get(30, TimeUnit.SECONDS); + Assert.fail("an insert into a dead endpoint must fail"); + } catch (ExecutionException e) { + expectedErrorType = e.getCause().getClass().getName(); + } catch (RuntimeException e) { + expectedErrorType = e.getClass().getName(); + } + } + + RecordedMetric count = recorder.getOnlyMetric(MetricName.OPERATION_COUNT); + Assert.assertEquals(count.getValue(), 1d); + Assert.assertEquals(count.getAttribute(MetricAttribute.DB_OPERATION_NAME), "insert"); + Assert.assertEquals(count.getAttribute(MetricAttribute.DB_COLLECTION_NAME), "target_table"); + Assert.assertEquals(count.getAttribute(MetricAttribute.ERROR_TYPE), expectedErrorType); + } + + @Test + public void testStreamInsertReportsTargetTable() throws Exception { + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + client.insert("target_table", new ByteArrayInputStream("a\n".getBytes(StandardCharsets.UTF_8)), + ClickHouseFormat.TabSeparated).get(10, TimeUnit.SECONDS).close(); + } + + RecordedMetric duration = recorder.getOnlyMetric(MetricName.OPERATION_DURATION); + Assert.assertTrue(duration.getValue() > 0, "Unexpected duration: " + duration.getValue()); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_OPERATION_NAME), "insert"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_COLLECTION_NAME), "target_table"); + Assert.assertEquals(recorder.getOnlyMetric(MetricName.OPERATION_COUNT).getValue(), 1d); + } + + @Test + public void testEveryRetriedInsertAttemptIsReported() throws Exception { + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(2).build()) { + try { + client.insert("target_table", new ByteArrayInputStream("a\n".getBytes(StandardCharsets.UTF_8)), + ClickHouseFormat.TabSeparated).get(30, TimeUnit.SECONDS).close(); + Assert.fail("an insert into a dead endpoint must fail"); + } catch (ExecutionException | RuntimeException e) { + // expected - nothing listens on the endpoint + } + } + + List retries = recorder.getMetrics(MetricName.OPERATION_RETRIES); + Assert.assertEquals(retries.size(), 2, "one retry event per retried attempt"); + RecordedMetric failure = recorder.getOnlyMetric(MetricName.OPERATION_COUNT); + for (RecordedMetric retry : retries) { + Assert.assertEquals(retry.getAttribute(MetricAttribute.DB_COLLECTION_NAME), "target_table"); + Assert.assertEquals(retry.getAttribute(MetricAttribute.ERROR_TYPE), + failure.getAttribute(MetricAttribute.ERROR_TYPE), + "a retry reports the failure the way the operation reports it"); + } + } + + @DataProvider(name = "retryCounts") + public static Object[][] retryCounts() { + return new Object[][]{{0}, {1}, {3}}; + } + + @Test(dataProvider = "retryCounts") + public void testEveryRetriedAttemptIsReported(int maxRetries) throws Exception { + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(maxRetries).build()) { + try { + client.query("SELECT 1").get(30, TimeUnit.SECONDS).close(); + Assert.fail("a query against a dead endpoint must fail"); + } catch (ExecutionException | RuntimeException e) { + // expected - nothing listens on the endpoint + } + } + + List retries = recorder.getMetrics(MetricName.OPERATION_RETRIES); + Assert.assertEquals(retries.size(), maxRetries, "one retry event per retried attempt"); + for (RecordedMetric retry : retries) { + Assert.assertEquals(retry.getValue(), 1d); + Assert.assertEquals(retry.getAttribute(MetricAttribute.DB_OPERATION_NAME), "query"); + Assert.assertNotNull(retry.getAttribute(MetricAttribute.ERROR_TYPE), + "a retry reports what made the attempt fail"); + } + Assert.assertEquals(recorder.getMetrics(MetricName.OPERATION_COUNT).size(), 1, + "a retried operation is still one operation"); + } + + @Test + public void testFailedQueryIsMeasuredFromTheSameOriginAsASuccessfulOne() throws Exception { + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(0) + .setSpanRecorder(new DelayingSpanRecorder()).build()) { + Assert.assertThrows(Exception.class, + () -> client.query("SELECT 1").get(30, TimeUnit.SECONDS).close()); + } + + assertFailureDurationOrigin(recorder.getOnlyMetric(MetricName.OPERATION_DURATION).getValue()); + } + + @Test + public void testFailedInsertIsMeasuredFromTheSameOriginAsASuccessfulOne() throws Exception { + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(0) + .setSpanRecorder(new DelayingSpanRecorder()).build()) { + client.register(ValuePojo.class, new TableSchema("target_table", null, "", + Collections.singletonList(ClickHouseColumn.of("value", "String")))); + Assert.assertThrows(Exception.class, () -> client + .insert("target_table", Collections.singletonList(new ValuePojo("a"))).get(30, TimeUnit.SECONDS)); + } + + assertFailureDurationOrigin(recorder.getOnlyMetric(MetricName.OPERATION_DURATION).getValue()); + } + + /** + * The client starts the duration of a successful operation before it prepares the request and before it + * starts the operation span, and stops it before any recorder runs. A failed operation must report a + * duration that covers the same work, so that both outcomes form one latency series. + */ + private static void assertFailureDurationOrigin(double reportedSeconds) { + double spanStartSeconds = SPAN_START_DELAY.toMillis() / 1000d; + double recordFailureSeconds = RECORD_FAILURE_DELAY.toMillis() / 1000d; + Assert.assertTrue(reportedSeconds >= spanStartSeconds, + "the duration must start where the client starts the duration of a successful operation, so that it " + + "covers the work done before the operation span starts; reported: " + reportedSeconds); + Assert.assertTrue(reportedSeconds < spanStartSeconds + recordFailureSeconds, + "the duration must be taken before any recorder runs, like the duration of a successful operation; " + + "reported: " + reportedSeconds); + } + + /** + * Span recorder that spends a known amount of time when the client starts an operation span and again + * when it reports a failure - the two points that fence the origin of a failed operation's duration. + */ + private static final class DelayingSpanRecorder extends DefaultSpanRecorder { + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + sleep(SPAN_START_DELAY); + return super.startQuerySpan(settings, sqlQuery, endpoint); + } + + @Override + public Span startInsertSpan(InsertSettings settings, String tableName, int batchSize, Endpoint endpoint) { + sleep(SPAN_START_DELAY); + return super.startInsertSpan(settings, tableName, batchSize, endpoint); + } + + @Override + public void recordFailure(Span operationSpan, Throwable t) { + sleep(RECORD_FAILURE_DELAY); + } + + private static void sleep(Duration delay) { + try { + Thread.sleep(delay.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + } + + @Test + public void testAttributesOfASuccessfulOperationAreLowCardinality() { + QuerySettings settings = new QuerySettings().setDatabase("test_db").setQueryId("query-id-1"); + + Map attributes = MetricsSupport.DEFAULT.queryAttributes(settings, null); + + Assert.assertEquals(attributes.get(MetricAttribute.DB_SYSTEM_NAME.getKey()), "clickhouse"); + Assert.assertEquals(attributes.get(MetricAttribute.DB_NAMESPACE.getKey()), "test_db"); + Assert.assertEquals(attributes.get(MetricAttribute.DB_OPERATION_NAME.getKey()), "query"); + Assert.assertEquals(attributes.size(), 3, + "the query id and the statement are recorded on spans, not on metrics: " + attributes); + } + + @DataProvider(name = "unmeasuredDurations") + public static Object[][] unmeasuredDurations() { + return new Object[][]{ + {MetricsSupport.DEFAULT.operationDuration(null)}, + {MetricsSupport.DEFAULT.serializationDuration(null)}, + {MetricsSupport.DEFAULT.duration(null)}, + }; + } + + @Test(dataProvider = "unmeasuredDurations") + public void testUnmeasuredDurationIsReportedAsUnknown(double duration) { + Assert.assertEquals(duration, MetricsSupport.DURATION_UNKNOWN); + } + + @Test + public void testDurationIsConvertedToSeconds() { + Assert.assertEquals(MetricsSupport.DEFAULT.duration(Duration.ofMillis(1500)), 1.5d); + Assert.assertEquals(MetricName.OPERATION_DURATION.getUnit(), "s"); + Assert.assertEquals(MetricName.OPERATION_DURATION.getKey(), "db.client.operation.duration"); + } + + @Test + public void testNullRecorderIsRejected() { + Assert.assertThrows(NullPointerException.class, () -> new Client.Builder().setMetricsRecorder(null)); + } + + private Client.Builder newClientBuilder() { + return new Client.Builder() + .setUsername("default") + .setPassword("") + .setDefaultDatabase("test_db") + .setMetricsRecorder(recorder); + } + + private String mockEndpoint() { + return "http://localhost:" + mockServer.port(); + } + + public static class ValuePojo { + private String value; + + public ValuePojo() { + } + + public ValuePojo(String value) { + this.value = value; + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/observability/MetricsRecorderTest.java b/client-v2/src/test/java/com/clickhouse/client/observability/MetricsRecorderTest.java new file mode 100644 index 000000000..944a7d9e8 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/observability/MetricsRecorderTest.java @@ -0,0 +1,137 @@ +package com.clickhouse.client.observability; + +import com.clickhouse.client.BaseIntegrationTest; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseServerForTest; +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.ServerException; +import com.clickhouse.client.api.enums.Protocol; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.observability.CapturingMetricsRecorder; +import com.clickhouse.client.api.observability.CapturingMetricsRecorder.RecordedMetric; +import com.clickhouse.client.api.observability.MetricAttribute; +import com.clickhouse.client.api.observability.MetricName; +import com.clickhouse.client.api.query.QueryResponse; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.util.Collections; +import java.util.concurrent.ExecutionException; + +public class MetricsRecorderTest extends BaseIntegrationTest { + + private static final String TABLE = "metrics_recorder_test_table"; + + private CapturingMetricsRecorder recorder; + private Client client; + private String database; + + @BeforeMethod(groups = {"integration"}) + void setUp() throws Exception { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + database = ClickHouseServerForTest.getDatabase(); + recorder = new CapturingMetricsRecorder(); + client = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isCloud()) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(database) + .setMetricsRecorder(recorder) + .build(); + client.execute("DROP TABLE IF EXISTS " + TABLE).get(); + client.execute("CREATE TABLE " + TABLE + " (value String) ENGINE = MergeTree ORDER BY value").get(); + recorder.clear(); + } + + @AfterMethod(groups = {"integration"}) + void tearDown() throws Exception { + if (client != null) { + client.execute("DROP TABLE IF EXISTS " + TABLE).get(); + client.close(); + } + } + + @Test(groups = {"integration"}) + public void testSuccessfulQueryReportsDurationAndOperationAttributes() throws Exception { + try (QueryResponse response = client.query("SELECT value FROM " + TABLE).get()) { + Assert.assertNotNull(response); + } + + RecordedMetric duration = recorder.getOnlyMetric(MetricName.OPERATION_DURATION); + Assert.assertTrue(duration.getValue() > 0, "Unexpected duration: " + duration.getValue()); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_SYSTEM_NAME), "clickhouse"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_NAMESPACE), database); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_OPERATION_NAME), "query"); + Assert.assertNull(duration.getAttribute(MetricAttribute.ERROR_TYPE)); + Assert.assertEquals(recorder.getOnlyMetric(MetricName.OPERATION_COUNT).getValue(), 1d); + } + + @Test(groups = {"integration"}) + public void testSuccessfulInsertReportsTargetTable() throws Exception { + client.register(ValuePojo.class, client.getTableSchema(TABLE)); + recorder.clear(); + + client.insert(TABLE, Collections.singletonList(new ValuePojo("a"))).get().close(); + + RecordedMetric duration = recorder.getOnlyMetric(MetricName.OPERATION_DURATION); + Assert.assertTrue(duration.getValue() > 0, "Unexpected duration: " + duration.getValue()); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_OPERATION_NAME), "insert"); + Assert.assertEquals(duration.getAttribute(MetricAttribute.DB_COLLECTION_NAME), TABLE); + Assert.assertNull(duration.getAttribute(MetricAttribute.ERROR_TYPE)); + Assert.assertTrue(recorder.getOnlyMetric(MetricName.OPERATION_SERIALIZATION_DURATION).getValue() > 0); + } + + @Test(groups = {"integration"}) + public void testServerErrorReportsErrorTypeAndStatusCode() { + try { + client.query("SELECT * FROM table_that_does_not_exist_at_all").get(); + Assert.fail("querying a missing table must fail"); + } catch (ExecutionException e) { + Assert.assertTrue(e.getCause() instanceof ServerException, "Unexpected cause: " + e.getCause()); + } catch (ServerException e) { + // synchronous operations report the server failure directly + } catch (Exception e) { + Assert.fail("Unexpected exception: " + e); + } + + RecordedMetric count = recorder.getOnlyMetric(MetricName.OPERATION_COUNT); + Assert.assertEquals(count.getValue(), 1d, "a failed operation is counted too"); + Assert.assertEquals(count.getAttribute(MetricAttribute.ERROR_TYPE), ServerException.class.getName()); + Assert.assertEquals(count.getAttribute(MetricAttribute.DB_RESPONSE_STATUS_CODE), + ServerException.TABLE_NOT_FOUND); + Assert.assertTrue(recorder.getOnlyMetric(MetricName.OPERATION_DURATION).getValue() > 0); + } + + @Test(groups = {"integration"}) + public void testTableSchemaLookupIsReportedAsQuery() { + TableSchema schema = client.getTableSchema(TABLE); + Assert.assertEquals(schema.getColumns().size(), 1); + + RecordedMetric count = recorder.getOnlyMetric(MetricName.OPERATION_COUNT); + Assert.assertEquals(count.getAttribute(MetricAttribute.DB_OPERATION_NAME), "query"); + Assert.assertNull(count.getAttribute(MetricAttribute.DB_COLLECTION_NAME), + "an operation the client implements on top of a query is reported as one"); + } + + public static class ValuePojo { + private String value; + + public ValuePojo() { + } + + public ValuePojo(String value) { + this.value = value; + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + } +} diff --git a/docs/features.md b/docs/features.md index 1ee6cffbd..071c99d2f 100644 --- a/docs/features.md +++ b/docs/features.md @@ -36,6 +36,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. A cancelled operation that is being retried stops instead of issuing another request, also when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`). - Metrics and observability: Exposes client/server operation metrics and optionally integrates connection-pool gauges with Micrometer. - Span recording (tracing SPI): `Client.Builder.setSpanRecorder(SpanRecorder)` registers a backend-agnostic recorder (package `com.clickhouse.client.api.observability`) that observes client operations. Every operation - a query, a command or an insert, including the `ping` and `getTableSchema` calls, which run a query - 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 what it wants to record, so a recorder keeps working when the client starts a kind of span it does not know about. The registered recorder is the first thing the client calls, and it is called with everything the client knows about the operation - its settings object, the statement, the target table, the batch size, the endpoint, the metrics of the completed operation and the failure - so an implementation is free to record whatever it needs and in whatever form. Deriving the standard span names and attribute values from those structures is done by `SpanSupport`, which a recorder implementation calls if it wants them; using it is opt-in, and its methods may be overridden to report other values. A recorder opts in through `DefaultSpanRecorder#getSpanSupport()` (or `SpanSupport.DEFAULT`). 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 derived by `SpanSupport`, so every recorder that uses it reports the same information. An operation span is named `query `, or `insert .` for an insert, and carries `db.system.name`, `db.namespace`, `clickhouse.query_id`, `db.query.text` (query/command), `db.query.parameter.` (parameterized query), `db.operation.name` (insert), `db.collection.name` (insert), `db.operation.batch.size` (POJO insert), `server.address`/`server.port` of the first configured endpoint (each attempt reports its own on the request span), the metrics of the completed operation on success, and `error.type` plus `db.response.status_code` on failure. Success is reported per operation kind, because the metrics that describe a read are not the ones that describe a write: `SpanRecorder#recordQuerySuccess` is called for a read operation and `SpanRecorder#recordInsertSuccess` for an insert, and each records only the metrics of its own kind - 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`, and both report `clickhouse.query_id`. These values come from the server's progress summary, which is not always available (for example a query reports them with `QuerySettings#waitEndOfQuery(true)`); a metric the server did not report is left out. The same distinction is on the metrics themselves: `OperationMetrics#getOperationType()` returns `OperationType.QUERY` or `OperationType.INSERT`. It reports the kind of the call the application made, not the kind of work the server did, so a command that writes - `INSERT INTO ... SELECT` run through `execute` - is reported as a query. A request span is named `POST` and carries `http.request.method`, the `server.address`/`server.port` of that attempt, `http.response.status_code` once a response is received, and `error.type`/`db.response.status_code` when the attempt fails. An operation span is started on the calling thread, so it joins the caller's ambient trace even when `async_operations` runs the operation on the client's executor, and it is ended when the operation returns its response to the caller - so it covers sending the request and receiving the response head, not streaming the response body afterwards. Each span of an operation that started is ended exactly once, also when the operation failed; an operation that could not be started at all - a closed client, for example - does not report a span. When no recorder is registered nothing is recorded and no span-related work is done. +- Metrics recording (metrics SPI): `Client.Builder.setMetricsRecorder(MetricsRecorder)` registers a backend-agnostic recorder (package `com.clickhouse.client.api.observability`) that receives the metrics of client operations, so an application can export them to any metrics backend. It follows the same pattern as the span SPI: `MetricsRecorder` is a plain interface, an implementation extends the `DefaultMetricsRecorder` base class and overrides only what it wants to record, so a recorder 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 (opt-in through `DefaultMetricsRecorder#getMetricsSupport()`, and overridable). Every completed operation reports exactly one event - `recordQuerySuccess`/`recordInsertSuccess` or `recordQueryFailure`/`recordInsertFailure` - so counting those events gives the number of operations by outcome, and every retried attempt reports `recordQueryRetry`/`recordInsertRetry`, so an operation that succeeds on its third attempt reports two retries. The durations of a successful operation are read from its `OperationMetrics`; a failed operation has no metrics, so the client measures its duration itself and passes it in. 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: the `MetricName` enum defines `db.client.operation.duration` (unit `s`), `clickhouse.client.operation.serialization.duration` (unit `s`), `clickhouse.client.operation.count` and `clickhouse.client.operation.retries`, each with the unit and description an exporter registers its instrument with. Durations are reported in **seconds**, while `ClientMetrics` values are milliseconds - `MetricsSupport` converts them and returns `MetricsSupport.DURATION_UNKNOWN` for a duration the client did not measure (the serialization duration is measured for a POJO insert only), so a recorder never reports a made-up value. The attribute keys are the `MetricAttribute` enum - `db.system.name`, `db.namespace`, `db.operation.name` (`query` or `insert`), `db.collection.name` (insert), and `error.type` plus `db.response.status_code` on a failure. That is deliberately a smaller set than `SpanAttribute`, because an attribute of a metric becomes a time series: the statement text, the query id and the statement parameters stay on spans. A successful operation has no `error.type` attribute, so the successful time series is the one without it. When no recorder is registered nothing is recorded and no metrics-related work is done. The recorder is registered per client instance and must be thread-safe. - Configuration surface: Supports arbitrary client options, cookies, custom headers, server-setting prefixes, client naming, query id suppliers, and buffer sizing. - SQL helpers: Includes SQL quoting and temporal formatting helpers used by callers building SQL text safely. @@ -69,6 +70,7 @@ Compatibility-sensitive traits: - SSL URL support: Supports HTTPS connections through URL and property configuration, including default protocol and port handling. The `ssl_mode` property selects the verification strictness (`disabled`, `trust`, `verify_ca`, `strict`); values are case-insensitive and the traditional JDBC value `none` is accepted as an alias for `trust`. Root CA and client certificate/key may be supplied as a file path or as inline PEM content. The `ssl_cipher_suites` property (a comma-separated list) restricts the negotiated TLS cipher suites and is forwarded to the underlying `client-v2` transport. - Custom SSL context via properties: A fully pre-built `javax.net.ssl.SSLContext` may be passed as a live object in the connection `Properties` under the `ssl_context` key (added with `Properties.put`, since it is not a string). It is forwarded to the underlying `client-v2` transport and used as is; trust/key material properties (e.g. `sslrootcert`, `sslcert`) cannot be combined with it and are rejected, while `ssl_mode` then only controls hostname verification. This supports diskless, in-memory TLS material behind connection pools that only expose `java.util.Properties`. - Driver and client properties: Separates JDBC-specific properties from passthrough client options used by the underlying `client-v2` transport. +- Metrics recording: The `jdbc_metrics_recorder` property names the `client-v2` `MetricsRecorder` implementation the connection reports the metrics of its operations to. A JDBC URL carries strings only, so the value is the fully-qualified class name of a recorder with a public no-argument constructor, and each connection creates its own instance and registers it with its own client. A class that is not on the classpath, does not implement `MetricsRecorder`, or cannot be instantiated is rejected when the connection is created. When the property is not set, nothing is recorded. - DataSource support: Provides a JDBC `DataSource` implementation backed by the same driver configuration model. - Connection lifecycle: Supports connection close, validity checks, ping-based health checks, and network timeout management. - Underlying client access: `ConnectionImpl#getClient()` exposes the connection-owned `client-v2` instance for operations that are not representable through the JDBC API, including direct consumption of raw response streams. diff --git a/jdbc-v2/src/main/java/com/clickhouse/jdbc/ConnectionImpl.java b/jdbc-v2/src/main/java/com/clickhouse/jdbc/ConnectionImpl.java index c5fe17d31..9db463546 100644 --- a/jdbc-v2/src/main/java/com/clickhouse/jdbc/ConnectionImpl.java +++ b/jdbc-v2/src/main/java/com/clickhouse/jdbc/ConnectionImpl.java @@ -4,6 +4,7 @@ import com.clickhouse.client.api.ClientConfigProperties; import com.clickhouse.client.api.data_formats.JsonParserFactory; import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.observability.MetricsRecorder; import com.clickhouse.client.api.query.GenericRecord; import com.clickhouse.client.api.query.QuerySettings; import com.clickhouse.data.ClickHouseColumn; @@ -104,9 +105,14 @@ public ConnectionImpl(String url, Properties info) throws SQLException { } } - this.client = this.config.applyClientProperties(new Client.Builder()) - .setClientName(clientName) - .build(); + Client.Builder clientBuilder = this.config.applyClientProperties(new Client.Builder()) + .setClientName(clientName); + final String metricsRecorderName = config.getDriverProperty(DriverProperties.METRICS_RECORDER.getKey(), null); + if (metricsRecorderName != null) { + clientBuilder.setMetricsRecorder(instantiateUserClass(metricsRecorderName, MetricsRecorder.class, + DriverProperties.METRICS_RECORDER.getKey())); + } + this.client = clientBuilder.build(); String serverTimezone = this.client.getServerTimeZone(); if (serverTimezone == null) { // we cannot operate without timezone @@ -125,7 +131,9 @@ public ConnectionImpl(String url, Properties info) throws SQLException { this.typeMap = ImmutableMap.>builder().putAll(this.config.getTypeMap()).buildKeepingLast(); final String jsonParserFactoryName = config.getDriverProperty(DriverProperties.JSON_PARSER_FACTORY.getKey(), null); - this.jsonParserFactory = jsonParserFactoryName == null ? null : instantiateJsonParserFactory(jsonParserFactoryName); + this.jsonParserFactory = jsonParserFactoryName == null ? null + : instantiateUserClass(jsonParserFactoryName, JsonParserFactory.class, + DriverProperties.JSON_PARSER_FACTORY.getKey()); } catch (SQLException e) { throw e; } catch (Exception e) { @@ -133,21 +141,26 @@ public ConnectionImpl(String url, Properties info) throws SQLException { } } - private JsonParserFactory instantiateJsonParserFactory(String className) throws SQLException { + /** + * Creates the instance of a class a connection property names - a JSON parser factory, a metrics + * recorder. The class is expected to have a public no-argument constructor, so every property that + * names one reports the same failures in the same way. + */ + private T instantiateUserClass(String className, Class type, String propertyKey) throws SQLException { if (className == null || className.trim().isEmpty()) { - throw new SQLException("Value of '" + DriverProperties.JSON_PARSER_FACTORY.getKey() + - "' is empty string but should be a FQN of factory class."); + throw new SQLException("Value of '" + propertyKey + "' is empty string but should be a FQN of a class " + + "implementing " + type.getName() + "."); } try { - Class factoryClass = loadFactoryClass(className); - if (!JsonParserFactory.class.isAssignableFrom(factoryClass)) { - throw new SQLException("Class '" + className + "' should implement " + JsonParserFactory.class.getName()); + Class userClass = loadUserClass(className); + if (!type.isAssignableFrom(userClass)) { + throw new SQLException("Class '" + className + "' should implement " + type.getName()); } - return (JsonParserFactory) factoryClass.getDeclaredConstructor().newInstance(); + return type.cast(userClass.getDeclaredConstructor().newInstance()); } catch (ClassNotFoundException e) { - throw new SQLException("Class '" + className + "' (implementing JsonParserFactory ) not found. Check " + - DriverProperties.JSON_PARSER_FACTORY.getKey() + " property", e); + throw new SQLException("Class '" + className + "' (implementing " + type.getName() + ") not found. Check " + + propertyKey + " property", e); } catch (InvocationTargetException | InstantiationException | IllegalAccessException | NoSuchMethodException e) { throw new SQLException("Failed to instantiate '" + className + "'. Check class implementation.", e); @@ -155,12 +168,12 @@ private JsonParserFactory instantiateJsonParserFactory(String className) throws } /** - * Resolves a user-supplied factory class name. JDBC drivers are commonly deployed in a + * Resolves a user-supplied class name. JDBC drivers are commonly deployed in a * parent class loader (e.g. servlet container {@code lib/}) while caller-supplied classes * live in the application class loader, so the thread context class loader is tried first * and the driver's own class loader is used as a fallback. */ - private Class loadFactoryClass(String className) throws ClassNotFoundException { + private Class loadUserClass(String className) throws ClassNotFoundException { ClassNotFoundException firstFailure = null; ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader(); if (contextClassLoader != null) { diff --git a/jdbc-v2/src/main/java/com/clickhouse/jdbc/DriverProperties.java b/jdbc-v2/src/main/java/com/clickhouse/jdbc/DriverProperties.java index 8d6123628..db4963fee 100644 --- a/jdbc-v2/src/main/java/com/clickhouse/jdbc/DriverProperties.java +++ b/jdbc-v2/src/main/java/com/clickhouse/jdbc/DriverProperties.java @@ -153,6 +153,13 @@ public enum DriverProperties { */ JSON_PARSER_FACTORY("jdbc_json_parser_factory", null), + /** + * Defines which {@link com.clickhouse.client.api.observability.MetricsRecorder} implementation the connection + * should report the metrics of its operations to. Value is the fully-qualified class name of a recorder that has + * a public no-argument constructor; each connection creates its own instance. When not set, nothing is recorded. + */ + METRICS_RECORDER("jdbc_metrics_recorder", null), + ; diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/MetricsRecorderTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/MetricsRecorderTest.java new file mode 100644 index 000000000..655a8af33 --- /dev/null +++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/MetricsRecorderTest.java @@ -0,0 +1,124 @@ +package com.clickhouse.jdbc; + +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.observability.DefaultMetricsRecorder; +import com.clickhouse.client.api.observability.MetricAttribute; +import com.clickhouse.client.api.observability.MetricsSupport; +import com.clickhouse.client.api.query.QuerySettings; +import org.testng.Assert; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +public class MetricsRecorderTest extends JdbcIntegrationTest { + + @BeforeMethod(groups = {"integration"}) + void setUp() { + CollectingMetricsRecorder.clear(); + } + + @Test(groups = {"integration"}) + public void testConnectionReportsMetricsToConfiguredRecorder() throws Exception { + Properties properties = new Properties(); + properties.setProperty(DriverProperties.METRICS_RECORDER.getKey(), CollectingMetricsRecorder.class.getName()); + + try (Connection connection = getJdbcConnection(properties); + Statement statement = connection.createStatement()) { + try (ResultSet rs = statement.executeQuery("SELECT 1")) { + Assert.assertTrue(rs.next()); + } + } + + List> operations = CollectingMetricsRecorder.getOperations(); + Assert.assertFalse(operations.isEmpty(), "the recorder named by the property must be used"); + for (Map attributes : operations) { + Assert.assertEquals(attributes.get(MetricAttribute.DB_SYSTEM_NAME.getKey()), "clickhouse"); + Assert.assertEquals(attributes.get(MetricAttribute.DB_OPERATION_NAME.getKey()), "query"); + } + } + + @Test(groups = {"integration"}) + public void testConnectionWithoutThePropertyRecordsNothing() throws Exception { + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + try (ResultSet rs = statement.executeQuery("SELECT 1")) { + Assert.assertTrue(rs.next()); + } + } + + Assert.assertTrue(CollectingMetricsRecorder.getOperations().isEmpty(), + "a connection that did not ask for metrics must not report any"); + } + + @Test(groups = {"integration"}) + public void testUnknownRecorderClassIsRejected() { + Properties properties = new Properties(); + properties.setProperty(DriverProperties.METRICS_RECORDER.getKey(), "com.acme.NoSuchRecorder"); + + try { + getJdbcConnection(properties).close(); + Assert.fail("a recorder class that is not on the classpath must be rejected"); + } catch (SQLException e) { + Assert.assertTrue(e.getMessage().contains(DriverProperties.METRICS_RECORDER.getKey()), + "Unexpected message: " + e.getMessage()); + } + } + + @Test(groups = {"integration"}) + public void testClassThatIsNotARecorderIsRejected() { + Properties properties = new Properties(); + properties.setProperty(DriverProperties.METRICS_RECORDER.getKey(), String.class.getName()); + + try { + getJdbcConnection(properties).close(); + Assert.fail("a class that does not implement the recorder interface must be rejected"); + } catch (SQLException e) { + Assert.assertTrue(e.getMessage().contains(String.class.getName()), + "Unexpected message: " + e.getMessage()); + } + } + + /** + * Recorder the driver instantiates through the connection property, so what it collects is static. + */ + public static class CollectingMetricsRecorder extends DefaultMetricsRecorder { + + private static final List> OPERATIONS = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void recordQuerySuccess(QuerySettings settings, OperationMetrics metrics) { + OPERATIONS.add(getMetricsSupport().queryAttributes(settings, null)); + } + + @Override + public void recordQueryFailure(QuerySettings settings, Duration duration, Throwable t) { + OPERATIONS.add(getMetricsSupport().queryAttributes(settings, t)); + } + + @Override + public void recordInsertSuccess(InsertSettings settings, String tableName, OperationMetrics metrics) { + OPERATIONS.add(MetricsSupport.DEFAULT.insertAttributes(settings, tableName, null)); + } + + static List> getOperations() { + synchronized (OPERATIONS) { + return new ArrayList<>(OPERATIONS); + } + } + + static void clear() { + OPERATIONS.clear(); + } + } +}