diff --git a/CHANGELOG.md b/CHANGELOG.md index 50e06bff0..c81e612fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,29 @@ ### New Features +- **[client-v2]** Added an observability SPI that lets an application observe client operations as spans. + `Client.Builder.setSpanRecorder(SpanRecorder)` registers a backend-agnostic recorder from the new + `com.clickhouse.client.api.observability` package: each operation (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 cares about, so it keeps working when + the client starts a kind of span it does not know about. The registered recorder is called first and receives + everything the client knows about the operation (its `QuerySettings`/`InsertSettings`, the statement, the target + table, the batch size, the endpoint, the metrics of the completed operation and the failure), so it is free to + record whatever it needs and in whatever form; the reusable `SpanSupport` class derives the standard span names + and attribute values from those same structures and is called by a recorder implementation that wants them, so + its logic is opt-in and overridable. 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 all recorders that use it report the same information (statement text, target database and table, query id, + statement parameters, batch size, the first configured endpoint on the operation span and the per-attempt + server address and port on the request spans, HTTP status, returned rows, and the error type and ClickHouse + error code on failure). An operation span is started on the calling thread, so it joins + the caller's ambient trace even when the operation runs on the client's executor, and it is ended exactly once + for every operation that starts. Previously the client exposed no hook for tracing, so an + application could not attribute a query or a retried request to its own trace. When no recorder is registered + nothing is recorded and no span-related work is done, so the default path is unchanged. An OpenTelemetry + implementation of the SPI follows in a separate module. + (https://github.com/ClickHouse/clickhouse-java/issues/2974) - **[client-v2, jdbc-v2]** Added support for the `BFloat16` data type (ClickHouse `24.11+`). `BFloat16` columns are read as Java `float` values (widening is lossless) and written from `float`/`Float` values, including through generic records, POJO binding, `Nullable(BFloat16)`, and `BFloat16` values held in `Dynamic`/`Variant` columns. On write the client keeps the 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 2b022e393..3e5764adb 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 @@ -30,6 +30,9 @@ import com.clickhouse.client.api.metadata.TableSchema; import com.clickhouse.client.api.metrics.ClientMetrics; import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.observability.DefaultSpanRecorder; +import com.clickhouse.client.api.observability.Span; +import com.clickhouse.client.api.observability.SpanRecorder; import com.clickhouse.client.api.query.GenericRecord; import com.clickhouse.client.api.query.QueryResponse; import com.clickhouse.client.api.query.QuerySettings; @@ -155,15 +158,25 @@ public class Client implements AutoCloseable { private final ClientNodeSelector nodeSelector; private final CredentialsManager credentialsManager; + /** + * Recorder registered by an application; called first for every span the client starts, with + * everything the client knows about the operation. Never {@code null} - it is + * {@link DefaultSpanRecorder#NOOP} when observability is not configured, so no null check is + * needed on the operation paths. + */ + private final SpanRecorder spanRecorder; + private Client(Collection endpoints, Map configuration, ExecutorService sharedOperationExecutor, ColumnToMethodMatchingStrategy columnToMethodMatchingStrategy, Object metricsRegistry, Supplier queryIdGenerator, CredentialsManager cManager, - SSLContext sslContext) { + SSLContext sslContext, SpanRecorder spanRecorder) { Map 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.session = Session.extractFrom(parsedConfiguration); this.configuration = new ConcurrentHashMap<>(parsedConfiguration); this.readOnlyConfig = Collections.unmodifiableMap(configuration); @@ -210,7 +223,8 @@ private Client(Collection endpoints, Map configuration, this.lz4Factory = LZ4Factory.fastestJavaInstance(); } - this.httpClientHelper = new HttpAPIClientHelper(this.configuration, metricsRegistry, initSslContext, lz4Factory); + this.httpClientHelper = new HttpAPIClientHelper(this.configuration, metricsRegistry, initSslContext, lz4Factory, + this.spanRecorder); this.serverVersion = configuration.getOrDefault(ClientConfigProperties.SERVER_VERSION.getKey(), "unknown"); this.dbUser = configuration.getOrDefault(ClientConfigProperties.USER.getKey(), ClientConfigProperties.USER.getDefObjVal()); this.typeHintMapping = (Map>) this.configuration.get(ClientConfigProperties.TYPE_HINT_MAPPING.getKey()); @@ -283,6 +297,7 @@ public static class Builder { private Object metricRegistry = null; private Supplier queryIdGenerator; private SSLContext sslContext = null; + private SpanRecorder spanRecorder = DefaultSpanRecorder.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()). @@ -1230,6 +1245,27 @@ public Builder setQueryIdGenerator(Supplier supplier) { return this; } + /** + *

Registers a {@link SpanRecorder} that observes client operations and the transport + * requests made for them. Each operation (query, command, insert, ping, table-schema + * lookup) produces one operation span, and every request attempt made for it - including + * retries - produces a child request span.

+ * + *

When no recorder is set nothing is recorded and no span-related work is done. The + * default is {@link DefaultSpanRecorder#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 spanRecorder - recorder to notify; must not be {@code null} + * @return same instance of the builder + * @throws NullPointerException when {@code spanRecorder} is {@code null} + */ + public Builder setSpanRecorder(SpanRecorder spanRecorder) { + this.spanRecorder = Objects.requireNonNull(spanRecorder, + "spanRecorder is required; use DefaultSpanRecorder.NOOP to record nothing"); + return this; + } + public Client build() { // check if endpoint are empty. so can not initiate client if (this.endpoints.isEmpty()) { @@ -1317,7 +1353,7 @@ public Client build() { return new Client(this.endpoints, this.configuration, this.sharedOperationExecutor, this.columnToMethodMatchingStrategy, this.metricRegistry, this.queryIdGenerator, cManager, - this.sslContext); + this.sslContext, this.spanRecorder); } } @@ -1459,6 +1495,8 @@ public CompletableFuture insert(String tableName, List data, if (requestSettings.getQueryId() == null && queryIdGenerator != null) { requestSettings.setQueryId(queryIdGenerator.get()); } + final Span operationSpan = orNoop(spanRecorder.startInsertSpan(requestSettings, tableName, data.size(), + endpoints.get(0))); Supplier supplier = () -> { long startTime = System.nanoTime(); // Selecting some node @@ -1491,10 +1529,11 @@ public CompletableFuture insert(String tableName, List data, registerTransportReq(queryId, transportRequest); - try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { + try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest, operationSpan)) { ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); OperationMetrics metrics = completeOperation(transportResponse, clientStats, requestSettings.getQueryId()); + spanRecorder.recordSuccess(operationSpan, metrics); return new InsertResponse(transportResponse, metrics); } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); @@ -1510,15 +1549,19 @@ public CompletableFuture insert(String tableName, List data, } } } + + String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); + LOG.warn(errMsg); + throw (lastException == null ? new ClientException(errMsg) : lastException); + } catch (RuntimeException | Error e) { + spanRecorder.recordFailure(operationSpan, e); + throw e; } finally { // The request of the last attempt stays registered until the operation is over, so a cancellation // landing between two attempts is not lost. unregisterTransportReq(queryId); + operationSpan.end(); } - - String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); - LOG.warn(errMsg); - throw (lastException == null ? new ClientException(errMsg) : lastException); }; return runAsyncOperation(supplier, requestSettings.getAllSettings()); @@ -1684,6 +1727,8 @@ public CompletableFuture insert(String tableName, final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings()); final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1); + final Span operationSpan = orNoop(spanRecorder.startInsertSpan(requestSettings, tableName, + SpanRecorder.BATCH_SIZE_UNKNOWN, endpoints.get(0))); Supplier responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node @@ -1702,8 +1747,9 @@ public CompletableFuture insert(String tableName, }); registerTransportReq(queryId, transportRequest); - try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { + try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest, operationSpan)) { OperationMetrics metrics = completeOperation(transportResponse, finalClientStats, requestSettings.getQueryId()); + spanRecorder.recordSuccess(operationSpan, metrics); return new InsertResponse(transportResponse, metrics); } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); @@ -1727,15 +1773,19 @@ public CompletableFuture insert(String tableName, } } } + + String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); + LOG.warn(errMsg); + throw (lastException == null ? new ClientException(errMsg) : lastException); + } catch (RuntimeException | Error e) { + spanRecorder.recordFailure(operationSpan, e); + throw e; } finally { // The request of the last attempt stays registered until the operation is over, so a cancellation // landing between two attempts is not lost. unregisterTransportReq(queryId); + operationSpan.end(); } - - String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); - LOG.warn(errMsg); - throw (lastException == null ? new ClientException(errMsg) : lastException); }; return runAsyncOperation(responseSupplier, requestSettings.getAllSettings()); @@ -1824,6 +1874,9 @@ public CompletableFuture query(String sqlQuery, Map responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node @@ -1837,13 +1890,14 @@ public CompletableFuture query(String sqlQuery, Map query(String sqlQuery, Map configuration, Object metricsRegistry, boolean initSslContext, LZ4Factory lz4Factory) { + /** + * Recorder a span is started on per transport request. Never {@code null} - + * {@link com.clickhouse.client.api.observability.DefaultSpanRecorder#NOOP} when observability is + * not configured. + */ + private final SpanRecorder spanRecorder; + + public HttpAPIClientHelper(Map configuration, Object metricsRegistry, boolean initSslContext, + LZ4Factory lz4Factory, SpanRecorder spanRecorder) { + this.spanRecorder = Objects.requireNonNull(spanRecorder, + "spanRecorder is required; use DefaultSpanRecorder.NOOP to record nothing"); this.metricsRegistry = metricsRegistry; this.httpClient = createHttpClient(initSslContext, configuration); this.lz4Factory = lz4Factory; @@ -686,7 +700,63 @@ public InputStream createDataInputStream() { } } + /** + * Executes a single transport request and records it as a child span of the given operation + * span, so a retried operation reports one request span per attempt. The request itself is + * executed by {@link #doExecuteRequest(TransportRequest, Span)}, which stays the single place + * where a request is actually executed and which records the HTTP status of the response. + * + *

+ * When no recorder is registered the request is executed by + * {@link #executeRequest(TransportRequest)}, so a client that does not observe spans keeps + * exactly the behaviour it had before and pays nothing for the span path. + * + * @param transportRequest - request to execute + * @param operationSpan - span of the operation this request is made for + * @return transport response + * @throws Exception when the request could not be completed + */ + public TransportResponse executeRequest(TransportRequest transportRequest, Span operationSpan) throws Exception { + if (spanRecorder == DefaultSpanRecorder.NOOP) { + return executeRequest(transportRequest); + } + + final Span requestSpan = startRequestSpan(operationSpan, transportRequest.getDelegate()); + try { + return doExecuteRequest(transportRequest, requestSpan); + } catch (Exception e) { + spanRecorder.recordRequestFailure(requestSpan, e); + throw e; + } finally { + requestSpan.end(); + } + } + + private Span startRequestSpan(Span operationSpan, HttpPost req) { + final URIAuthority authority = req.getAuthority(); + Span span = spanRecorder.startRequestSpan(operationSpan, + authority == null ? null : authority.getHostName(), + authority == null ? -1 : authority.getPort()); + return span == null ? DefaultSpanRecorder.NOOP_SPAN : span; + } + public TransportResponse executeRequest(TransportRequest transportRequest) throws Exception { + return doExecuteRequest(transportRequest, DefaultSpanRecorder.NOOP_SPAN); + } + + /** + * Executes a single transport request and records the HTTP status on the given request span as + * soon as a response is received - so the status is reported for every response, also for the + * ones this method maps onto an exception that does not carry it (for example {@code 502} and + * {@code 503} onto {@link ConnectException}). + * + * @param transportRequest - request to execute + * @param requestSpan - span of this request; {@link DefaultSpanRecorder#NOOP_SPAN} when the + * request is not recorded + * @return transport response + * @throws Exception when the request could not be completed + */ + private TransportResponse doExecuteRequest(TransportRequest transportRequest, Span requestSpan) throws Exception { final Map requestConfig = transportRequest.getConfig(); final HttpPost req = transportRequest.getDelegate(); @@ -698,6 +768,11 @@ public TransportResponse executeRequest(TransportRequest transportRequest) throw HttpContext context = createRequestHttpContext(requestConfig); try { httpResponse = httpClient.executeOpen(null, req, context); + if (requestSpan != DefaultSpanRecorder.NOOP_SPAN) { + // nothing to report when this request is not recorded - and a recorder is never handed a + // span it did not create + spanRecorder.recordHttpStatus(requestSpan, httpResponse.getCode()); + } httpResponse.setEntity(wrapResponseEntity(httpResponse.getEntity(), httpResponse.getCode(), diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/DefaultSpanRecorder.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/DefaultSpanRecorder.java new file mode 100644 index 000000000..c5cf2cfa5 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/DefaultSpanRecorder.java @@ -0,0 +1,113 @@ +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 com.clickhouse.client.api.transport.Endpoint; + +/** + * Base class for {@link SpanRecorder} implementations. Every method records nothing and every + * {@code start...} method returns {@link #NOOP_SPAN}, so a subclass overrides only what it wants to + * record and keeps working when the client starts a kind of span the subclass does not know about. + *

+ * A subclass creates its own spans and decides what to put on them. To report the client's standard + * span names and attributes it can hand the structures it is given to {@link #getSpanSupport()}: + * 

{@code
+ * public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) {
+ *     SpanSupport support = getSpanSupport();
+ *     MySpan span = new MySpan(support.querySpanName(settings));
+ *     support.fillQueryAttributes(span, settings, sqlQuery, endpoint);
+ *     return span;
+ * }
+ * }
+ * Using it is optional - a recorder that reports something else, or in another form, ignores it, and + * one that wants other values overrides {@link #getSpanSupport()} with its own subclass of + * {@link SpanSupport}. + *

+ * An instance of this class itself records nothing and is what the client uses when no recorder is + * registered. + */ +public class DefaultSpanRecorder implements SpanRecorder { + + /** + * Span that records nothing. Returned by every method of this class and used by the client + * whenever there is nothing to record, so that a span reference is never {@code null}. + */ + public static final Span NOOP_SPAN = new NoopSpan(); + + /** + * Shared instance that records nothing. + */ + public static final DefaultSpanRecorder NOOP = new DefaultSpanRecorder(); + + /** + * Returns the helper a subclass can use to derive the client's standard span names and + * attributes. Override to report other values. + * + * @return span support; never {@code null} + */ + protected SpanSupport getSpanSupport() { + return SpanSupport.DEFAULT; + } + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + return NOOP_SPAN; + } + + @Override + public Span startInsertSpan(InsertSettings settings, String tableName, int batchSize, Endpoint endpoint) { + return NOOP_SPAN; + } + + @Override + public Span startRequestSpan(Span operationSpan, String host, int port) { + return NOOP_SPAN; + } + + @Override + public void recordHttpStatus(Span requestSpan, int statusCode) { + // records nothing + } + + @Override + public void recordSuccess(Span operationSpan, OperationMetrics metrics) { + // records nothing + } + + @Override + public void recordFailure(Span operationSpan, Throwable t) { + // records nothing + } + + @Override + public void recordRequestFailure(Span requestSpan, Throwable t) { + // records nothing + } + + /** + * Span implementation that discards everything reported to it. + */ + private static final class NoopSpan implements Span { + + @Override + public void setAttribute(String key, Object value) { + // records nothing + } + + @Override + public void setError(String errorType) { + // records nothing + } + + @Override + public void end() { + // records nothing + } + + @Override + public String toString() { + return "NoopSpan"; + } + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/Span.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/Span.java new file mode 100644 index 000000000..a71c13871 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/Span.java @@ -0,0 +1,45 @@ +package com.clickhouse.client.api.observability; + +/** + * A single unit of work observed by a {@link SpanRecorder} - either a client operation or one + * transport request made for it. + *

+ * An operation span covers sending the request and receiving the response head; it is ended when + * the operation hands its response to the caller, so it does not cover reading the response body + * (rows are streamed by the caller afterwards). + *

+ * A recorder that does not record a given kind of span returns {@link DefaultSpanRecorder#NOOP_SPAN} + * instead, so the client never has to check for {@code null}. + *

+ * A span is used by a single operation at a time, but the operation span and its request spans may + * be touched from different threads (an operation may run on the shared operation executor), so an + * implementation should not assume single-thread access. + */ +public interface Span { + + /** + * Records an attribute. The keys the client uses are listed in {@link SpanAttribute}; there is a + * single attribute method so that an implementation cannot miss a value by overriding only one + * of several overloads. + * + * @param key - attribute key, see {@link SpanAttribute#getKey()} + * @param value - attribute value; a {@code String}, {@code Number} or {@code Boolean}, never + * {@code null} + */ + void setAttribute(String key, Object value); + + /** + * Marks the span as failed and records {@link SpanAttribute#ERROR_TYPE} with the given value. + * An implementation should map this to its own failure status - for example an OpenTelemetry + * span status of {@code ERROR}. + * + * @param errorType - short, low-cardinality error identifier, usually an exception class name + */ + void setError(String errorType); + + /** + * Ends the span. Called exactly once by the client, also when the operation failed. + * An implementation should be idempotent, so that ending an already-ended span is harmless. + */ + void end(); +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/SpanAttribute.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/SpanAttribute.java new file mode 100644 index 000000000..62b1f805b --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/SpanAttribute.java @@ -0,0 +1,117 @@ +package com.clickhouse.client.api.observability; + +/** + * Attribute keys recorded on {@link Span}s by the client. + *

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

+ * A recorder is registered with + * {@link com.clickhouse.client.api.Client.Builder#setSpanRecorder(SpanRecorder)}. It is the first + * thing the client calls, and it is called with everything the client knows about the operation, so + * an implementation is free to record whatever it needs and however it wants - including nothing. + *

+ * Two kinds of spans are started: + *

    + *
  • Operation span - one per client operation, started with + * {@link #startQuerySpan(QuerySettings, String, Endpoint)} or + * {@link #startInsertSpan(InsertSettings, String, int, Endpoint)}. It is expected to join the + * caller's ambient trace, so client spans appear under the application's own span.
    • + *
  • Request span - one per transport request made for the operation, including every + * retry, started with {@link #startRequestSpan(Span, String, int)}. It is a child of the + * operation span.
    • + *
+ * The client reports the outcome of what it started through the {@code record...} methods and finally + * ends the span with {@link Span#end()}. + *

+ * An implementation does not have to derive the standard span names and attributes itself: it may + * call {@link SpanSupport}, which computes them - the keys listed in {@link SpanAttribute} - 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 DefaultSpanRecorder} and override only what they care about; + * the inherited methods record nothing, so a recorder keeps working when the client starts a kind of + * span it does not know about. + *

+ * A recorder is shared by all operations of a client instance and must be thread-safe. + */ +public interface SpanRecorder { + + /** + * Batch size the client reports when it does not know how many rows an insert sends (stream and + * writer inserts). + */ + int BATCH_SIZE_UNKNOWN = -1; + + /** + * Starts a span for a read operation - a query, a command, a ping or a table-schema lookup. + * + * @param settings - resolved settings of the operation; source of the target database, the query + * id and the statement parameters + * @param sqlQuery - statement sent to the server + * @param endpoint - first configured endpoint, which the operation is expected to use; the endpoint + * of each attempt is reported on its request span instead + * @return new span; never {@code null} + */ + Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint); + + /** + * Starts a span for an insert operation. + * + * @param settings - resolved settings of the operation; source of the target database and the + * query id + * @param tableName - target table + * @param batchSize - number of items in the batch, or {@link #BATCH_SIZE_UNKNOWN} when the client + * does not know it + * @param endpoint - first configured endpoint, which the operation is expected to use; the endpoint + * of each attempt is reported on its request span instead + * @return new span; never {@code null} + */ + Span startInsertSpan(InsertSettings settings, String tableName, int batchSize, Endpoint endpoint); + + /** + * Starts a span for a single transport request made for an operation. Called once per attempt, so + * a retried operation produces several request spans. + * + * @param operationSpan - span of the operation this request belongs to; the returned span should + * be its child + * @param host - server the request is sent to, or {@code null} when it is not known + * @param port - port the request is sent to, or a non-positive value when it is not known + * @return new span; never {@code null} + */ + Span startRequestSpan(Span operationSpan, String host, int port); + + /** + * Reports the HTTP status of the response received for a transport request. Called as soon as a + * response is received, also when the client maps that response onto a failure. + * + * @param requestSpan - span of the request + * @param statusCode - HTTP status code the server answered with + */ + void recordHttpStatus(Span requestSpan, int statusCode); + + /** + * Reports that an operation completed successfully. + * + * @param operationSpan - span of the operation + * @param metrics - metrics of the completed operation; source of the query id and of the number + * of returned rows. May be {@code null} + */ + void recordSuccess(Span operationSpan, OperationMetrics metrics); + + /** + * Reports that an operation failed. + * + * @param operationSpan - span of the operation + * @param t - failure the caller receives + */ + void recordFailure(Span operationSpan, Throwable t); + + /** + * Reports that a single transport request failed. The operation itself may still succeed, because + * the client retries. + * + * @param requestSpan - span of the request + * @param t - failure of this attempt + */ + void recordRequestFailure(Span requestSpan, Throwable t); +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/SpanSupport.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/SpanSupport.java new file mode 100644 index 000000000..71b2726ce --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/SpanSupport.java @@ -0,0 +1,293 @@ +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.HttpAPIClientHelper; +import com.clickhouse.client.api.metrics.Metric; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; +import com.clickhouse.client.api.query.QuerySettings; +import com.clickhouse.client.api.transport.Endpoint; + +import java.util.Map; + +/** + * Derives the client's standard span names and attributes from the structures a + * {@link SpanRecorder} 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 names and the same {@link SpanAttribute} values 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 SpanSupport { + + /** + * Shared instance with the standard behaviour. + */ + public static final SpanSupport DEFAULT = new SpanSupport(); + + /** + * Value of {@link SpanAttribute#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"; + + /** + * Name of a transport request span. All requests the client makes are HTTP {@code POST}s. + */ + public static final String REQUEST_SPAN_NAME = "POST"; + + /** + * Key under which the statement parameters are kept in the request settings. + */ + protected static final String KEY_STATEMENT_PARAMS = HttpAPIClientHelper.KEY_STATEMENT_PARAMS; + + /** + * Returns the name of the span of a read operation, following the OpenTelemetry database span + * naming convention. + * + * @param settings - settings of the operation + * @return span name + */ + public String querySpanName(QuerySettings settings) { + return spanName(OPERATION_QUERY, settings.getDatabase(), null); + } + + /** + * Returns the name of the span of an insert operation, following the OpenTelemetry database span + * naming convention. + * + * @param settings - settings of the operation + * @param tableName - target table + * @return span name + */ + public String insertSpanName(InsertSettings settings, String tableName) { + return spanName(OPERATION_INSERT, settings.getDatabase(), tableName); + } + + /** + * Returns the name of the span of a single transport request. + * + * @return span name + */ + public String requestSpanName() { + return REQUEST_SPAN_NAME; + } + + /** + * Records the attributes of a read operation - a query, a command, a ping or a table-schema + * lookup. Every operation the client implements on top of a query is described as a query; a + * recorder that wants to describe one of them differently derives that from the settings and the + * statement it is given. + * + * @param span - span of the operation + * @param settings - resolved settings of the operation + * @param sqlQuery - statement sent to the server + * @param endpoint - endpoint the operation is expected to use; may be {@code null} + */ + public void fillQueryAttributes(Span span, QuerySettings settings, String sqlQuery, Endpoint endpoint) { + recordCommonAttributes(span, settings.getDatabase(), settings.getQueryId(), null, null, + SpanRecorder.BATCH_SIZE_UNKNOWN, endpoint); + span.setAttribute(SpanAttribute.DB_QUERY_TEXT.getKey(), sqlQuery); + recordStatementParams(span, settings.getAllSettings()); + } + + /** + * Records the attributes of an insert operation. + * + * @param span - span of the operation + * @param settings - resolved settings of the operation + * @param tableName - target table + * @param batchSize - number of items in the batch, or {@link SpanRecorder#BATCH_SIZE_UNKNOWN} + * when the client does not know it + * @param endpoint - endpoint the operation is expected to use; may be {@code null} + */ + public void fillInsertAttributes(Span span, InsertSettings settings, String tableName, int batchSize, + Endpoint endpoint) { + recordCommonAttributes(span, settings.getDatabase(), settings.getQueryId(), OPERATION_INSERT, + tableName, batchSize, endpoint); + } + + /** + * Records the attributes of a single transport request. + * + * @param span - span of the request + * @param host - server the request is sent to, or {@code null} when it is not known + * @param port - port the request is sent to, or a non-positive value when it is not known + */ + public void fillRequestAttributes(Span span, String host, int port) { + span.setAttribute(SpanAttribute.HTTP_REQUEST_METHOD.getKey(), REQUEST_SPAN_NAME); + recordEndpoint(span, host, port); + } + + /** + * Records the HTTP status returned for a transport request. + * + * @param span - span of the request + * @param statusCode - HTTP status code the server answered with + */ + public void recordHttpStatus(Span span, int statusCode) { + span.setAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE.getKey(), statusCode); + } + + /** + * Records the server a request is sent to. Recorded per attempt, because a retry may go to + * another node. + * + * @param span - span to record on + * @param host - server hostname, or {@code null} when it is not known + * @param port - server port, or a non-positive value when it is not known + */ + public void recordEndpoint(Span span, String host, int port) { + if (host != null) { + span.setAttribute(SpanAttribute.SERVER_ADDRESS.getKey(), host); + } + if (port > 0) { + span.setAttribute(SpanAttribute.SERVER_PORT.getKey(), port); + } + } + + /** + * Records the outcome of a successfully completed operation. + * + * @param span - span of the operation + * @param metrics - metrics of the completed operation, may be {@code null} + */ + public void recordSuccess(Span span, OperationMetrics metrics) { + if (metrics == null) { + return; + } + + if (metrics.getQueryId() != null) { + span.setAttribute(SpanAttribute.CLICKHOUSE_QUERY_ID.getKey(), metrics.getQueryId()); + } + // the row count comes from the server's progress summary, which is not always available + Metric returnedRows = metrics.getMetric(ServerMetrics.RESULT_ROWS); + if (returnedRows != null && returnedRows.getLong() >= 0) { + span.setAttribute(SpanAttribute.DB_RESPONSE_RETURNED_ROWS.getKey(), returnedRows.getLong()); + } + } + + /** + * Records the failure of a single transport request. In addition to + * {@link #recordFailure(Span, Throwable)} the HTTP status is recorded when the server answered + * with an error response that carries it. + * + * @param span - span of the request + * @param t - failure + */ + public void recordRequestFailure(Span span, Throwable t) { + if (t == null) { + return; + } + + ServerException serverException = findServerException(t); + if (serverException != null && serverException.getTransportProtocolCode() > 0) { + recordHttpStatus(span, serverException.getTransportProtocolCode()); + } + recordFailure(span, t); + } + + /** + * Records the failure of an operation or of a single transport request. The ClickHouse error code + * is recorded when the server reported one. + * + * @param span - span to record on + * @param t - failure + */ + public void recordFailure(Span span, Throwable t) { + if (t == null) { + return; + } + + ServerException serverException = findServerException(t); + if (serverException == null) { + span.setError(t.getClass().getName()); + } else { + span.setAttribute(SpanAttribute.DB_RESPONSE_STATUS_CODE.getKey(), serverException.getCode()); + span.setError(serverException.getClass().getName()); + } + } + + /** + * Records the attributes that describe the operation itself. + */ + protected void recordCommonAttributes(Span span, String namespace, String queryId, String operationName, + String collectionName, int batchSize, Endpoint endpoint) { + span.setAttribute(SpanAttribute.DB_SYSTEM_NAME.getKey(), DB_SYSTEM_NAME); + if (namespace != null) { + span.setAttribute(SpanAttribute.DB_NAMESPACE.getKey(), namespace); + } + if (queryId != null) { + span.setAttribute(SpanAttribute.CLICKHOUSE_QUERY_ID.getKey(), queryId); + } + if (operationName != null) { + span.setAttribute(SpanAttribute.DB_OPERATION_NAME.getKey(), operationName); + } + if (collectionName != null) { + span.setAttribute(SpanAttribute.DB_COLLECTION_NAME.getKey(), collectionName); + } + if (batchSize >= 0) { + span.setAttribute(SpanAttribute.DB_OPERATION_BATCH_SIZE.getKey(), batchSize); + } + if (endpoint != null) { + recordEndpoint(span, endpoint.getHost(), endpoint.getPort()); + } + } + + /** + * Records the values of the statement parameters sent with a query. + */ + @SuppressWarnings("unchecked") + protected void recordStatementParams(Span span, Map settings) { + Object params = settings.get(KEY_STATEMENT_PARAMS); + if (!(params instanceof Map)) { + return; + } + for (Map.Entry param : ((Map) params).entrySet()) { + span.setAttribute(SpanAttribute.DB_QUERY_PARAMETER.getKey(param.getKey()), param.getValue()); + } + } + + /** + * 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; + } + + /** + * Builds a span name out of the operation name and its target, following the OpenTelemetry + * database span naming convention - {@code .}. + * + * @param operationName - name of the operation + * @param namespace - target database, may be {@code null} + * @param collectionName - target table, may be {@code null} + * @return span name + */ + protected String spanName(String operationName, String namespace, String collectionName) { + boolean hasNamespace = namespace != null && !namespace.isEmpty(); + if (collectionName != null && !collectionName.isEmpty()) { + return hasNamespace + ? operationName + " " + namespace + "." + collectionName + : operationName + " " + collectionName; + } + return hasNamespace ? operationName + " " + namespace : operationName; + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java index 8e2fc4c65..5dc1e6269 100644 --- a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java @@ -22,6 +22,7 @@ import com.clickhouse.client.api.insert.InsertSettings; import com.clickhouse.client.api.internal.DataTypeConverter; import com.clickhouse.client.api.internal.HttpAPIClientHelper; +import com.clickhouse.client.api.internal.HttpAPIClientHelperFactory; import com.clickhouse.client.api.internal.ServerSettings; import com.clickhouse.client.api.internal.ValidationUtils; import com.clickhouse.client.api.query.GenericRecord; @@ -2695,8 +2696,7 @@ public void testTransportRequestCancel() throws Exception { configuration.put(ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getKey(), Boolean.FALSE); configuration.put(ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getKey(), Boolean.FALSE); - HttpAPIClientHelper helper = new HttpAPIClientHelper(new HashMap<>(configuration), null, false, - LZ4Factory.fastestInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(new HashMap<>(configuration), LZ4Factory.fastestInstance()); try (Client verifyClient = new Client.Builder() .addEndpoint(Protocol.HTTP, server.getHost(), server.getPort(), false) diff --git a/client-v2/src/test/java/com/clickhouse/client/InsertExceptionClassificationTest.java b/client-v2/src/test/java/com/clickhouse/client/InsertExceptionClassificationTest.java index 786cf955e..f9fa0a383 100644 --- a/client-v2/src/test/java/com/clickhouse/client/InsertExceptionClassificationTest.java +++ b/client-v2/src/test/java/com/clickhouse/client/InsertExceptionClassificationTest.java @@ -4,6 +4,7 @@ import com.clickhouse.client.api.DataTransferException; import com.clickhouse.client.api.internal.HttpAPIClientHelper; import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.observability.DefaultSpanRecorder; import com.clickhouse.client.api.serde.DataSerializationException; import com.clickhouse.client.api.transport.Endpoint; import com.clickhouse.client.api.transport.internal.TransportRequest; @@ -100,7 +101,7 @@ private static final class CallbackHttpClientHelper extends HttpAPIClientHelper private final OutputStream outputStream; private IOCallback writeCallback; private CallbackHttpClientHelper(OutputStream outputStream) { - super(Collections.emptyMap(), null, false, LZ4Factory.fastestJavaInstance()); + super(Collections.emptyMap(), null, false, LZ4Factory.fastestJavaInstance(), DefaultSpanRecorder.NOOP); this.outputStream = outputStream; } diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperFactory.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperFactory.java new file mode 100644 index 000000000..d24fdd08c --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperFactory.java @@ -0,0 +1,29 @@ +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.observability.DefaultSpanRecorder; +import net.jpountz.lz4.LZ4Factory; + +import java.util.Map; + +/** + * Builds {@link HttpAPIClientHelper} instances for tests. Tests construct the helper through this + * factory instead of calling the constructor directly so that a change to the constructor signature + * is applied in one place instead of in every test. + */ +public final class HttpAPIClientHelperFactory { + + private HttpAPIClientHelperFactory() { + } + + /** + * Creates a helper with no metrics registry, no SSL context and observability disabled - the + * configuration every test that does not exercise those features needs. + * + * @param configuration - client configuration + * @param lz4Factory - LZ4 factory the helper should use + * @return a new helper instance + */ + public static HttpAPIClientHelper newHelper(Map configuration, LZ4Factory lz4Factory) { + return new HttpAPIClientHelper(configuration, null, false, lz4Factory, DefaultSpanRecorder.NOOP); + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java index ca8b732e0..16841dc6e 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java @@ -243,7 +243,7 @@ public void testCreateHttpClientEmptyCipherSuitesTreatedAsNoRestriction() { @Test public void testExecuteRequestThrowsConnectExceptionOn502() throws Exception { Map configuration = new HashMap<>(); - HttpAPIClientHelper helper = new HttpAPIClientHelper(configuration, null, false, LZ4Factory.fastestInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(configuration, LZ4Factory.fastestInstance()); CloseableHttpClient mockHttpClient = mock(CloseableHttpClient.class); Field httpClientField = HttpAPIClientHelper.class.getDeclaredField("httpClient"); @@ -272,7 +272,7 @@ public void testExecuteRequestThrowsConnectExceptionOn502() throws Exception { @Test public void testExecuteRequestThrowsConnectExceptionOn503() throws Exception { Map configuration = new HashMap<>(); - HttpAPIClientHelper helper = new HttpAPIClientHelper(configuration, null, false, LZ4Factory.fastestInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(configuration, LZ4Factory.fastestInstance()); CloseableHttpClient mockHttpClient = mock(CloseableHttpClient.class); Field httpClientField = HttpAPIClientHelper.class.getDeclaredField("httpClient"); @@ -322,7 +322,7 @@ public static Object[][] serverExceptionRetryCases() { */ @Test(dataProvider = "serverExceptionRetryCases") public void testShouldRetryUsesServerExceptionFromCause(Throwable ex, boolean expectedRetry) { - HttpAPIClientHelper helper = new HttpAPIClientHelper(new HashMap<>(), null, false, LZ4Factory.fastestInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(new HashMap<>(), LZ4Factory.fastestInstance()); // Empty request settings -> default client_retry_on_failures, which includes ServerRetryable. assertEquals(helper.shouldRetry(ex, new HashMap<>()), expectedRetry); } @@ -346,7 +346,7 @@ public static Object[][] serverErrorLogging() { @Test(dataProvider = "serverErrorLogging") public void testServerErrorLoggedOnlyForUnknownStatus(int statusCode, String exceptionCode, boolean expectServerErrorWarn) throws Exception { - HttpAPIClientHelper helper = new HttpAPIClientHelper(new HashMap<>(), null, false, LZ4Factory.fastestInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(new HashMap<>(), LZ4Factory.fastestInstance()); injectMockHttpClient(helper, mockResponse(statusCode, exceptionCode)); Map reqConfig = new HashMap<>(); @@ -378,7 +378,7 @@ public void testServerErrorLoggedOnlyForUnknownStatus(int statusCode, String exc */ @Test public void testLogServerErrorResponseIsNullSafeAndLogsExceptionCode() throws Exception { - HttpAPIClientHelper helper = new HttpAPIClientHelper(new HashMap<>(), null, false, LZ4Factory.fastestInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(new HashMap<>(), LZ4Factory.fastestInstance()); Method log = HttpAPIClientHelper.class.getDeclaredMethod( "logServerErrorResponse", HttpPost.class, ClassicHttpResponse.class); log.setAccessible(true); @@ -460,8 +460,7 @@ private static String captureStdErr(Runnable action) { * constructor arguments of each construction (empty when the plain factory branch is taken instead). */ private static List> captureCustomFactoryConstruction(Map sslConfig) { - HttpAPIClientHelper helper = new HttpAPIClientHelper(new HashMap<>(), null, false, - LZ4Factory.fastestJavaInstance()); + HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(new HashMap<>(), LZ4Factory.fastestJavaInstance()); List> constructorArgs = new ArrayList<>(); try (MockedConstruction mocked = mockConstruction( CustomSSLConnectionFactory.class, diff --git a/client-v2/src/test/java/com/clickhouse/client/api/observability/CapturingSpanRecorder.java b/client-v2/src/test/java/com/clickhouse/client/api/observability/CapturingSpanRecorder.java new file mode 100644 index 000000000..1acf24a91 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/observability/CapturingSpanRecorder.java @@ -0,0 +1,190 @@ +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 com.clickhouse.client.api.transport.Endpoint; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Recorder that keeps every span it starts, so tests can assert what the client reported. It takes the + * names and the attributes from {@link SpanSupport}, which is how a recorder opts in to the client's + * standard values. + */ +public class CapturingSpanRecorder extends DefaultSpanRecorder { + + private final List spans = Collections.synchronizedList(new ArrayList<>()); + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + SpanSupport support = getSpanSupport(); + CapturedSpan span = add(new CapturedSpan(support.querySpanName(settings), null, + settings.getDatabase(), settings.getQueryId())); + support.fillQueryAttributes(span, settings, sqlQuery, endpoint); + return span; + } + + @Override + public Span startInsertSpan(InsertSettings settings, String tableName, int batchSize, Endpoint endpoint) { + SpanSupport support = getSpanSupport(); + CapturedSpan span = add(new CapturedSpan(support.insertSpanName(settings, tableName), null, + settings.getDatabase(), settings.getQueryId())); + support.fillInsertAttributes(span, settings, tableName, batchSize, endpoint); + return span; + } + + @Override + public Span startRequestSpan(Span operationSpan, String host, int port) { + SpanSupport support = getSpanSupport(); + CapturedSpan span = add(new CapturedSpan(support.requestSpanName(), operationSpan, null, null)); + support.fillRequestAttributes(span, host, port); + return span; + } + + @Override + public void recordHttpStatus(Span requestSpan, int statusCode) { + getSpanSupport().recordHttpStatus(requestSpan, statusCode); + } + + @Override + public void recordSuccess(Span operationSpan, OperationMetrics metrics) { + getSpanSupport().recordSuccess(operationSpan, metrics); + } + + @Override + public void recordFailure(Span operationSpan, Throwable t) { + getSpanSupport().recordFailure(operationSpan, t); + } + + @Override + public void recordRequestFailure(Span requestSpan, Throwable t) { + getSpanSupport().recordRequestFailure(requestSpan, t); + } + + private CapturedSpan add(CapturedSpan span) { + spans.add(span); + return span; + } + + public List getSpans() { + synchronized (spans) { + return new ArrayList<>(spans); + } + } + + /** + * Returns the only operation span recorded so far. + */ + public CapturedSpan operationSpan() { + List operations = new ArrayList<>(); + for (CapturedSpan span : getSpans()) { + if (span.getParent() == null) { + operations.add(span); + } + } + if (operations.size() != 1) { + throw new AssertionError("Expected exactly one operation span but got " + operations); + } + return operations.get(0); + } + + /** + * Returns the request spans started for the given operation span, in the order they were started. + */ + public List requestSpans(CapturedSpan operationSpan) { + List requests = new ArrayList<>(); + for (CapturedSpan span : getSpans()) { + if (span.getParent() == operationSpan) { + requests.add(span); + } + } + return requests; + } + + public void clear() { + spans.clear(); + } + + public static final class CapturedSpan implements Span { + + private final String name; + private final Span parent; + private final String settingsDatabase; + private final String settingsQueryId; + private final Map attributes = Collections.synchronizedMap(new LinkedHashMap<>()); + private final AtomicInteger endCount = new AtomicInteger(); + private volatile String errorType; + + CapturedSpan(String name, Span parent, String settingsDatabase, String settingsQueryId) { + this.name = name; + this.parent = parent; + this.settingsDatabase = settingsDatabase; + this.settingsQueryId = settingsQueryId; + } + + @Override + public void setAttribute(String key, Object value) { + attributes.put(key, value); + } + + @Override + public void setError(String errorType) { + this.errorType = errorType; + } + + @Override + public void end() { + endCount.incrementAndGet(); + } + + public String getName() { + return name; + } + + public Span getParent() { + return parent; + } + + public String getSettingsDatabase() { + return settingsDatabase; + } + + public String getSettingsQueryId() { + return settingsQueryId; + } + + public Map getAttributes() { + synchronized (attributes) { + return new LinkedHashMap<>(attributes); + } + } + + public Object getAttribute(SpanAttribute attribute) { + return attributes.get(attribute.getKey()); + } + + public Object getAttribute(String key) { + return attributes.get(key); + } + + public String getErrorType() { + return errorType; + } + + public int getEndCount() { + return endCount.get(); + } + + @Override + public String toString() { + return "CapturedSpan{name='" + name + "', attributes=" + getAttributes() + + ", errorType=" + errorType + ", endCount=" + endCount.get() + '}'; + } + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/observability/SpanRecorderUnitTest.java b/client-v2/src/test/java/com/clickhouse/client/api/observability/SpanRecorderUnitTest.java new file mode 100644 index 000000000..d1763b92b --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/observability/SpanRecorderUnitTest.java @@ -0,0 +1,570 @@ +package com.clickhouse.client.api.observability; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.insert.InsertResponse; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.observability.CapturingSpanRecorder.CapturedSpan; +import com.clickhouse.client.api.metrics.OperationMetrics; +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.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +public class SpanRecorderUnitTest { + + private static final String DEAD_ENDPOINT = "http://127.0.0.1:1"; // nothing listens here + + private CapturingSpanRecorder recorder; + private WireMockServer mockServer; + + @BeforeMethod + void setUp() { + recorder = new CapturingSpanRecorder(); + 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 testQuerySpanReportsStatementAndParameters() throws Exception { + Map params = new HashMap<>(); + params.put("id", 42); + params.put("phrase", "hello"); + + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + QuerySettings settings = new QuerySettings().setQueryId("query-id-1"); + try (QueryResponse response = client.query("SELECT {id:UInt8}, {phrase:String}", params, settings) + .get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "query test_db"); + Assert.assertEquals(operationSpan.getSettingsDatabase(), "test_db", + "the recorder must receive the resolved operation settings"); + Assert.assertEquals(operationSpan.getSettingsQueryId(), "query-id-1"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_SYSTEM_NAME), "clickhouse"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_NAMESPACE), "test_db"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), + "SELECT {id:UInt8}, {phrase:String}"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.CLICKHOUSE_QUERY_ID), "query-id-1"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_PARAMETER.getKey("id")), "42"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_PARAMETER.getKey("phrase")), "hello"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.SERVER_ADDRESS), "localhost"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.SERVER_PORT), mockServer.port()); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_NAME), + "a plain query has no operation name"); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_COLLECTION_NAME)); + Assert.assertNull(operationSpan.getErrorType()); + Assert.assertEquals(operationSpan.getEndCount(), 1); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), 1); + CapturedSpan requestSpan = requestSpans.get(0); + Assert.assertEquals(requestSpan.getName(), "POST"); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.HTTP_REQUEST_METHOD), "POST"); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE), 200); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.SERVER_ADDRESS), "localhost"); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.SERVER_PORT), mockServer.port()); + Assert.assertNull(requestSpan.getErrorType()); + Assert.assertEquals(requestSpan.getEndCount(), 1); + } + + @Test + public void testPojoInsertSpanReportsBatchSize() throws Exception { + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + client.register(ValuePojo.class, new TableSchema("target_table", null, "", + Collections.singletonList(ClickHouseColumn.of("value", "String")))); + try (InsertResponse response = client.insert("target_table", + Arrays.asList(new ValuePojo("a"), new ValuePojo("b"), new ValuePojo("c"))) + .get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "insert test_db.target_table"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_NAME), "insert"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_COLLECTION_NAME), "target_table"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_NAMESPACE), "test_db"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_BATCH_SIZE), 3); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), + "an insert does not report a statement"); + Assert.assertEquals(operationSpan.getEndCount(), 1); + Assert.assertEquals(recorder.requestSpans(operationSpan).size(), 1); + } + + @Test + public void testStreamInsertSpanHasNoBatchSize() throws Exception { + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).build()) { + try (InsertResponse response = client.insert("target_table", + new ByteArrayInputStream("value\n".getBytes(StandardCharsets.UTF_8)), + ClickHouseFormat.CSV, new InsertSettings()).get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "insert test_db.target_table"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_COLLECTION_NAME), "target_table"); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_BATCH_SIZE), + "the number of rows in a stream is not known to the client"); + Assert.assertEquals(operationSpan.getEndCount(), 1); + } + + @DataProvider(name = "retryCounts") + public static Object[][] retryCounts() { + return new Object[][]{{0, 1}, {1, 2}, {3, 4}}; + } + + @Test(dataProvider = "retryCounts") + public void testEveryAttemptIsRecordedAsChildRequestSpan(int maxRetries, int expectedAttempts) throws Exception { + String expectedErrorType = null; + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(maxRetries).build()) { + try { + client.query("SELECT 1").get(30, TimeUnit.SECONDS); + 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(); + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getErrorType(), expectedErrorType); + Assert.assertEquals(operationSpan.getEndCount(), 1, "an operation span is ended exactly once"); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), expectedAttempts, "one request span per attempt"); + for (CapturedSpan requestSpan : requestSpans) { + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.SERVER_ADDRESS), "127.0.0.1"); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.SERVER_PORT), 1); + Assert.assertNull(requestSpan.getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE), + "no response was received"); + Assert.assertNotNull(requestSpan.getErrorType()); + Assert.assertEquals(requestSpan.getEndCount(), 1); + } + } + + @Test + public void testFirstEndpointOnOperationSpanAndPerAttemptOnRequestSpans() throws Exception { + try (Client client = newClientBuilder() + .addEndpoint(DEAD_ENDPOINT) + .addEndpoint(mockEndpoint()) + .setMaxRetries(3) + .build()) { + try (QueryResponse response = client.query("SELECT 1").get(30, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.SERVER_ADDRESS), "127.0.0.1", + "the operation reports the first configured endpoint; every attempt reports its own"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.SERVER_PORT), 1); + Assert.assertNull(operationSpan.getErrorType()); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertFalse(requestSpans.isEmpty()); + for (CapturedSpan requestSpan : requestSpans) { + Assert.assertNotNull(requestSpan.getAttribute(SpanAttribute.SERVER_ADDRESS)); + Assert.assertNotNull(requestSpan.getAttribute(SpanAttribute.SERVER_PORT)); + } + CapturedSpan lastRequestSpan = requestSpans.get(requestSpans.size() - 1); + Assert.assertEquals(lastRequestSpan.getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE), 200); + Assert.assertEquals(lastRequestSpan.getAttribute(SpanAttribute.SERVER_PORT), mockServer.port()); + } + + @Test + public void testFailedInsertRecordsErrorOnOperationAndRequestSpans() throws Exception { + String expectedErrorType = null; + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(1).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(); + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "insert test_db.target_table"); + Assert.assertEquals(operationSpan.getErrorType(), expectedErrorType); + Assert.assertEquals(operationSpan.getEndCount(), 1); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), 2, "one request span per attempt"); + for (CapturedSpan requestSpan : requestSpans) { + Assert.assertNotNull(requestSpan.getErrorType()); + Assert.assertEquals(requestSpan.getEndCount(), 1); + } + } + + @DataProvider(name = "mappedErrorStatuses") + public static Object[][] mappedErrorStatuses() { + // statuses the transport maps onto an exception that does not carry the HTTP status itself + return new Object[][] { + {407}, // proxy authentication required -> ClientMisconfigurationException + {502}, // bad gateway -> ConnectException + {503}, // service unavailable -> ConnectException + {418}, // unknown status -> ClientException + }; + } + + @Test(dataProvider = "mappedErrorStatuses") + public void testRequestSpanReportsHttpStatusOfMappedErrorResponses(int status) throws Exception { + // the request span reports the HTTP status of every response the server sent, also when the + // transport maps that response onto an exception that does not carry the status + mockServer.resetAll(); + mockServer.stubFor(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse().withStatus(status) + .withHeader("Content-Type", "text/plain") + .withBody(""))); + + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()).setMaxRetries(0).build()) { + try { + client.query("SELECT 1").get(10, TimeUnit.SECONDS).close(); + Assert.fail("a query answered with status " + status + " must fail"); + } catch (ExecutionException | RuntimeException e) { + // expected - the response is mapped onto a failure + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertNotNull(operationSpan.getErrorType(), "the failed operation must report an error type"); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), 1, "one request span per attempt"); + CapturedSpan requestSpan = requestSpans.get(0); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE), status, + "the status of the received response must be reported on the request span"); + Assert.assertNotNull(requestSpan.getErrorType(), "the failed request must report an error type"); + Assert.assertEquals(requestSpan.getEndCount(), 1); + } + + @Test + public void testRequestSpanReportsNoHttpStatusWhenNoResponseArrives() throws Exception { + // contrast case: without a response there is no status to report + try (Client client = newClientBuilder().addEndpoint(DEAD_ENDPOINT).setMaxRetries(0).build()) { + try { + client.query("SELECT 1").get(30, TimeUnit.SECONDS).close(); + Assert.fail("a query to a dead endpoint must fail"); + } catch (ExecutionException | RuntimeException e) { + // expected - nothing listens on the endpoint + } + } + + CapturedSpan operationSpan = recorder.operationSpan(); + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), 1); + CapturedSpan requestSpan = requestSpans.get(0); + Assert.assertNull(requestSpan.getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE), + "no response arrived, so no HTTP status is reported"); + Assert.assertNotNull(requestSpan.getErrorType()); + Assert.assertEquals(requestSpan.getEndCount(), 1); + } + + @Test + public void testOperationSpanIsStartedOnCallingThreadWithAsyncRequests() throws Exception { + Thread callingThread = Thread.currentThread(); + ThreadRecordingSpanRecorder threadRecorder = new ThreadRecordingSpanRecorder(); + try (Client client = new Client.Builder() + .addEndpoint(mockEndpoint()) + .setUsername("default") + .setPassword("") + .setDefaultDatabase("test_db") + .useAsyncRequests(true) + .setSpanRecorder(threadRecorder) + .build()) { + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + Assert.assertEquals(threadRecorder.operationSpan().getName(), "query test_db"); + Assert.assertSame(threadRecorder.operationStartThread, callingThread, + "an operation span must be started on the caller's thread so it joins the caller's trace"); + Assert.assertNotSame(threadRecorder.requestStartThread, callingThread, + "the request itself is executed on the operation executor"); + Assert.assertEquals(threadRecorder.operationSpan().getEndCount(), 1); + Assert.assertEquals(threadRecorder.requestSpans(threadRecorder.operationSpan()).size(), 1); + } + + @Test + public void testNullRecorderIsRejected() { + // the default recorder already records nothing, so a null recorder is a configuration error + Assert.assertThrows(NullPointerException.class, + () -> newClientBuilder().addEndpoint(mockEndpoint()).setSpanRecorder(null)); + } + + @Test + public void testNothingIsRecordedWithTheDefaultRecorder() throws Exception { + try (Client client = newClientBuilder().addEndpoint(mockEndpoint()) + .setSpanRecorder(DefaultSpanRecorder.NOOP).build()) { + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + Assert.assertTrue(recorder.getSpans().isEmpty(), + "a recorder that records nothing replaces a previously set one and records nothing"); + } + + @Test + public void testDefaultSpanRecorderRecordsNothing() { + SpanRecorder defaultRecorder = new DefaultSpanRecorder(); + Assert.assertSame(defaultRecorder.startQuerySpan(new QuerySettings(), "SELECT 1", null), + DefaultSpanRecorder.NOOP_SPAN); + Assert.assertSame(defaultRecorder.startInsertSpan(new InsertSettings(), "t1", 3, null), + DefaultSpanRecorder.NOOP_SPAN); + Assert.assertSame(defaultRecorder.startRequestSpan(DefaultSpanRecorder.NOOP_SPAN, "localhost", 8123), + DefaultSpanRecorder.NOOP_SPAN); + + Span noopSpan = DefaultSpanRecorder.NOOP_SPAN; + // the base class records nothing for every outcome the client reports + defaultRecorder.recordHttpStatus(noopSpan, 200); + defaultRecorder.recordSuccess(noopSpan, null); + defaultRecorder.recordFailure(noopSpan, new IllegalStateException("boom")); + defaultRecorder.recordRequestFailure(noopSpan, new IllegalStateException("boom")); + noopSpan.setAttribute(SpanAttribute.DB_NAMESPACE.getKey(), "db"); + noopSpan.setError("java.lang.IllegalStateException"); + noopSpan.end(); + } + + @Test + public void testRecorderMayRecordQuerySpansOnly() throws Exception { + // a recorder that overrides one kind of span only must keep working - the kinds it does not + // override are answered by the base class with a span that records nothing + QueryOnlySpanRecorder queryOnlyRecorder = new QueryOnlySpanRecorder(); + try (Client client = new Client.Builder() + .setUsername("default") + .setPassword("") + .setDefaultDatabase("test_db") + .setSpanRecorder(queryOnlyRecorder) + .addEndpoint(mockEndpoint()) + .build()) { + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + client.insert("t1", new ByteArrayInputStream("1\n".getBytes(StandardCharsets.UTF_8)), + ClickHouseFormat.TSV).get(10, TimeUnit.SECONDS).close(); + } + + Assert.assertEquals(queryOnlyRecorder.querySpans.size(), 1, + "the overridden method is the only one that recorded a span"); + CapturedSpan querySpan = queryOnlyRecorder.querySpans.get(0); + Assert.assertEquals(querySpan.getName(), "query test_db"); + Assert.assertEquals(querySpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), "SELECT 1"); + Assert.assertEquals(querySpan.getEndCount(), 1); + } + + @Test + public void testClientCallsTheRecorderAndNotTheSupport() throws Exception { + // the client must go through the registered recorder for everything it reports, so a recorder + // that reports its own values never has SpanSupport applied to its spans - the support is opt-in + OwnValuesSpanRecorder ownValuesRecorder = new OwnValuesSpanRecorder(); + try (Client client = new Client.Builder() + .setUsername("default") + .setPassword("") + .setDefaultDatabase("test_db") + .setSpanRecorder(ownValuesRecorder) + .addEndpoint(mockEndpoint()) + .build()) { + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response, "a recorder that ignores SpanSupport must not fail the operation"); + } + } + + Assert.assertEquals(ownValuesRecorder.spans.size(), 2, "an operation span and one request span"); + CapturedSpan operationSpan = ownValuesRecorder.spans.get(0); + Assert.assertEquals(operationSpan.getAttributes().size(), 1, + "only the attribute the recorder set itself is reported: " + operationSpan.getAttributes()); + Assert.assertEquals(operationSpan.getAttribute("my.statement"), "SELECT 1"); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_SYSTEM_NAME), + "the client must not apply the standard attributes behind the recorder's back"); + Assert.assertEquals(operationSpan.getEndCount(), 1); + } + + @Test + public void testRecorderReturningNullSpansDoesNotBreakOperations() throws Exception { + SpanRecorder nullRecorder = new SpanRecorder() { + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + return null; + } + + @Override + public Span startInsertSpan(InsertSettings settings, String tableName, int batchSize, + Endpoint endpoint) { + return null; + } + + @Override + public Span startRequestSpan(Span operationSpan, String host, int port) { + return null; + } + + @Override + public void recordHttpStatus(Span requestSpan, int statusCode) { + // records nothing + } + + @Override + public void recordSuccess(Span operationSpan, OperationMetrics metrics) { + // records nothing + } + + @Override + public void recordFailure(Span operationSpan, Throwable t) { + // records nothing + } + + @Override + public void recordRequestFailure(Span requestSpan, Throwable t) { + // records nothing + } + }; + + try (Client client = new Client.Builder() + .setUsername("default") + .setPassword("") + .setDefaultDatabase("test_db") + .setSpanRecorder(nullRecorder) + .addEndpoint(mockEndpoint()) + .build()) { + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response, "a recorder that returns no span must not fail the operation"); + } + } + } + + /** + * Recorder that records its own values only. It fails when the client makes it go through + * {@link com.clickhouse.client.api.observability.SpanSupport}, so the test can assert that using the + * support stays the implementation's own choice. + */ + private static class OwnValuesSpanRecorder extends DefaultSpanRecorder { + final List spans = Collections.synchronizedList(new ArrayList<>()); + + @Override + protected SpanSupport getSpanSupport() { + throw new AssertionError("the client must not make a recorder use SpanSupport"); + } + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + CapturedSpan span = new CapturedSpan("my " + settings.getDatabase(), null, + settings.getDatabase(), settings.getQueryId()); + span.setAttribute("my.statement", sqlQuery); + spans.add(span); + return span; + } + + @Override + public Span startRequestSpan(Span operationSpan, String host, int port) { + CapturedSpan span = new CapturedSpan("my request", operationSpan, null, null); + spans.add(span); + return span; + } + } + + /** + * Recorder that implements the query span only and inherits everything else from the base class. + */ + private static class QueryOnlySpanRecorder extends DefaultSpanRecorder { + final List querySpans = Collections.synchronizedList(new ArrayList<>()); + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + CapturedSpan span = new CapturedSpan(getSpanSupport().querySpanName(settings), null, + settings.getDatabase(), settings.getQueryId()); + getSpanSupport().fillQueryAttributes(span, settings, sqlQuery, endpoint); + querySpans.add(span); + return span; + } + } + + private static class ThreadRecordingSpanRecorder extends CapturingSpanRecorder { + volatile Thread operationStartThread; + volatile Thread requestStartThread; + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + operationStartThread = Thread.currentThread(); + return super.startQuerySpan(settings, sqlQuery, endpoint); + } + + @Override + public Span startRequestSpan(Span operationSpan, String host, int port) { + requestStartThread = Thread.currentThread(); + return super.startRequestSpan(operationSpan, host, port); + } + } + + private Client.Builder newClientBuilder() { + return new Client.Builder() + .setUsername("default") + .setPassword("") + .setDefaultDatabase("test_db") + .setSpanRecorder(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/api/transport/TransportBaseTests.java b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java index 6d34d7e06..092c7bc35 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -14,6 +14,9 @@ import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.observability.DefaultSpanRecorder; +import com.clickhouse.client.api.observability.Span; +import com.clickhouse.client.api.observability.SpanRecorder; import com.clickhouse.client.api.query.GenericRecord; import com.clickhouse.client.api.query.QueryResponse; import com.clickhouse.client.api.query.QuerySettings; @@ -33,9 +36,11 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.lang.reflect.Field; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -691,6 +696,225 @@ public void testFirstAttemptNotStoppedByAnotherCancelledOperation() throws Excep } } + /** + * The transport request of an operation must stay registered for the whole retry loop: the registration is + * what {@link Client#cancelTransportRequest(String)} resolves a query id against, so unregistering after + * every attempt left the query id resolving to nothing at all between two attempts. Probed deterministically + * through {@link DataStreamWriter#onRetry()}, which the client calls between two attempts of a stream insert - + * exactly that window. + */ + @Test(groups = {"integration"}) + public void testTransportRequestStaysRegisteredBetweenRetries() throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + stubRetryableThenSuccess(mockServer); + + final String queryId = "retry-registry-" + UUID.randomUUID(); + AtomicInteger retries = new AtomicInteger(); + AtomicBoolean registeredOnRetry = new AtomicBoolean(); + + try (Client client = mockServerClient(mockServer, 1)) { + DataStreamWriter writer = new DataStreamWriter() { + @Override + public void onOutput(OutputStream out) throws IOException { + out.write("1\t2\t3\n".getBytes(StandardCharsets.US_ASCII)); + } + + @Override + public void onRetry() { + retries.incrementAndGet(); + registeredOnRetry.set(ongoingRequests(client).containsKey(queryId)); + } + }; + + try (InsertResponse response = client.insert("table01", writer, ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + Assert.assertNotNull(response, "the insert should have recovered after a retry"); + } + + Assert.assertEquals(retries.get(), 1, "expected exactly one retry between the two attempts"); + Assert.assertTrue(registeredOnRetry.get(), + "the transport request must still be registered between two attempts so that " + + "cancelTransportRequest() keeps working while the operation retries"); + } finally { + mockServer.stop(); + } + } + + /** + * Counterpart of {@link #testTransportRequestStaysRegisteredBetweenRetries()}: keeping the registration for + * the whole retry loop must not leak it. Every operation has to unregister its transport request once it + * finished, whether it succeeded on a retry or failed after exhausting the retries. A span recorder is used + * to assert that the request really was registered while the operation ran, so that the cleanup assertion + * cannot pass just because nothing was ever registered. + */ + @Test(groups = {"integration"}, dataProvider = "registryCleanupProvider") + public void testTransportRequestUnregisteredWhenOperationEnds(String operation, boolean succeeds) + throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + if (succeeds) { + stubRetryableThenSuccess(mockServer); + } else { + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withHeader("X-ClickHouse-Exception-Code", String.valueOf(RETRYABLE_CODE)) + .withBody(RETRYABLE_BODY)).build()); + } + + String queryId = "registry-cleanup-" + UUID.randomUUID(); + AtomicReference clientRef = new AtomicReference<>(); + AtomicBoolean registeredWhileRunning = new AtomicBoolean(); + // A request span is started for every attempt, right after the transport request was registered. + SpanRecorder registryProbe = new DefaultSpanRecorder() { + @Override + public Span startRequestSpan(Span operationSpan, String host, int port) { + if (ongoingRequests(clientRef.get()).containsKey(queryId)) { + registeredWhileRunning.set(true); + } + return super.startRequestSpan(operationSpan, host, port); + } + }; + + try (Client client = mockServerClientWithRecorder(mockServer, 1, registryProbe)) { + clientRef.set(client); + try { + runOperationWithQueryId(client, operation, queryId); + Assert.assertTrue(succeeds, "[" + operation + "] expected the operation to fail"); + } catch (Exception e) { + Assert.assertFalse(succeeds, "[" + operation + "] operation should have recovered after a retry"); + Assert.assertEquals(serverErrorCode(e), RETRYABLE_CODE, + "[" + operation + "] the operation must fail with the retryable server error, but failed with " + + e); + } + + Assert.assertTrue(registeredWhileRunning.get(), + "[" + operation + "] the transport request must be registered while the operation runs " + + "(otherwise the cleanup assertion below is vacuous)"); + Assert.assertFalse(ongoingRequests(client).containsKey(queryId), + "[" + operation + "] the transport request must be unregistered once the operation ended"); + } finally { + mockServer.stop(); + } + } + + /** + * Resolves the ClickHouse error code of a failed operation, unwrapping the {@code ExecutionException} the + * operation future may have wrapped it in. + */ + private static int serverErrorCode(Throwable t) { + for (Throwable cause = t; cause != null; cause = cause.getCause()) { + if (cause instanceof ServerException) { + return ((ServerException) cause).getCode(); + } + } + throw new AssertionError("expected a ServerException in the cause chain", t); + } + + @DataProvider(name = "registryCleanupProvider") + public static Object[][] registryCleanupProvider() { + return new Object[][]{ + {"query", true}, + {"query", false}, + {"insert-stream", true}, + {"insert-stream", false}, + {"insert-pojo", true}, + {"insert-pojo", false} + }; + } + + /** + * Builds a client against the mocked server that reports its spans to the given recorder - used to observe + * the client's own state at a known point of an operation. + */ + private Client mockServerClientWithRecorder(WireMockServer mockServer, int maxRetries, SpanRecorder recorder) { + return new Client.Builder() + .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressClientRequest(false) + .compressServerResponse(false) + .setMaxRetries(maxRetries) + .setSpanRecorder(recorder) + .build(); + } + + /** + * Stubs a retryable server error on the first attempt followed by a successful response, so an operation + * with at least one retry configured goes through the retry branch of the operation loop and then succeeds. + */ + private static void stubRetryableThenSuccess(WireMockServer mockServer) { + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("Retry") + .whenScenarioStateIs(STARTED) + .willSetStateTo("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withHeader("X-ClickHouse-Exception-Code", String.valueOf(RETRYABLE_CODE)) + .withBody(RETRYABLE_BODY)).build()); + + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("Retry") + .whenScenarioStateIs("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Summary", + "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); + } + + /** + * Runs the named operation with an explicit {@code queryId} - required for the transport request to be + * registered at all - and waits for it to complete. + */ + private static void runOperationWithQueryId(Client client, String operation, String queryId) throws Exception { + switch (operation) { + case "query": + try (QueryResponse response = client.query("SELECT timezone()", + new QuerySettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + return; + } + case "insert-stream": + try (InsertResponse response = client.insert("table01", + new ByteArrayInputStream("1\t2\t3\n".getBytes(StandardCharsets.US_ASCII)), + ClickHouseFormat.TSV, new InsertSettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + return; + } + case "insert-pojo": + client.register(InsertablePojo.class, new TableSchema("table01", null, "default", + Collections.singletonList(ClickHouseColumn.of("id", "Int32")))); + try (InsertResponse response = client.insert("table01", + Collections.singletonList(new InsertablePojo(1)), + new InsertSettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + return; + } + default: + throw new IllegalArgumentException("unknown operation: " + operation); + } + } + + /** + * Reads the client's registry of in-flight transport requests - the map {@link + * Client#cancelTransportRequest(String)} resolves a query id against. It has no public accessor, so the + * registry is read reflectively. + */ + @SuppressWarnings("unchecked") + private static Map ongoingRequests(Client client) { + try { + Field field = Client.class.getDeclaredField("ongoingRequests"); + field.setAccessible(true); + return (Map) field.get(client); + } catch (ReflectiveOperationException e) { + throw new AssertionError("failed to read the client's registry of ongoing requests", e); + } + } + /** * Reissues {@link Client#cancelTransportRequest(String)} until the worker stops or the timeout elapses. * Retrying makes the test robust against thread-scheduling races where a single cancel could land before diff --git a/client-v2/src/test/java/com/clickhouse/client/observability/SpanRecorderTest.java b/client-v2/src/test/java/com/clickhouse/client/observability/SpanRecorderTest.java new file mode 100644 index 000000000..19d0ad10e --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/observability/SpanRecorderTest.java @@ -0,0 +1,164 @@ +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.CapturingSpanRecorder; +import com.clickhouse.client.api.observability.CapturingSpanRecorder.CapturedSpan; +import com.clickhouse.client.api.observability.SpanAttribute; +import com.clickhouse.client.api.query.QueryResponse; +import com.clickhouse.client.api.query.QuerySettings; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.util.List; +import java.util.concurrent.ExecutionException; + +public class SpanRecorderTest extends BaseIntegrationTest { + + private static final String TABLE = "span_recorder_test_table"; + + private CapturingSpanRecorder recorder; + private Client client; + private String database; + + @BeforeMethod(groups = {"integration"}) + void setUp() throws Exception { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + database = ClickHouseServerForTest.getDatabase(); + recorder = new CapturingSpanRecorder(); + client = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isCloud()) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(database) + .setSpanRecorder(recorder) + .build(); + client.execute("DROP TABLE IF EXISTS " + TABLE).get(); + client.execute("CREATE TABLE " + TABLE + " (id Int32, name String) ENGINE = MergeTree ORDER BY id").get(); + client.execute("INSERT INTO " + TABLE + " VALUES (1, 'a'), (2, 'b'), (3, 'c')").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 testQuerySpanReportsReturnedRowsAndServerQueryId() throws Exception { + QuerySettings settings = new QuerySettings().waitEndOfQuery(true); + try (QueryResponse response = client.query("SELECT id FROM " + TABLE + " ORDER BY id", settings).get()) { + Assert.assertNotNull(response); + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "query " + database); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_SYSTEM_NAME), "clickhouse"); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_NAMESPACE), database); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_RESPONSE_RETURNED_ROWS), 3L); + Assert.assertNotNull(operationSpan.getAttribute(SpanAttribute.CLICKHOUSE_QUERY_ID), + "the server assigns a query id when the client did not"); + Assert.assertNull(operationSpan.getErrorType()); + Assert.assertEquals(operationSpan.getEndCount(), 1); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), 1); + Assert.assertEquals(requestSpans.get(0).getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE), 200); + } + + @Test(groups = {"integration"}) + public void testServerErrorRecordsErrorTypeAndStatusCode() { + 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); + } + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getErrorType(), ServerException.class.getName()); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_RESPONSE_STATUS_CODE), + ServerException.TABLE_NOT_FOUND); + Assert.assertEquals(operationSpan.getEndCount(), 1); + + List requestSpans = recorder.requestSpans(operationSpan); + Assert.assertEquals(requestSpans.size(), 1); + CapturedSpan requestSpan = requestSpans.get(0); + Assert.assertEquals(requestSpan.getErrorType(), ServerException.class.getName()); + Assert.assertEquals(requestSpan.getAttribute(SpanAttribute.DB_RESPONSE_STATUS_CODE), + ServerException.TABLE_NOT_FOUND); + Object httpStatus = requestSpan.getAttribute(SpanAttribute.HTTP_RESPONSE_STATUS_CODE); + Assert.assertNotNull(httpStatus, "the HTTP status of the error response must be recorded"); + Assert.assertTrue((Integer) httpStatus >= 400, "Unexpected HTTP status: " + httpStatus); + Assert.assertEquals(requestSpan.getEndCount(), 1); + } + + @Test(groups = {"integration"}) + public void testPingSpan() { + Assert.assertTrue(client.ping()); + + // a ping runs a query, so it is reported as one - the client does not name the operations it + // implements on top of a query + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "query " + database); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), "SELECT 1 FORMAT TabSeparated"); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_NAME)); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_COLLECTION_NAME)); + Assert.assertEquals(operationSpan.getEndCount(), 1); + Assert.assertEquals(recorder.requestSpans(operationSpan).size(), 1); + } + + @Test(groups = {"integration"}) + public void testCommandSpanReportsStatement() throws Exception { + client.execute("TRUNCATE TABLE " + TABLE).get(); + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "query " + database); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), "TRUNCATE TABLE " + TABLE); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_NAME)); + Assert.assertNull(operationSpan.getErrorType()); + Assert.assertEquals(operationSpan.getEndCount(), 1); + } + + @Test(groups = {"integration"}) + public void testTableSchemaSpanIsReportedAsQuery() { + TableSchema schema = client.getTableSchema(TABLE); + Assert.assertEquals(schema.getColumns().size(), 2); + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "query " + database); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), + "DESCRIBE TABLE " + TABLE + " FORMAT TSKV"); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_NAME)); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_COLLECTION_NAME)); + Assert.assertEquals(operationSpan.getEndCount(), 1); + } + + @Test(groups = {"integration"}) + public void testTableSchemaFromQuerySpanIsReportedAsQuery() { + TableSchema schema = client.getTableSchemaFromQuery("SELECT id FROM " + TABLE); + Assert.assertEquals(schema.getColumns().size(), 1); + + CapturedSpan operationSpan = recorder.operationSpan(); + Assert.assertEquals(operationSpan.getName(), "query " + database); + Assert.assertEquals(operationSpan.getAttribute(SpanAttribute.DB_QUERY_TEXT), + "DESC (SELECT id FROM " + TABLE + ") FORMAT TSKV"); + Assert.assertNull(operationSpan.getAttribute(SpanAttribute.DB_OPERATION_NAME)); + Assert.assertEquals(operationSpan.getEndCount(), 1); + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/observability/SpanSupportUsageTest.java b/client-v2/src/test/java/com/clickhouse/client/observability/SpanSupportUsageTest.java new file mode 100644 index 000000000..818dca1ee --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/observability/SpanSupportUsageTest.java @@ -0,0 +1,177 @@ +package com.clickhouse.client.observability; + +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.observability.CapturingSpanRecorder; +import com.clickhouse.client.api.observability.CapturingSpanRecorder.CapturedSpan; +import com.clickhouse.client.api.observability.DefaultSpanRecorder; +import com.clickhouse.client.api.observability.Span; +import com.clickhouse.client.api.observability.SpanAttribute; +import com.clickhouse.client.api.observability.SpanRecorder; +import com.clickhouse.client.api.observability.SpanSupport; +import com.clickhouse.client.api.query.QuerySettings; +import com.clickhouse.client.api.transport.Endpoint; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Asserts the layering of the observability SPI as seen from outside its own package - the way a + * recorder shipped separately uses it: the recorder owns the spans and calls {@link SpanSupport} when + * it wants the client's standard names and attributes, and it is free not to call it at all. + */ +public class SpanSupportUsageTest { + + @Test + public void testSupportFillsStandardValuesOnARecordersOwnSpan() { + // the shape of a recorder implementation: it creates the span, then asks the support to fill + // the standard name and attributes from the structures the client handed it + SpanSupport support = SpanSupport.DEFAULT; + QuerySettings settings = new QuerySettings().setDatabase("db1").setQueryId("q1"); + RecordingSpan span = new RecordingSpan(); + + Assert.assertEquals(support.querySpanName(settings), "query db1"); + support.fillQueryAttributes(span, settings, "SELECT 1", null); + support.recordFailure(span, new IllegalStateException("boom")); + + Assert.assertEquals(span.attributes.get(SpanAttribute.DB_SYSTEM_NAME.getKey()), "clickhouse"); + Assert.assertEquals(span.attributes.get(SpanAttribute.DB_NAMESPACE.getKey()), "db1"); + Assert.assertEquals(span.attributes.get(SpanAttribute.DB_QUERY_TEXT.getKey()), "SELECT 1"); + Assert.assertEquals(span.attributes.get(SpanAttribute.CLICKHOUSE_QUERY_ID.getKey()), "q1"); + Assert.assertEquals(span.errorType, IllegalStateException.class.getName()); + } + + @Test + public void testSupportFillsInsertAndRequestValues() { + SpanSupport support = SpanSupport.DEFAULT; + InsertSettings settings = new InsertSettings().setDatabase("db1").setQueryId("q1"); + RecordingSpan operationSpan = new RecordingSpan(); + RecordingSpan requestSpan = new RecordingSpan(); + + Assert.assertEquals(support.insertSpanName(settings, "t1"), "insert db1.t1"); + support.fillInsertAttributes(operationSpan, settings, "t1", 3, null); + Assert.assertEquals(support.requestSpanName(), "POST"); + support.fillRequestAttributes(requestSpan, "localhost", 8123); + support.recordHttpStatus(requestSpan, 200); + + Assert.assertEquals(operationSpan.attributes.get(SpanAttribute.DB_COLLECTION_NAME.getKey()), "t1"); + Assert.assertEquals(operationSpan.attributes.get(SpanAttribute.DB_OPERATION_NAME.getKey()), "insert"); + Assert.assertEquals(operationSpan.attributes.get(SpanAttribute.DB_OPERATION_BATCH_SIZE.getKey()), 3); + Assert.assertEquals(requestSpan.attributes.get(SpanAttribute.HTTP_REQUEST_METHOD.getKey()), "POST"); + Assert.assertEquals(requestSpan.attributes.get(SpanAttribute.SERVER_ADDRESS.getKey()), "localhost"); + Assert.assertEquals(requestSpan.attributes.get(SpanAttribute.SERVER_PORT.getKey()), 8123); + Assert.assertEquals(requestSpan.attributes.get(SpanAttribute.HTTP_RESPONSE_STATUS_CODE.getKey()), 200); + } + + @Test + public void testUnknownBatchSizeIsNotReported() { + SpanSupport support = SpanSupport.DEFAULT; + RecordingSpan span = new RecordingSpan(); + + support.fillInsertAttributes(span, new InsertSettings().setDatabase("db1"), "t1", + SpanRecorder.BATCH_SIZE_UNKNOWN, null); + + Assert.assertFalse(span.attributes.containsKey(SpanAttribute.DB_OPERATION_BATCH_SIZE.getKey()), + "a batch size the client does not know must not be reported"); + } + + @Test + public void testSupportMayBeExtendedToChangeSpanNames() { + // a recorder that wants other values overrides the method that computes them and hands its own + // support to the client's values + SpanSupport support = new SpanSupport() { + @Override + protected String spanName(String operationName, String namespace, String collectionName) { + return "custom:" + super.spanName(operationName, namespace, collectionName); + } + }; + + Assert.assertEquals(support.querySpanName(new QuerySettings().setDatabase("db1")), "custom:query db1"); + } + + @Test + public void testRecorderIsCalledFirstAndMayIgnoreTheSupport() { + // the client calls the recorder, not the support, so a recorder that reports its own values + // never goes through SpanSupport at all + OwnValuesSpanRecorder recorder = new OwnValuesSpanRecorder(); + + Span span = recorder.startQuerySpan(new QuerySettings().setDatabase("db1"), "SELECT 1", null); + span.end(); + + Assert.assertEquals(((RecordingSpan) span).attributes.size(), 1, + "a recorder that ignores the support reports only what it set itself"); + Assert.assertEquals(((RecordingSpan) span).attributes.get("my.statement"), "SELECT 1"); + Assert.assertFalse(recorder.usedSupport, "using SpanSupport must stay opt-in"); + } + + @Test + public void testRecorderThatOptsInGetsTheStandardValues() { + CapturingSpanRecorder recorder = new CapturingSpanRecorder(); + + Span span = recorder.startQuerySpan(new QuerySettings().setDatabase("db1").setQueryId("q1"), + "SELECT 1", null); + span.end(); + + CapturedSpan captured = recorder.operationSpan(); + Assert.assertSame(captured, span); + Assert.assertEquals(captured.getName(), "query db1"); + Assert.assertEquals(captured.getAttribute(SpanAttribute.DB_SYSTEM_NAME), "clickhouse"); + Assert.assertEquals(captured.getAttribute(SpanAttribute.DB_QUERY_TEXT), "SELECT 1"); + Assert.assertEquals(captured.getAttribute(SpanAttribute.CLICKHOUSE_QUERY_ID), "q1"); + Assert.assertEquals(captured.getEndCount(), 1); + } + + @Test + public void testDefaultRecorderRecordsNothing() { + Assert.assertSame(DefaultSpanRecorder.NOOP.startQuerySpan(new QuerySettings().setDatabase("db1"), + "SELECT 1", null), DefaultSpanRecorder.NOOP_SPAN); + Assert.assertSame(DefaultSpanRecorder.NOOP.startRequestSpan(DefaultSpanRecorder.NOOP_SPAN, + "localhost", 8123), DefaultSpanRecorder.NOOP_SPAN); + } + + /** + * Recorder that reports its own values and never asks {@link SpanSupport} for the standard ones. + */ + private static final class OwnValuesSpanRecorder extends DefaultSpanRecorder { + + volatile boolean usedSupport; + + @Override + protected SpanSupport getSpanSupport() { + usedSupport = true; + return super.getSpanSupport(); + } + + @Override + public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) { + RecordingSpan span = new RecordingSpan(); + span.setAttribute("my.statement", sqlQuery); + return span; + } + } + + /** + * Span outside the client's packages, so the test uses the SPI the way an external recorder does. + */ + private static final class RecordingSpan implements Span { + + final Map attributes = new LinkedHashMap<>(); + String errorType; + + @Override + public void setAttribute(String key, Object value) { + attributes.put(key, value); + } + + @Override + public void setError(String errorType) { + this.errorType = errorType; + } + + @Override + public void end() { + // nothing to do + } + } +} diff --git a/docs/features.md b/docs/features.md index 7ba874072..ff87a167f 100644 --- a/docs/features.md +++ b/docs/features.md @@ -35,6 +35,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Retry behavior: Can retry failed operations for configured failure causes and retry limits. - 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), `db.response.returned_rows` on success when the server reported a progress summary (for example with `QuerySettings#waitEndOfQuery(true)`), and `error.type` plus `db.response.status_code` on failure. 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. - 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.