= 0) {
- span.setAttribute(SpanAttribute.DB_RESPONSE_RETURNED_ROWS.getKey(), returnedRows.getLong());
+ }
+
+ /**
+ * Records one server metric of a completed operation. The value comes from the server's progress
+ * summary, which is not always available, so a metric the server did not report is left out.
+ *
+ * @param span - span of the operation
+ * @param metrics - metrics of the completed operation
+ * @param metric - server metric to read
+ * @param attribute - attribute to record it under
+ */
+ protected void recordServerMetric(Span span, OperationMetrics metrics, ServerMetrics metric,
+ SpanAttribute attribute) {
+ Metric value = metrics.getMetric(metric);
+ if (value != null && value.getLong() >= 0) {
+ span.setAttribute(attribute.getKey(), value.getLong());
}
}
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/observability/otel/OpenTelemetrySpanRecorder.java b/client-v2/src/main/java/com/clickhouse/client/api/observability/otel/OpenTelemetrySpanRecorder.java
new file mode 100644
index 000000000..8dbeb63c7
--- /dev/null
+++ b/client-v2/src/main/java/com/clickhouse/client/api/observability/otel/OpenTelemetrySpanRecorder.java
@@ -0,0 +1,271 @@
+package com.clickhouse.client.api.observability.otel;
+
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.metrics.OperationMetrics;
+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 io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * {@link SpanRecorder} that reports client operations and transport requests as OpenTelemetry spans.
+ *
+ * It is registered like any other recorder:
+ *
{@code
+ * Client client = new Client.Builder()
+ * .addEndpoint("http://localhost:8123")
+ * .setSpanRecorder(new OpenTelemetrySpanRecorder(openTelemetry))
+ * .build();
+ * }
+ * Every span is a {@link SpanKind#CLIENT} span and carries the client's standard name and
+ * attributes, which are derived by {@link SpanSupport} - so the recorded keys are the ones listed in
+ * {@link SpanAttribute} and mean the same as for every other recorder.
+ *
+ * An operation span is started as a child of the {@linkplain Context#current() current context}, so
+ * it appears under the application's own span when the operation is started on a thread that has
+ * one. A request span is a child of the operation span it was started for. The recorder does not
+ * make any span current: the client hands the response to the caller before the response body is
+ * read, so a span is ended on a thread the recorder does not control.
+ *
+ * Instances are thread-safe and can be shared by several clients.
+ */
+public class OpenTelemetrySpanRecorder implements SpanRecorder {
+
+ /**
+ * Default instrumentation scope name. It is reported for the spans of a recorder created by the
+ * no-argument constructor or by {@link #OpenTelemetrySpanRecorder(OpenTelemetry)}. A recorder
+ * created by {@link #OpenTelemetrySpanRecorder(Tracer)} reports the scope of the given tracer
+ * instead.
+ */
+ public static final String INSTRUMENTATION_SCOPE_NAME = "com.clickhouse.client";
+
+ private final SpanSupport spanSupport = SpanSupport.DEFAULT;
+
+ /**
+ * Tracer the spans are created with, or {@code null} when they are created with the tracer of the
+ * global OpenTelemetry instance, which is then read every time a span is started.
+ */
+ private final Tracer tracer;
+
+ /**
+ * Creates a recorder that reports to the {@linkplain GlobalOpenTelemetry#get() global}
+ * OpenTelemetry instance. Use it when the application configures OpenTelemetry globally, for
+ * example through the OpenTelemetry Java agent or the autoconfigure SDK extension.
+ *
+ * The global instance is read when a span is started, not here, so a client may be created before
+ * the application installs its OpenTelemetry SDK.
+ */
+ public OpenTelemetrySpanRecorder() {
+ this.tracer = null;
+ }
+
+ /**
+ * Creates a recorder that reports to the given OpenTelemetry instance.
+ *
+ * @param openTelemetry - OpenTelemetry instance to report to; must not be {@code null}
+ */
+ public OpenTelemetrySpanRecorder(OpenTelemetry openTelemetry) {
+ this(tracerOf(openTelemetry));
+ }
+
+ /**
+ * Creates a recorder that reports to the given tracer. Use it to report the client's spans under
+ * an instrumentation scope of the application's choice.
+ *
+ * @param tracer - tracer to create spans with; must not be {@code null}
+ */
+ public OpenTelemetrySpanRecorder(Tracer tracer) {
+ if (tracer == null) {
+ throw new IllegalArgumentException("tracer must not be null");
+ }
+ this.tracer = tracer;
+ }
+
+ private static Tracer tracerOf(OpenTelemetry openTelemetry) {
+ if (openTelemetry == null) {
+ throw new IllegalArgumentException("openTelemetry must not be null");
+ }
+ return openTelemetry.getTracer(INSTRUMENTATION_SCOPE_NAME);
+ }
+
+ /**
+ * Returns the tracer the next span is created with - the one given to this recorder, or the tracer
+ * of the global OpenTelemetry instance as it is installed now.
+ *
+ * @return tracer; never {@code null}
+ */
+ protected Tracer getTracer() {
+ return tracer != null ? tracer : GlobalOpenTelemetry.get().getTracer(INSTRUMENTATION_SCOPE_NAME);
+ }
+
+ @Override
+ public Span startQuerySpan(QuerySettings settings, String sqlQuery, Endpoint endpoint) {
+ OpenTelemetrySpan span = startSpan(spanSupport.querySpanName(settings), Context.current());
+ spanSupport.fillQueryAttributes(span, settings, sqlQuery, endpoint);
+ return span;
+ }
+
+ @Override
+ public Span startInsertSpan(InsertSettings settings, String tableName, int batchSize, Endpoint endpoint) {
+ OpenTelemetrySpan span = startSpan(spanSupport.insertSpanName(settings, tableName), Context.current());
+ spanSupport.fillInsertAttributes(span, settings, tableName, batchSize, endpoint);
+ return span;
+ }
+
+ @Override
+ public Span startRequestSpan(Span operationSpan, String host, int port) {
+ OpenTelemetrySpan span = startSpan(spanSupport.requestSpanName(), parentContextOf(operationSpan));
+ spanSupport.fillRequestAttributes(span, host, port);
+ return span;
+ }
+
+ @Override
+ public void recordHttpStatus(Span requestSpan, int statusCode) {
+ spanSupport.recordHttpStatus(requestSpan, statusCode);
+ }
+
+ @Override
+ public void recordQuerySuccess(Span operationSpan, OperationMetrics metrics) {
+ spanSupport.recordQuerySuccess(operationSpan, metrics);
+ }
+
+ @Override
+ public void recordInsertSuccess(Span operationSpan, OperationMetrics metrics) {
+ spanSupport.recordInsertSuccess(operationSpan, metrics);
+ }
+
+ @Override
+ public void recordFailure(Span operationSpan, Throwable t) {
+ spanSupport.recordFailure(operationSpan, t);
+ recordException(operationSpan, t);
+ }
+
+ @Override
+ public void recordRequestFailure(Span requestSpan, Throwable t) {
+ spanSupport.recordRequestFailure(requestSpan, t);
+ recordException(requestSpan, t);
+ }
+
+ /**
+ * Records the failure itself as an OpenTelemetry exception event, so that its message and stack
+ * trace are reported next to the {@link SpanAttribute#ERROR_TYPE} attribute.
+ *
+ * @param span - span the failure was reported on
+ * @param t - failure, may be {@code null}
+ */
+ protected void recordException(Span span, Throwable t) {
+ if (t != null && span instanceof OpenTelemetrySpan) {
+ ((OpenTelemetrySpan) span).getSpan().recordException(t);
+ }
+ }
+
+ /**
+ * Starts a client span with the given name under the given parent context.
+ *
+ * @param spanName - name of the span
+ * @param parentContext - context the span is started under
+ * @return new span
+ */
+ protected OpenTelemetrySpan startSpan(String spanName, Context parentContext) {
+ io.opentelemetry.api.trace.Span span = getTracer().spanBuilder(spanName)
+ .setSpanKind(SpanKind.CLIENT)
+ .setParent(parentContext)
+ .startSpan();
+ return new OpenTelemetrySpan(span, parentContext.with(span));
+ }
+
+ /**
+ * Returns the context a request span is started under - the context of its operation span, or the
+ * current context when the operation span was not created by this recorder.
+ */
+ private static Context parentContextOf(Span operationSpan) {
+ return operationSpan instanceof OpenTelemetrySpan
+ ? ((OpenTelemetrySpan) operationSpan).getContext()
+ : Context.current();
+ }
+
+ /**
+ * {@link Span} backed by an OpenTelemetry span.
+ */
+ public static class OpenTelemetrySpan implements Span {
+
+ private final io.opentelemetry.api.trace.Span span;
+
+ private final Context context;
+
+ private final AtomicBoolean ended = new AtomicBoolean();
+
+ OpenTelemetrySpan(io.opentelemetry.api.trace.Span span, Context context) {
+ this.span = span;
+ this.context = context;
+ }
+
+ /**
+ * Returns the OpenTelemetry span this span records on.
+ *
+ * @return OpenTelemetry span
+ */
+ public io.opentelemetry.api.trace.Span getSpan() {
+ return span;
+ }
+
+ /**
+ * Returns the context that holds this span. It is the parent context of the spans started for
+ * the same operation.
+ *
+ * @return context holding this span
+ */
+ public Context getContext() {
+ return context;
+ }
+
+ @Override
+ public void setAttribute(String key, Object value) {
+ if (key == null || value == null) {
+ return;
+ }
+ if (value instanceof String) {
+ span.setAttribute(AttributeKey.stringKey(key), (String) value);
+ } else if (value instanceof Boolean) {
+ span.setAttribute(AttributeKey.booleanKey(key), (Boolean) value);
+ } else if (value instanceof Double || value instanceof Float) {
+ span.setAttribute(AttributeKey.doubleKey(key), ((Number) value).doubleValue());
+ } else if (value instanceof Number) {
+ span.setAttribute(AttributeKey.longKey(key), ((Number) value).longValue());
+ } else {
+ span.setAttribute(AttributeKey.stringKey(key), String.valueOf(value));
+ }
+ }
+
+ @Override
+ public void setError(String errorType) {
+ span.setStatus(StatusCode.ERROR);
+ if (errorType != null) {
+ span.setAttribute(AttributeKey.stringKey(SpanAttribute.ERROR_TYPE.getKey()), errorType);
+ }
+ }
+
+ @Override
+ public void end() {
+ if (ended.compareAndSet(false, true)) {
+ span.end();
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "OpenTelemetrySpan[" + span.getSpanContext().getSpanId() + "]";
+ }
+ }
+}
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
index 1acf24a91..cd2738dfe 100644
--- 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
@@ -53,8 +53,13 @@ public void recordHttpStatus(Span requestSpan, int statusCode) {
}
@Override
- public void recordSuccess(Span operationSpan, OperationMetrics metrics) {
- getSpanSupport().recordSuccess(operationSpan, metrics);
+ public void recordQuerySuccess(Span operationSpan, OperationMetrics metrics) {
+ getSpanSupport().recordQuerySuccess(operationSpan, metrics);
+ }
+
+ @Override
+ public void recordInsertSuccess(Span operationSpan, OperationMetrics metrics) {
+ getSpanSupport().recordInsertSuccess(operationSpan, metrics);
}
@Override
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
index d1763b92b..007d0cb6f 100644
--- 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
@@ -359,7 +359,8 @@ public void testDefaultSpanRecorderRecordsNothing() {
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.recordQuerySuccess(noopSpan, null);
+ defaultRecorder.recordInsertSuccess(noopSpan, null);
defaultRecorder.recordFailure(noopSpan, new IllegalStateException("boom"));
defaultRecorder.recordRequestFailure(noopSpan, new IllegalStateException("boom"));
noopSpan.setAttribute(SpanAttribute.DB_NAMESPACE.getKey(), "db");
@@ -446,7 +447,12 @@ public void recordHttpStatus(Span requestSpan, int statusCode) {
}
@Override
- public void recordSuccess(Span operationSpan, OperationMetrics metrics) {
+ public void recordQuerySuccess(Span operationSpan, OperationMetrics metrics) {
+ // records nothing
+ }
+
+ @Override
+ public void recordInsertSuccess(Span operationSpan, OperationMetrics metrics) {
// records nothing
}
diff --git a/client-v2/src/test/java/com/clickhouse/client/api/observability/otel/OpenTelemetrySpanRecorderUnitTest.java b/client-v2/src/test/java/com/clickhouse/client/api/observability/otel/OpenTelemetrySpanRecorderUnitTest.java
new file mode 100644
index 000000000..e42b3fccc
--- /dev/null
+++ b/client-v2/src/test/java/com/clickhouse/client/api/observability/otel/OpenTelemetrySpanRecorderUnitTest.java
@@ -0,0 +1,533 @@
+package com.clickhouse.client.api.observability.otel;
+
+import com.clickhouse.client.api.ServerException;
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.internal.ClientStatisticsHolder;
+import com.clickhouse.client.api.metrics.OperationMetrics;
+import com.clickhouse.client.api.metrics.OperationType;
+import com.clickhouse.client.api.metrics.ServerMetrics;
+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.query.QuerySettings;
+import com.clickhouse.client.api.transport.Endpoint;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.AttributeType;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.EventData;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+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.net.URI;
+import java.util.List;
+
+public class OpenTelemetrySpanRecorderUnitTest {
+
+ private static final String DATABASE = "spans_db";
+
+ private InMemorySpanExporter exporter;
+ private OpenTelemetrySdk openTelemetry;
+ private OpenTelemetrySpanRecorder recorder;
+
+ @BeforeMethod
+ void setUp() {
+ exporter = InMemorySpanExporter.create();
+ openTelemetry = OpenTelemetrySdk.builder()
+ .setTracerProvider(SdkTracerProvider.builder()
+ .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+ .build())
+ .build();
+ recorder = new OpenTelemetrySpanRecorder(openTelemetry);
+ }
+
+ @AfterMethod
+ void tearDown() {
+ openTelemetry.close();
+ }
+
+ @Test
+ public void testQuerySpanReportsStandardNameAndAttributes() {
+ Span span = recorder.startQuerySpan(querySettings("q-42"), "SELECT 1", endpoint("ch-host", 8123));
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(exported.getName(), "query " + DATABASE);
+ Assert.assertEquals(exported.getKind(), SpanKind.CLIENT);
+ Assert.assertEquals(exported.getInstrumentationScopeInfo().getName(),
+ OpenTelemetrySpanRecorder.INSTRUMENTATION_SCOPE_NAME);
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.DB_SYSTEM_NAME), "clickhouse");
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.DB_NAMESPACE), DATABASE);
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.DB_QUERY_TEXT), "SELECT 1");
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.CLICKHOUSE_QUERY_ID), "q-42");
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.SERVER_ADDRESS), "ch-host");
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.SERVER_PORT), Long.valueOf(8123L));
+ Assert.assertEquals(exported.getStatus().getStatusCode(), StatusCode.UNSET);
+ }
+
+ @Test
+ public void testInsertSpanReportsCollectionAndBatchSize() {
+ Span span = recorder.startInsertSpan(insertSettings("i-1"), "events", 5, endpoint("ch-host", 8123));
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(exported.getName(), "insert " + DATABASE + ".events");
+ Assert.assertEquals(exported.getKind(), SpanKind.CLIENT);
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.DB_OPERATION_NAME), "insert");
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.DB_COLLECTION_NAME), "events");
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.DB_OPERATION_BATCH_SIZE), Long.valueOf(5L));
+ Assert.assertNull(stringAttribute(exported, SpanAttribute.DB_QUERY_TEXT),
+ "an insert sends no user statement");
+ }
+
+ @Test
+ public void testInsertSpanOmitsUnknownBatchSize() {
+ Span span = recorder.startInsertSpan(insertSettings("i-2"), "events", SpanRecorder.BATCH_SIZE_UNKNOWN,
+ endpoint("ch-host", 8123));
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(exported.getName(), "insert " + DATABASE + ".events");
+ Assert.assertNull(longAttribute(exported, SpanAttribute.DB_OPERATION_BATCH_SIZE));
+ }
+
+ @Test
+ public void testRequestSpanIsChildOfOperationSpan() {
+ Span operationSpan = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", endpoint("ch-host", 8123));
+ Span requestSpan = recorder.startRequestSpan(operationSpan, "node-2", 8443);
+ recorder.recordHttpStatus(requestSpan, 200);
+ requestSpan.end();
+ operationSpan.end();
+
+ SpanData request = spanByName("POST");
+ SpanData operation = spanByName("query " + DATABASE);
+ Assert.assertEquals(request.getTraceId(), operation.getTraceId());
+ Assert.assertEquals(request.getParentSpanId(), operation.getSpanId());
+ Assert.assertEquals(request.getKind(), SpanKind.CLIENT);
+ Assert.assertEquals(stringAttribute(request, SpanAttribute.HTTP_REQUEST_METHOD), "POST");
+ Assert.assertEquals(longAttribute(request, SpanAttribute.HTTP_RESPONSE_STATUS_CODE), Long.valueOf(200L));
+ Assert.assertEquals(stringAttribute(request, SpanAttribute.SERVER_ADDRESS), "node-2",
+ "the attempt reports the endpoint it used");
+ Assert.assertEquals(longAttribute(request, SpanAttribute.SERVER_PORT), Long.valueOf(8443L));
+ }
+
+ @Test
+ public void testOperationSpanJoinsAmbientTrace() {
+ io.opentelemetry.api.trace.Span ambient = openTelemetry.getTracer("test").spanBuilder("application")
+ .startSpan();
+ Span operationSpan;
+ try (Scope scope = ambient.makeCurrent()) {
+ operationSpan = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ }
+ operationSpan.end();
+ ambient.end();
+
+ SpanData operation = spanByName("query " + DATABASE);
+ Assert.assertEquals(operation.getTraceId(), ambient.getSpanContext().getTraceId());
+ Assert.assertEquals(operation.getParentSpanId(), ambient.getSpanContext().getSpanId());
+ }
+
+ @Test
+ public void testRequestSpanFallsBackToCurrentContextWhenOperationSpanIsForeign() {
+ Span requestSpan = recorder.startRequestSpan(DefaultSpanRecorder.NOOP_SPAN, "node-1", 8123);
+ requestSpan.end();
+
+ SpanData request = onlySpan();
+ Assert.assertEquals(request.getName(), "POST");
+ Assert.assertFalse(request.getParentSpanContext().isValid(),
+ "without an operation span and without an ambient context there is no parent to attach to");
+
+ io.opentelemetry.api.trace.Span ambient = openTelemetry.getTracer("test").spanBuilder("application")
+ .startSpan();
+ Span secondRequestSpan;
+ try (Scope scope = ambient.makeCurrent()) {
+ secondRequestSpan = recorder.startRequestSpan(DefaultSpanRecorder.NOOP_SPAN, "node-1", 8123);
+ }
+ secondRequestSpan.end();
+ ambient.end();
+
+ Assert.assertEquals(exporter.getFinishedSpanItems().get(1).getParentSpanId(),
+ ambient.getSpanContext().getSpanId(),
+ "with an ambient context the request span is started under it");
+ }
+
+ @Test
+ public void testEveryAttemptReportsItsOwnRequestSpanUnderOneOperationSpan() {
+ Span operationSpan = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ Span firstAttempt = recorder.startRequestSpan(operationSpan, "node-1", 8123);
+ recorder.recordRequestFailure(firstAttempt, new IllegalStateException("first attempt failed"));
+ firstAttempt.end();
+ Span secondAttempt = recorder.startRequestSpan(operationSpan, "node-2", 8123);
+ recorder.recordHttpStatus(secondAttempt, 200);
+ secondAttempt.end();
+ operationSpan.end();
+
+ List spans = exporter.getFinishedSpanItems();
+ Assert.assertEquals(spans.size(), 3, "Unexpected spans: " + spans);
+ SpanData operation = spans.get(2);
+ Assert.assertEquals(spans.get(0).getParentSpanId(), operation.getSpanId());
+ Assert.assertEquals(spans.get(1).getParentSpanId(), operation.getSpanId());
+ Assert.assertEquals(spans.get(0).getStatus().getStatusCode(), StatusCode.ERROR);
+ Assert.assertEquals(stringAttribute(spans.get(0), SpanAttribute.SERVER_ADDRESS), "node-1");
+ Assert.assertEquals(spans.get(1).getStatus().getStatusCode(), StatusCode.UNSET);
+ Assert.assertEquals(stringAttribute(spans.get(1), SpanAttribute.SERVER_ADDRESS), "node-2");
+ Assert.assertEquals(operation.getStatus().getStatusCode(), StatusCode.UNSET,
+ "a retried operation that succeeded is not failed");
+ }
+
+ @Test
+ public void testQuerySuccessRecordsQueryIdAndWhatTheServerReadAndReturned() {
+ OperationMetrics metrics = queryMetrics();
+ metrics.setQueryId("server-assigned-id");
+ metrics.updateMetric(ServerMetrics.RESULT_ROWS, 7);
+ metrics.updateMetric(ServerMetrics.NUM_ROWS_READ, 4096);
+ metrics.updateMetric(ServerMetrics.NUM_BYTES_READ, 65536);
+
+ Span span = recorder.startQuerySpan(querySettings(null), "SELECT 1", null);
+ recorder.recordQuerySuccess(span, metrics);
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.CLICKHOUSE_QUERY_ID), "server-assigned-id");
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.DB_RESPONSE_RETURNED_ROWS), Long.valueOf(7L));
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS), Long.valueOf(4096L));
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_READ_BYTES), Long.valueOf(65536L));
+ Assert.assertEquals(exported.getStatus().getStatusCode(), StatusCode.UNSET);
+ }
+
+ @Test
+ public void testInsertSuccessRecordsQueryIdAndWhatTheServerWrote() {
+ OperationMetrics metrics = insertMetrics();
+ metrics.setQueryId("server-assigned-id");
+ metrics.updateMetric(ServerMetrics.NUM_ROWS_WRITTEN, 12);
+ metrics.updateMetric(ServerMetrics.NUM_BYTES_WRITTEN, 480);
+
+ Span span = recorder.startInsertSpan(insertSettings(null), "t1", 12, null);
+ recorder.recordInsertSuccess(span, metrics);
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.CLICKHOUSE_QUERY_ID), "server-assigned-id");
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS), Long.valueOf(12L));
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES), Long.valueOf(480L));
+ Assert.assertEquals(exported.getStatus().getStatusCode(), StatusCode.UNSET);
+ }
+
+ @Test
+ public void testEachTrackRecordsOnlyTheMetricsOfItsOwnOperation() {
+ // the server reports the whole summary for both kinds of operation; each track picks the
+ // metrics that are meaningful for it, so a query never claims written rows and the reverse
+ OperationMetrics metrics = queryMetrics();
+ metrics.updateMetric(ServerMetrics.RESULT_ROWS, 7);
+ metrics.updateMetric(ServerMetrics.NUM_ROWS_READ, 4096);
+ metrics.updateMetric(ServerMetrics.NUM_BYTES_READ, 65536);
+ metrics.updateMetric(ServerMetrics.NUM_ROWS_WRITTEN, 12);
+ metrics.updateMetric(ServerMetrics.NUM_BYTES_WRITTEN, 480);
+
+ Span querySpan = recorder.startQuerySpan(querySettings(null), "SELECT 1", null);
+ recorder.recordQuerySuccess(querySpan, metrics);
+ querySpan.end();
+ Span insertSpan = recorder.startInsertSpan(insertSettings(null), "t1", 12, null);
+ recorder.recordInsertSuccess(insertSpan, metrics);
+ insertSpan.end();
+
+ List spans = exporter.getFinishedSpanItems();
+ Assert.assertEquals(spans.size(), 2, "Unexpected spans: " + spans);
+ SpanData query = spans.get(0);
+ SpanData insert = spans.get(1);
+
+ Assert.assertEquals(longAttribute(query, SpanAttribute.DB_RESPONSE_RETURNED_ROWS), Long.valueOf(7L));
+ Assert.assertEquals(longAttribute(query, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS), Long.valueOf(4096L));
+ Assert.assertNull(longAttribute(query, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS));
+ Assert.assertNull(longAttribute(query, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES));
+
+ Assert.assertEquals(longAttribute(insert, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS), Long.valueOf(12L));
+ Assert.assertEquals(longAttribute(insert, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES), Long.valueOf(480L));
+ Assert.assertNull(longAttribute(insert, SpanAttribute.DB_RESPONSE_RETURNED_ROWS));
+ Assert.assertNull(longAttribute(insert, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS));
+ Assert.assertNull(longAttribute(insert, SpanAttribute.CLICKHOUSE_RESPONSE_READ_BYTES));
+ }
+
+ @Test
+ public void testMetricTheServerDidNotReportIsNotRecorded() {
+ // ProcessParser sets every server metric to -1 before it applies the summary, so a metric
+ // missing from the summary must not reach the span as a negative count
+ OperationMetrics metrics = queryMetrics();
+ metrics.updateMetric(ServerMetrics.RESULT_ROWS, -1);
+ metrics.updateMetric(ServerMetrics.NUM_ROWS_READ, -1);
+ metrics.updateMetric(ServerMetrics.NUM_BYTES_READ, 65536);
+
+ Span span = recorder.startQuerySpan(querySettings(null), "SELECT 1", null);
+ recorder.recordQuerySuccess(span, metrics);
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertNull(longAttribute(exported, SpanAttribute.DB_RESPONSE_RETURNED_ROWS));
+ Assert.assertNull(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS));
+ Assert.assertEquals(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_READ_BYTES), Long.valueOf(65536L));
+ }
+
+ @Test
+ public void testSuccessWithoutMetricsRecordsNothing() {
+ Span querySpan = recorder.startQuerySpan(querySettings(null), "SELECT 1", null);
+ recorder.recordQuerySuccess(querySpan, null);
+ querySpan.end();
+ Span insertSpan = recorder.startInsertSpan(insertSettings(null), "t1", 1, null);
+ recorder.recordInsertSuccess(insertSpan, null);
+ insertSpan.end();
+
+ List spans = exporter.getFinishedSpanItems();
+ Assert.assertEquals(spans.size(), 2, "Unexpected spans: " + spans);
+ for (SpanData exported : spans) {
+ Assert.assertNull(stringAttribute(exported, SpanAttribute.CLICKHOUSE_QUERY_ID));
+ Assert.assertNull(longAttribute(exported, SpanAttribute.DB_RESPONSE_RETURNED_ROWS));
+ Assert.assertNull(longAttribute(exported, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS));
+ Assert.assertEquals(exported.getStatus().getStatusCode(), StatusCode.UNSET);
+ }
+ }
+
+ @Test
+ public void testFailureIsRecordedAsExceptionEvent() {
+ Span span = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ recorder.recordFailure(span, new IllegalStateException("boom"));
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(exported.getEvents().size(), 1, "Unexpected events: " + exported.getEvents());
+ EventData event = exported.getEvents().get(0);
+ Assert.assertEquals(event.getAttributes().get(AttributeKey.stringKey("exception.type")),
+ IllegalStateException.class.getName());
+ Assert.assertEquals(event.getAttributes().get(AttributeKey.stringKey("exception.message")), "boom");
+ }
+
+ @Test
+ public void testSpansAreReportedUnderTheGivenTracerScope() {
+ OpenTelemetrySpanRecorder tracerRecorder =
+ new OpenTelemetrySpanRecorder(openTelemetry.getTracer("application-scope", "1.2.3"));
+
+ tracerRecorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null).end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(exported.getInstrumentationScopeInfo().getName(), "application-scope");
+ Assert.assertEquals(exported.getInstrumentationScopeInfo().getVersion(), "1.2.3");
+ Assert.assertEquals(exported.getName(), "query " + DATABASE);
+ }
+
+ @Test
+ public void testGlobalInstanceIsReadWhenSpanStartsNotWhenRecorderIsCreated() {
+ GlobalOpenTelemetry.resetForTest();
+ try {
+ // the recorder is created before the application installs its SDK
+ OpenTelemetrySpanRecorder globalRecorder = new OpenTelemetrySpanRecorder();
+
+ InMemorySpanExporter lateExporter = InMemorySpanExporter.create();
+ OpenTelemetrySdk lateSdk = OpenTelemetrySdk.builder()
+ .setTracerProvider(SdkTracerProvider.builder()
+ .addSpanProcessor(SimpleSpanProcessor.create(lateExporter))
+ .build())
+ .build();
+ GlobalOpenTelemetry.set(lateSdk);
+ try {
+ globalRecorder.startQuerySpan(querySettings("q-late"), "SELECT 1", endpoint("ch-host", 8123)).end();
+
+ List exported = lateExporter.getFinishedSpanItems();
+ Assert.assertEquals(exported.size(), 1,
+ "a span must reach the SDK installed after the recorder was created");
+ Assert.assertEquals(exported.get(0).getName(), "query " + DATABASE);
+ Assert.assertEquals(exported.get(0).getInstrumentationScopeInfo().getName(),
+ OpenTelemetrySpanRecorder.INSTRUMENTATION_SCOPE_NAME);
+ } finally {
+ lateSdk.close();
+ }
+ } finally {
+ GlobalOpenTelemetry.resetForTest();
+ }
+ }
+
+ @Test
+ public void testFailureRecordsErrorStatusAndErrorType() {
+ Span span = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ recorder.recordFailure(span, new IllegalStateException("boom"));
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertEquals(exported.getStatus().getStatusCode(), StatusCode.ERROR);
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.ERROR_TYPE),
+ IllegalStateException.class.getName());
+ Assert.assertNull(longAttribute(exported, SpanAttribute.DB_RESPONSE_STATUS_CODE),
+ "a client-side failure carries no server error code");
+ }
+
+ @Test
+ public void testServerFailureRecordsClickHouseCodeAndHttpStatus() {
+ Span operationSpan = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ Span requestSpan = recorder.startRequestSpan(operationSpan, "node-1", 8123);
+ ServerException serverException = new ServerException(60, "table not found", 404, "q-1");
+ recorder.recordRequestFailure(requestSpan, serverException);
+ recorder.recordFailure(operationSpan, new RuntimeException(serverException));
+ requestSpan.end();
+ operationSpan.end();
+
+ SpanData request = spanByName("POST");
+ Assert.assertEquals(request.getStatus().getStatusCode(), StatusCode.ERROR);
+ Assert.assertEquals(stringAttribute(request, SpanAttribute.ERROR_TYPE), ServerException.class.getName());
+ Assert.assertEquals(longAttribute(request, SpanAttribute.DB_RESPONSE_STATUS_CODE), Long.valueOf(60L));
+ Assert.assertEquals(longAttribute(request, SpanAttribute.HTTP_RESPONSE_STATUS_CODE), Long.valueOf(404L));
+
+ SpanData operation = spanByName("query " + DATABASE);
+ Assert.assertEquals(operation.getStatus().getStatusCode(), StatusCode.ERROR);
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.ERROR_TYPE), ServerException.class.getName());
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.DB_RESPONSE_STATUS_CODE), Long.valueOf(60L));
+ }
+
+ @Test
+ public void testEndIsIdempotent() {
+ Span span = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ span.end();
+ span.end();
+
+ Assert.assertEquals(exporter.getFinishedSpanItems().size(), 1);
+ }
+
+ @DataProvider(name = "attributeValues")
+ public static Object[][] attributeValues() {
+ return new Object[][]{
+ {"text", AttributeType.STRING, "text"},
+ {Boolean.TRUE, AttributeType.BOOLEAN, Boolean.TRUE},
+ {42, AttributeType.LONG, 42L},
+ {42L, AttributeType.LONG, 42L},
+ {(short) 42, AttributeType.LONG, 42L},
+ {1.5d, AttributeType.DOUBLE, 1.5d},
+ {1.5f, AttributeType.DOUBLE, 1.5d},
+ {URI.create("http://localhost:8123"), AttributeType.STRING, "http://localhost:8123"},
+ };
+ }
+
+ @Test(dataProvider = "attributeValues")
+ public void testAttributeValueTyping(Object value, AttributeType expectedType, Object expectedValue) {
+ Span span = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ span.setAttribute("custom.attribute", value);
+ span.end();
+
+ SpanData exported = onlySpan();
+ AttributeKey key = keyOf(exported, "custom.attribute");
+ Assert.assertNotNull(key, "attribute was not recorded");
+ Assert.assertEquals(key.getType(), expectedType);
+ Assert.assertEquals(exported.getAttributes().get(key), expectedValue);
+ }
+
+ @Test
+ public void testNullAttributeKeyOrValueIsIgnored() {
+ Span span = recorder.startQuerySpan(querySettings("q-1"), "SELECT 1", null);
+ span.setAttribute(null, "value");
+ span.setAttribute("custom.attribute", null);
+ span.end();
+
+ SpanData exported = onlySpan();
+ Assert.assertNull(keyOf(exported, "custom.attribute"));
+ Assert.assertEquals(stringAttribute(exported, SpanAttribute.DB_QUERY_TEXT), "SELECT 1",
+ "the other attributes are still recorded");
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testNullOpenTelemetryIsRejected() {
+ new OpenTelemetrySpanRecorder((OpenTelemetry) null);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testNullTracerIsRejected() {
+ new OpenTelemetrySpanRecorder((Tracer) null);
+ }
+
+ @Test
+ public void testMetricsCreatedWithoutAKindReportUnknown() {
+ // the client always reports the kind; metrics created by user code may not know it
+ Assert.assertEquals(new OperationMetrics(new ClientStatisticsHolder()).getOperationType(),
+ OperationType.UNKNOWN);
+ Assert.assertEquals(new OperationMetrics(new ClientStatisticsHolder(), null).getOperationType(),
+ OperationType.UNKNOWN);
+ Assert.assertEquals(queryMetrics().getOperationType(), OperationType.QUERY);
+ Assert.assertEquals(insertMetrics().getOperationType(), OperationType.INSERT);
+ }
+
+ private OperationMetrics queryMetrics() {
+ return new OperationMetrics(new ClientStatisticsHolder(), OperationType.QUERY);
+ }
+
+ private OperationMetrics insertMetrics() {
+ return new OperationMetrics(new ClientStatisticsHolder(), OperationType.INSERT);
+ }
+
+ private QuerySettings querySettings(String queryId) {
+ return new QuerySettings().setDatabase(DATABASE).setQueryId(queryId);
+ }
+
+ private InsertSettings insertSettings(String queryId) {
+ return new InsertSettings().setDatabase(DATABASE).setQueryId(queryId);
+ }
+
+ private static Endpoint endpoint(String host, int port) {
+ return new Endpoint() {
+ @Override
+ public URI getURI() {
+ return URI.create("http://" + host + ":" + port);
+ }
+
+ @Override
+ public String getHost() {
+ return host;
+ }
+
+ @Override
+ public int getPort() {
+ return port;
+ }
+ };
+ }
+
+ private SpanData onlySpan() {
+ List spans = exporter.getFinishedSpanItems();
+ Assert.assertEquals(spans.size(), 1, "Unexpected spans: " + spans);
+ return spans.get(0);
+ }
+
+ private SpanData spanByName(String name) {
+ for (SpanData span : exporter.getFinishedSpanItems()) {
+ if (name.equals(span.getName())) {
+ return span;
+ }
+ }
+ Assert.fail("No span named '" + name + "' in " + exporter.getFinishedSpanItems());
+ return null;
+ }
+
+ private static String stringAttribute(SpanData span, SpanAttribute attribute) {
+ return span.getAttributes().get(AttributeKey.stringKey(attribute.getKey()));
+ }
+
+ private static Long longAttribute(SpanData span, SpanAttribute attribute) {
+ return span.getAttributes().get(AttributeKey.longKey(attribute.getKey()));
+ }
+
+ private static AttributeKey keyOf(SpanData span, String key) {
+ for (AttributeKey candidate : span.getAttributes().asMap().keySet()) {
+ if (candidate.getKey().equals(key)) {
+ return candidate;
+ }
+ }
+ return null;
+ }
+}
diff --git a/client-v2/src/test/java/com/clickhouse/client/observability/otel/OpenTelemetrySpanRecorderTest.java b/client-v2/src/test/java/com/clickhouse/client/observability/otel/OpenTelemetrySpanRecorderTest.java
new file mode 100644
index 000000000..af7c3a08c
--- /dev/null
+++ b/client-v2/src/test/java/com/clickhouse/client/observability/otel/OpenTelemetrySpanRecorderTest.java
@@ -0,0 +1,218 @@
+package com.clickhouse.client.observability.otel;
+
+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.insert.InsertResponse;
+import com.clickhouse.client.api.metrics.OperationType;
+import com.clickhouse.client.api.observability.SpanAttribute;
+import com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder;
+import com.clickhouse.client.api.query.QueryResponse;
+import com.clickhouse.client.api.query.QuerySettings;
+import com.clickhouse.data.ClickHouseFormat;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+public class OpenTelemetrySpanRecorderTest extends BaseIntegrationTest {
+
+ private static final String TABLE = "otel_span_recorder_test_table";
+
+ private InMemorySpanExporter exporter;
+ private OpenTelemetrySdk openTelemetry;
+ private Client client;
+ private String database;
+
+ @BeforeMethod(groups = {"integration"})
+ void setUp() throws Exception {
+ ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
+ database = ClickHouseServerForTest.getDatabase();
+ exporter = InMemorySpanExporter.create();
+ openTelemetry = OpenTelemetrySdk.builder()
+ .setTracerProvider(SdkTracerProvider.builder()
+ .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+ .build())
+ .build();
+ client = new Client.Builder()
+ .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isCloud())
+ .setUsername("default")
+ .setPassword(ClickHouseServerForTest.getPassword())
+ .setDefaultDatabase(database)
+ .setSpanRecorder(new OpenTelemetrySpanRecorder(openTelemetry))
+ .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();
+ exporter.reset();
+ }
+
+ @AfterMethod(groups = {"integration"})
+ void tearDown() throws Exception {
+ if (client != null) {
+ client.execute("DROP TABLE IF EXISTS " + TABLE).get();
+ client.close();
+ }
+ if (openTelemetry != null) {
+ openTelemetry.close();
+ }
+ }
+
+ @Test(groups = {"integration"})
+ public void testQueryExportsOperationSpanWithRequestChild() 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);
+ Assert.assertEquals(response.getMetrics().getOperationType(), OperationType.QUERY);
+ }
+
+ SpanData operation = spanByName("query " + database);
+ Assert.assertEquals(operation.getKind(), SpanKind.CLIENT);
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.DB_SYSTEM_NAME), "clickhouse");
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.DB_NAMESPACE), database);
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.DB_QUERY_TEXT),
+ "SELECT id FROM " + TABLE + " ORDER BY id");
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.DB_RESPONSE_RETURNED_ROWS), Long.valueOf(3L));
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS), Long.valueOf(3L));
+ Assert.assertNotNull(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_READ_BYTES));
+ // the query track does not report what an insert would
+ Assert.assertNull(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS));
+ Assert.assertNull(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES));
+ Assert.assertNotNull(stringAttribute(operation, SpanAttribute.CLICKHOUSE_QUERY_ID));
+ Assert.assertEquals(operation.getStatus().getStatusCode(), StatusCode.UNSET);
+
+ SpanData request = spanByName("POST");
+ Assert.assertEquals(request.getTraceId(), operation.getTraceId());
+ Assert.assertEquals(request.getParentSpanId(), operation.getSpanId());
+ Assert.assertEquals(longAttribute(request, SpanAttribute.HTTP_RESPONSE_STATUS_CODE), Long.valueOf(200L));
+ Assert.assertEquals(stringAttribute(request, SpanAttribute.SERVER_ADDRESS),
+ getServer(ClickHouseProtocol.HTTP).getHost());
+ }
+
+ @Test(groups = {"integration"})
+ public void testFailingQueryExportsErrorStatusAndServerCode() {
+ 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);
+ }
+
+ SpanData operation = spanByName("query " + database);
+ Assert.assertEquals(operation.getStatus().getStatusCode(), StatusCode.ERROR);
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.ERROR_TYPE), ServerException.class.getName());
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.DB_RESPONSE_STATUS_CODE), Long.valueOf(60L));
+
+ SpanData request = spanByName("POST");
+ Assert.assertEquals(request.getStatus().getStatusCode(), StatusCode.ERROR);
+ Assert.assertEquals(stringAttribute(request, SpanAttribute.ERROR_TYPE), ServerException.class.getName());
+ Assert.assertEquals(longAttribute(request, SpanAttribute.HTTP_RESPONSE_STATUS_CODE), Long.valueOf(404L));
+ }
+
+ @Test(groups = {"integration"})
+ public void testInsertExportsSpanWithBatchSize() throws Exception {
+ client.register(SpanRecorderPojo.class, client.getTableSchema(TABLE));
+ exporter.reset();
+
+ SpanRecorderPojo pojo = new SpanRecorderPojo();
+ pojo.setId(4);
+ pojo.setName("d");
+ try (InsertResponse response = client.insert(TABLE, java.util.Collections.singletonList(pojo)).get()) {
+ Assert.assertEquals(response.getMetrics().getOperationType(), OperationType.INSERT);
+ }
+
+ SpanData operation = spanByName("insert " + database + "." + TABLE);
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.DB_OPERATION_NAME), "insert");
+ Assert.assertEquals(stringAttribute(operation, SpanAttribute.DB_COLLECTION_NAME), TABLE);
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.DB_OPERATION_BATCH_SIZE), Long.valueOf(1L));
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS), Long.valueOf(1L));
+ Assert.assertNotNull(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES));
+ // the insert track does not report what a query would
+ Assert.assertNull(longAttribute(operation, SpanAttribute.DB_RESPONSE_RETURNED_ROWS));
+ Assert.assertNull(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_READ_ROWS));
+ Assert.assertEquals(operation.getStatus().getStatusCode(), StatusCode.UNSET);
+ Assert.assertEquals(spanByName("POST").getParentSpanId(), operation.getSpanId());
+ }
+
+ @Test(groups = {"integration"})
+ public void testStreamInsertExportsSpanWithWrittenRows() throws Exception {
+ // the second insert entry point: it does not know the batch size, but it reports the same
+ // insert track as a POJO insert
+ byte[] rows = "4,d\n5,e\n".getBytes(StandardCharsets.UTF_8);
+ try (InsertResponse response = client.insert(TABLE, new ByteArrayInputStream(rows),
+ ClickHouseFormat.CSV).get()) {
+ Assert.assertEquals(response.getMetrics().getOperationType(), OperationType.INSERT);
+ }
+
+ SpanData operation = spanByName("insert " + database + "." + TABLE);
+ Assert.assertEquals(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_ROWS), Long.valueOf(2L));
+ Assert.assertNotNull(longAttribute(operation, SpanAttribute.CLICKHOUSE_RESPONSE_WRITTEN_BYTES));
+ Assert.assertNull(longAttribute(operation, SpanAttribute.DB_RESPONSE_RETURNED_ROWS));
+ Assert.assertNull(longAttribute(operation, SpanAttribute.DB_OPERATION_BATCH_SIZE),
+ "a stream insert does not know the batch size");
+ Assert.assertEquals(operation.getStatus().getStatusCode(), StatusCode.UNSET);
+ }
+
+ private SpanData spanByName(String name) {
+ List spans = exporter.getFinishedSpanItems();
+ for (SpanData span : spans) {
+ if (name.equals(span.getName())) {
+ return span;
+ }
+ }
+ Assert.fail("No span named '" + name + "' in " + spans);
+ return null;
+ }
+
+ private static String stringAttribute(SpanData span, SpanAttribute attribute) {
+ return span.getAttributes().get(AttributeKey.stringKey(attribute.getKey()));
+ }
+
+ private static Long longAttribute(SpanData span, SpanAttribute attribute) {
+ return span.getAttributes().get(AttributeKey.longKey(attribute.getKey()));
+ }
+
+ public static class SpanRecorderPojo {
+
+ private int id;
+
+ private String name;
+
+ public int getId() {
+ return id;
+ }
+
+ public void setId(int id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+ }
+}
diff --git a/docs/features.md b/docs/features.md
index ff87a167f..ad4b2295e 100644
--- a/docs/features.md
+++ b/docs/features.md
@@ -35,7 +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.
+- Span recording (tracing SPI): `Client.Builder.setSpanRecorder(SpanRecorder)` registers a backend-agnostic recorder (package `com.clickhouse.client.api.observability`) that observes client operations. Every operation - a query, a command or an insert, including the `ping` and `getTableSchema` calls, which run a query - starts one operation span, and every transport request made for it, including each retry, starts a child request span. `SpanRecorder` and `Span` are plain interfaces; an implementation extends the `DefaultSpanRecorder` base class and overrides only what it wants to record, so a recorder keeps working when the client starts a kind of span it does not know about. The registered recorder is the first thing the client calls, and it is called with everything the client knows about the operation - its settings object, the statement, the target table, the batch size, the endpoint, the metrics of the completed operation and the failure - so an implementation is free to record whatever it needs and in whatever form. Deriving the standard span names and attribute values from those structures is done by `SpanSupport`, which a recorder implementation calls if it wants them; using it is opt-in, and its methods may be overridden to report other values. A recorder opts in through `DefaultSpanRecorder#getSpanSupport()` (or `SpanSupport.DEFAULT`). Span names and attribute keys follow the OpenTelemetry semantic conventions for database and HTTP client spans, the keys are defined by the `SpanAttribute` enum, and the values are derived by `SpanSupport`, so every recorder that uses it reports the same information. An operation span is named `query `, or `insert .` for an insert, and carries `db.system.name`, `db.namespace`, `clickhouse.query_id`, `db.query.text` (query/command), `db.query.parameter.` (parameterized query), `db.operation.name` (insert), `db.collection.name` (insert), `db.operation.batch.size` (POJO insert), `server.address`/`server.port` of the first configured endpoint (each attempt reports its own on the request span), the metrics of the completed operation on success, and `error.type` plus `db.response.status_code` on failure. Success is reported per operation kind, because the metrics that describe a read are not the ones that describe a write: `SpanRecorder#recordQuerySuccess` is called for a read operation and `SpanRecorder#recordInsertSuccess` for an insert, and each records only the metrics of its own kind - a query reports `db.response.returned_rows`, `clickhouse.response.read_rows` and `clickhouse.response.read_bytes`, an insert reports `clickhouse.response.written_rows` and `clickhouse.response.written_bytes`, and both report `clickhouse.query_id`. These values come from the server's progress summary, which is not always available (for example a query reports them with `QuerySettings#waitEndOfQuery(true)`); a metric the server did not report is left out. The same distinction is on the metrics themselves: `OperationMetrics#getOperationType()` returns `OperationType.QUERY` or `OperationType.INSERT`. It reports the kind of the call the application made, not the kind of work the server did, so a command that writes - `INSERT INTO ... SELECT` run through `execute` - is reported as a query. A request span is named `POST` and carries `http.request.method`, the `server.address`/`server.port` of that attempt, `http.response.status_code` once a response is received, and `error.type`/`db.response.status_code` when the attempt fails. An operation span is started on the calling thread, so it joins the caller's ambient trace even when `async_operations` runs the operation on the client's executor, and it is ended when the operation returns its response to the caller - so it covers sending the request and receiving the response head, not streaming the response body afterwards. Each span of an operation that started is ended exactly once, also when the operation failed; an operation that could not be started at all - a closed client, for example - does not report a span. When no recorder is registered nothing is recorded and no span-related work is done.
- 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.
@@ -115,3 +115,16 @@ Compatibility-sensitive traits:
- JDBC `ssl_mode` handling is compatibility-sensitive: values are case-insensitive, `none` is aliased to `trust` (the no-verification mode), and an unrecognized value throws `SQLException` during connection configuration. The normalized canonical mode name is forwarded to the underlying `client-v2` transport.
- Connection `Properties` values must be strings, with one scoped exception: the `ssl_context` key may carry a live `javax.net.ssl.SSLContext` object. Any other non-string property value still throws `IllegalArgumentException` during connection configuration. A string `ssl_context` (supplied via `setProperty` or a URL query parameter) is rejected with `SQLException`, since a string cannot represent a live context.
- INSERT result semantics depend on server-side `async_insert` and `wait_for_async_insert`. The driver does not override these settings, so it follows whatever the server profile or user configuration sets. When `async_insert=1` and `wait_for_async_insert=0`, `Statement.executeUpdate(...)` and `PreparedStatement.executeUpdate(...)` may return `0` (or an under-counted value), and parsing/data errors in the INSERT body may not be reported synchronously as a `SQLException`. Set `async_insert=0` (or `wait_for_async_insert=1`) per connection or statement to restore synchronous row counts and error reporting.
+
+## `client-v2` OpenTelemetry span recording
+
+- OpenTelemetry span recording: `new OpenTelemetrySpanRecorder(openTelemetry)` (package `com.clickhouse.client.api.observability.otel`) is a `SpanRecorder` that reports the client's operation and request spans to OpenTelemetry. It is registered like any other recorder, with `Client.Builder.setSpanRecorder(...)`. The no-argument constructor reports to `GlobalOpenTelemetry`, which it reads when a span is started, so a client may be built before the application installs its OpenTelemetry SDK; `new OpenTelemetrySpanRecorder(Tracer)` reports the spans under an instrumentation scope of the application's choice. The default scope name is `com.clickhouse.client`. `opentelemetry-api` is a compile-only (`provided`) dependency of `client-v2`: this recorder is usable only by an application that already provides the OpenTelemetry API at runtime, and it is not shaded into the `client-v2` `all` artifact or into `clickhouse-jdbc-all`, so a client that does not use it needs no OpenTelemetry on the classpath.
+- Recorded spans follow the `client-v2` span contract: the recorder derives every name and attribute through `SpanSupport`, so an operation span is named `query ` or `insert .`, a request span is named `POST`, and the recorded keys are the ones listed in `SpanAttribute`.
+
+Compatibility-sensitive traits:
+
+- Span kind and nesting should not drift: every span is a `CLIENT` span, an operation span is started as a child of the current OpenTelemetry context (so it joins the application's ambient trace), and each request span - including one per retry - is a child of its operation span. A request span whose operation span was not created by this recorder is started under the current OpenTelemetry context instead of failing.
+- Attribute value typing is part of the contract, because a backend indexes by type: a `String` value is recorded as a string attribute, a `Boolean` as a boolean, a `Double`/`Float` as a double, any other `Number` as a long, and any other value as its `String.valueOf` form. A `null` key or value records nothing.
+- A failure sets the OpenTelemetry span status to `ERROR`, records `error.type`, and records the failure itself as an OpenTelemetry exception event, so its message and stack trace are reported too; the ClickHouse error code and the HTTP status are recorded as separate attributes, not as the status description.
+- `Span#end()` is idempotent: a span is exported once even if it is ended more than once.
+- The recorder does not make any span current. The client hands its response to the caller before the response body is read, so spans are ended on threads the recorder does not control and an application that wants the client's span in its own context must make it current itself.
diff --git a/pom.xml b/pom.xml
index 7e402d341..66c9077c9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -93,6 +93,7 @@
2.10.1
4.0.1
0.31.1
+ 1.51.0
3.23.4
1.11.1
0.9.5