From 2a5ef56a2e63e59d3d9f088c577a19c384fadb76 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 2 Sep 2026 14:34:57 -0700 Subject: [PATCH] Added demo-spring-service as example for observability along with documentation and partner spec --- docs/clickhouse-docs/client.mdx | 167 +++++++++ docs/clickhouse-docs/jdbc.mdx | 56 +++ docs/integration-client.md | 168 +++++++++ docs/integration-index.md | 1 + docs/integration-jdbc.md | 58 +++ docs/integration-ops.md | 269 ++++++++++++++ examples/demo-spring-service/.gitignore | 16 + examples/demo-spring-service/README.md | 221 ++++++++++++ examples/demo-spring-service/compose.yaml | 15 + examples/demo-spring-service/pom.xml | 173 +++++++++ .../scripts/reconciliation-generator.py | 234 ++++++++++++ .../scripts/signal-generator.py | 235 ++++++++++++ .../scripts/slice-query-generator.py | 180 +++++++++ .../examples/IotIngestApplication.java | 21 ++ .../examples/config/AuthProperties.java | 29 ++ .../examples/config/ClickHouseDialect.java | 18 + .../examples/config/OpenTelemetryConfig.java | 107 ++++++ .../examples/config/ReconciliationConfig.java | 109 ++++++ .../examples/model/ReconciliationSignal.java | 135 +++++++ .../com/clickhouse/examples/model/Signal.java | 32 ++ .../examples/model/SignalEntity.java | 85 +++++ .../examples/model/SignalSlice.java | 20 + .../clickhouse/examples/model/SignalType.java | 17 + .../examples/repository/SignalRepository.java | 12 + .../schema/ClickHouseSchemaInitializer.java | 66 ++++ .../service/ReconciliationService.java | 53 +++ .../examples/service/SignalSliceService.java | 84 +++++ .../telemetry/ClickHouseMetricExporter.java | 122 +++++++ .../examples/telemetry/SignalMetrics.java | 79 ++++ .../examples/web/ApiKeyAuthFilter.java | 55 +++ .../examples/web/GlobalExceptionHandler.java | 68 ++++ .../web/ReconciliationController.java | 49 +++ .../examples/web/SignalController.java | 77 ++++ .../examples/web/SignalSliceController.java | 71 ++++ .../src/main/resources/application.yml | 63 ++++ .../examples/SignalIngestionTests.java | 341 ++++++++++++++++++ .../examples/TestIotIngestApplication.java | 18 + .../examples/TestcontainersConfiguration.java | 22 ++ 38 files changed, 3546 insertions(+) create mode 100644 docs/integration-ops.md create mode 100755 examples/demo-spring-service/.gitignore create mode 100755 examples/demo-spring-service/README.md create mode 100755 examples/demo-spring-service/compose.yaml create mode 100755 examples/demo-spring-service/pom.xml create mode 100755 examples/demo-spring-service/scripts/reconciliation-generator.py create mode 100755 examples/demo-spring-service/scripts/signal-generator.py create mode 100755 examples/demo-spring-service/scripts/slice-query-generator.py create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/IotIngestApplication.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/AuthProperties.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ClickHouseDialect.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/OpenTelemetryConfig.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ReconciliationConfig.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/ReconciliationSignal.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/Signal.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalEntity.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalSlice.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalType.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/repository/SignalRepository.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/schema/ClickHouseSchemaInitializer.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/ReconciliationService.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/SignalSliceService.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/ClickHouseMetricExporter.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/SignalMetrics.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ApiKeyAuthFilter.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/GlobalExceptionHandler.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ReconciliationController.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalController.java create mode 100755 examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalSliceController.java create mode 100755 examples/demo-spring-service/src/main/resources/application.yml create mode 100755 examples/demo-spring-service/src/test/java/com/clickhouse/examples/SignalIngestionTests.java create mode 100755 examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestIotIngestApplication.java create mode 100755 examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestcontainersConfiguration.java diff --git a/docs/clickhouse-docs/client.mdx b/docs/clickhouse-docs/client.mdx index 6a2ff98b2..acf805f59 100644 --- a/docs/clickhouse-docs/client.mdx +++ b/docs/clickhouse-docs/client.mdx @@ -899,6 +899,173 @@ try (QueryResponse response = client.query("SELECT * FROM my_table").get()) { } ``` +## Observability & Monitoring {#v2-o11y} + +The ClickHouse Java Client (V2) provides built-in observability features designed around OpenTelemetry semantic conventions and Micrometer metrics. The client enables monitoring operational durations, throughput, retries, server execution statistics, connection pool status, and distributed tracing spans. + +For a complete production-style Spring Boot application demonstrating Client V2 configuration, connection pooling, trace context propagation across `@Async` boundaries, and full telemetry integration, see the `examples/demo-spring-service` module in this repository. + +### Client Builder Observability API {#v2-o11y-builder-api} + +The `Client.Builder` exposes three primary methods to configure observability and metrics collection: + +- `setMetricsRecorder(MetricsRecorder recorder)`: Registers a metrics recorder for exporting operational metrics (durations, counts, retries, errors). When not explicitly set, defaults to `DefaultMetricsRecorder.NOOP`. +- `setSpanRecorder(SpanRecorder recorder)`: Registers a span recorder for distributed tracing (operation spans and transport request spans). When not explicitly set, defaults to `DefaultSpanRecorder.NOOP`. +- `registerClientMetrics(Object registry, String groupName)`: Binds Apache HttpClient connection pool metrics (such as available and leased connections, pending requests, and connection creation time) to a Micrometer `MeterRegistry`. + +**Dependencies Requirement** + +The built-in `OpenTelemetrySpanRecorder` and `MicrometerMetricsRecorder` classes are provided as part of the `client-v2` package. However, they require the OpenTelemetry API (`io.opentelemetry:opentelemetry-api`) and Micrometer (`io.micrometer:micrometer-core`) libraries to be present on the application's classpath at runtime. The main client artifact does not bundle these third-party dependencies nor pull them in transitively. Applications using OpenTelemetry tracing or Micrometer metrics must explicitly declare these dependencies in their project build configuration (`pom.xml` or `build.gradle`). + +**Builder Configuration Example** + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder; +import com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder; + +import io.micrometer.core.instrument.MeterRegistry; +import io.opentelemetry.api.OpenTelemetry; + +public Client createObservedClient(MeterRegistry meterRegistry, OpenTelemetry openTelemetry) { + return new Client.Builder() + .addEndpoint("http://localhost:8123") + .setUsername("default") + .setPassword("secret") + .setMetricsRecorder(new MicrometerMetricsRecorder(meterRegistry)) + .setSpanRecorder(new OpenTelemetrySpanRecorder(openTelemetry)) + .registerClientMetrics(meterRegistry, "clickhouse-client-pool") + .build(); +} +``` + +**Integration Approaches** + +Applications can integrate observability using two primary patterns: + +1. **OpenTelemetry Agent & Global Autoconfiguration**: + - For applications using the OpenTelemetry Java Agent or global SDK autoconfiguration, instantiate `OpenTelemetrySpanRecorder` and `MicrometerMetricsRecorder` using their default no-argument constructors. + - The default constructors automatically bind to `GlobalOpenTelemetry.get()` and Micrometer's `Metrics.globalRegistry`. +2. **Explicit Client-Level Integration**: + - Pass explicitly created `OpenTelemetry`, `Tracer`, or Micrometer `MeterRegistry` instances directly into the recorder constructors. + - Ideal when managing multiple isolated clients or multi-tenant services with independent telemetry sinks. + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder; +import com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder; + +public Client.Builder configureGlobalTelemetry(Client.Builder builder) { + return builder + .setMetricsRecorder(new MicrometerMetricsRecorder()) // uses Metrics.globalRegistry + .setSpanRecorder(new OpenTelemetrySpanRecorder()); // uses GlobalOpenTelemetry +} +``` + +### Metrics Reported {#v2-o11y-metrics} + +#### Operational Metrics + +Metrics exported via `MetricsRecorder` follow the OpenTelemetry semantic conventions for database client metrics (`MetricName` and `MetricAttribute`): + +| Metric Name | Type / Unit | Description | +|-------------|-------------|-------------| +| `db.client.operation.duration` | Timer (`s`) | Duration of a ClickHouse client operation (queries and inserts), recorded for both successful and failed operations. | +| `clickhouse.client.operation.serialization.duration` | Timer (`s`) | Duration of the serialization step (e.g. POJO serialization during inserts). | +| `clickhouse.client.operation.count` | Counter (`{operation}`) | Total number of completed client operations, grouped by outcome. | +| `clickhouse.client.operation.retries` | Counter (`{retry}`) | Number of retried attempts across client operations. | + +#### Metric Attributes / Tags + +Meters created by `MicrometerMetricsRecorder` carry low-cardinality tags (`MetricAttribute`): + +| Attribute Key | Description | Example Values | +|---------------|-------------|----------------| +| `db.system.name` | Database system identifier. | `clickhouse` | +| `db.namespace` | Target database name. | `default`, `analytics` | +| `db.operation.name` | Operation type. | `query`, `insert` | +| `db.collection.name` | Target table name (recorded for inserts). | `events` | +| `db.response.status_code` | ClickHouse server error code on failure. | `60` | +| `error.type` | Exception class name on failure, or `none` on success. | `com.clickhouse.client.api.ServerException` | + +#### Connection Pool Metrics + +When calling `registerClientMetrics(meterRegistry, "groupName")`, HTTP connection pool gauges are registered with Micrometer: + +- `httpcomponents.httpclient.pool.total.max`: Configured maximum allowed persistent connections. +- `httpcomponents.httpclient.pool.total.connections` (tagged with `state="available"` or `state="leased"`): Active connections in the pool. +- `httpcomponents.httpclient.pool.total.pending`: Number of connection requests awaiting a free connection. +- `httpcomponents.httpclient.pool.route.max.default`: Configured default maximum connections per route. +- `httpcomponents.httpclient.connect.time`: Running average connection establishment time. + +#### Response Metrics API + +In addition to metrics recorders, detailed execution statistics are accessible directly on response objects (`QueryResponse`, `InsertResponse`) via `OperationMetrics`: + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.metrics.ClientMetrics; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; +import com.clickhouse.client.api.query.QueryResponse; + +public void inspectResponseMetrics(Client client) throws Exception { + try (QueryResponse response = client.query("SELECT * FROM my_table").get()) { + OperationMetrics metrics = response.getMetrics(); + long readRows = metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(); + long readBytes = metrics.getMetric(ServerMetrics.NUM_BYTES_READ).getLong(); + long durationMs = metrics.getMetric(ClientMetrics.OP_DURATION).getLong(); + } +} +``` + +### Spans Reported {#v2-o11y-spans} + +Tracing in Client V2 produces two types of spans arranged in a parent-child hierarchy: + +1. **Operation Span**: Represents a high-level client operation (`query ` or `insert .`). Started under the current thread's trace context (`Context.current()`) so that database operations join the caller's active trace span. +2. **Transport Request Span**: Represents an individual HTTP transport request attempt (`POST`). Created as a child of the corresponding operation span. Each retry attempt creates a new transport request span under the operation span. + +#### Span Attributes + +Spans are populated with standard OpenTelemetry attributes (`SpanAttribute`): + +| Span Attribute Key | Scope / Type | Recorded Moment | Description | +|--------------------|--------------|-----------------|-------------| +| `db.system.name` | All operations | Before request start | Always `clickhouse`. | +| `db.namespace` | All operations | Before request start | Target database name. | +| `db.query.text` | Query | Before request start | SQL statement text for queries or commands. | +| `db.collection.name` | Insert | Before request start | Target table name. | +| `db.operation.name` | Insert | Before request start | Operation type (`insert`, `ping`, `getTableSchema`). | +| `db.operation.batch.size` | Insert | Before request start | Number of items in insert batch. | +| `db.query.parameter.` | Query | Before request start | Statement parameter values. | +| `clickhouse.query_id` | All operations | Before request start, updated on completion | ClickHouse query ID assigned by client or server. | +| `db.response.returned_rows` | Query | On success completion | Number of rows returned by server. | +| `clickhouse.response.read_rows` | Query | On success completion | Server rows read from storage. | +| `clickhouse.response.read_bytes` | Query | On success completion | Server bytes read from storage. | +| `clickhouse.response.written_rows` | Insert | On success completion | Server rows written to storage. | +| `clickhouse.response.written_bytes` | Insert | On success completion | Server bytes written to storage. | +| `server.address` | Transport request | Before attempt | Target server hostname or IP address. | +| `server.port` | Transport request | Before attempt | Target server port. | +| `http.request.method` | Transport request | Before attempt | HTTP method (always `POST`). | +| `http.response.status_code` | Transport request | On response / failure | HTTP status code returned by server (e.g. `200`, `500`). | +| `db.response.status_code` | All & Transport request | On failure | ClickHouse server error code on failure. | +| `error.type` | All & Transport request | On failure | Exception class name on error. | + +### Developing Custom Recorders {#v2-o11y-custom-recorders} + +Applications with custom observability backends can implement custom recorders by extending `DefaultMetricsRecorder` (for metrics) or `DefaultSpanRecorder` (for tracing). + +- **`DefaultMetricsRecorder`**: Base implementation of `MetricsRecorder`. Override specific callbacks like `recordQuerySuccess` or `recordQueryFailure`. Use `MetricsSupport.DEFAULT` to derive standard metric names (`MetricName`), units, and attribute maps (`MetricAttribute`). +- **`DefaultSpanRecorder`**: Base implementation of `SpanRecorder`. Override span creation callbacks like `startQuerySpan`, `startInsertSpan`, or `startRequestSpan`. Use `SpanSupport.DEFAULT` to calculate standard OpenTelemetry span names and attributes (`SpanAttribute`). + +For production reference implementations, see the built-in client classes: +- `com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder` (OpenTelemetry tracing integration) +- `com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder` (Micrometer metrics integration) + + + + ## Migration Guide {#migration_guide} Old client (V1) was using `com.clickhouse.client.ClickHouseClient#builder` as start point. The new client (V2) uses similar pattern with `com.clickhouse.client.api.Client.Builder`. Main diff --git a/docs/clickhouse-docs/jdbc.mdx b/docs/clickhouse-docs/jdbc.mdx index b70c83009..6c7f74790 100644 --- a/docs/clickhouse-docs/jdbc.mdx +++ b/docs/clickhouse-docs/jdbc.mdx @@ -711,6 +711,62 @@ try (PreparedStatement ps = conn.prepareStatement("INSERT INTO mytable VALUES (? } ``` + +## Observability & Monitoring {#jdbc-v2-o11y} + +The ClickHouse JDBC driver (V2) supports metrics and distributed tracing by leveraging the underlying Client V2 engine and standard JDBC application instrumentation. For detailed metric definitions, span attributes, and OpenTelemetry semantic conventions, see the [Client V2 Observability Documentation](/integrations/language-clients/java/client#v2-o11y). + +### Distributed Tracing & Spans + +When executing JDBC statements (`executeQuery`, `executeUpdate`, `executeBatch`), distributed tracing operates across a parent-child span hierarchy: + +1. **Outer JDBC / Application Span**: Created automatically by the OpenTelemetry Java Agent or APM instrumentation when intercepting `java.sql` method calls (e.g., `PreparedStatement.executeBatch()`). +2. **Client V2 Operation Span**: Created by the underlying Client V2 engine under the current active trace context (`Context.current()`). Carries ClickHouse-specific metadata such as `clickhouse.query_id`, statement text (`db.query.text`), and server execution statistics (`clickhouse.response.read_rows`, `clickhouse.response.written_rows`). +3. **Transport Request Span**: Created per HTTP POST request attempt to the ClickHouse server, recording endpoint details (`server.address`, `server.port`), HTTP status codes (`http.response.status_code`), and retry attempts. + +```text +Application HTTP Request Span + └── JDBC Statement Span (e.g., PreparedStatement.executeQuery) + └── Client V2 Operation Span (query default) + └── Transport Request Span (POST http://localhost:8123) +``` + +Because Client V2 inherits `Context.current()`, JDBC database spans automatically join the ambient trace of the surrounding HTTP or messaging context without manual context propagation. + +### JDBC Auto-Instrumentation + +Standard JDBC operations (`Connection`, `PreparedStatement`, `ResultSet`, `executeBatch`) are automatically tracked by Java application runtime frameworks and APM agents: + +- **OpenTelemetry Java Agent**: Automatically intercepts standard JDBC method invocations, creating trace spans for database executions (`SELECT`, `INSERT`, `TRUNCATE`) with `db.system=clickhouse`, statement text, and execution timings. +- **Spring Boot & Micrometer**: Spring Data JPA repositories and `JdbcTemplate` automatically instrument database calls when Spring Boot's Micrometer Observation or Spring Actuator metrics are active. + +### Configuring Driver Metrics Recorders + +You can enable Client V2 operational metrics (durations, counts, retries) for JDBC connections using the `jdbc_metrics_recorder` connection property. The driver instantiates the specified class via its public no-argument constructor per connection. + +```java +// Configure via JDBC URL parameter +String url = "jdbc:clickhouse://localhost:8123/default?jdbc_metrics_recorder=com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder"; + +// Or set via java.util.Properties +Properties properties = new Properties(); +properties.setProperty("jdbc_metrics_recorder", "com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder"); + +try (Connection conn = DriverManager.getConnection(url, properties)) { + // JDBC operations report operation metrics to Micrometer's globalRegistry +} +``` + +*Note: `MicrometerMetricsRecorder` is shipped with the driver, but requires `io.micrometer:micrometer-core` to be available on the application's runtime classpath.* + +### Spring Boot Demo Reference + +For a complete sample demonstrating JDBC driver setup and telemetry in a Spring Boot service, see the `examples/demo-spring-service` module in this repository: + +- **JDBC Datasource Setup** (`src/main/resources/application.yml`): Configures `spring.datasource` with `com.clickhouse.jdbc.ClickHouseDriver` and driver properties (`jdbc_ignore_unsupported_values: true`). +- **Spring Data JPA Integration** (`com.clickhouse.examples.repository.SignalRepository`): Shows repository-based entity persistence over ClickHouse JDBC. +- **`JdbcTemplate` Utilities** (`com.clickhouse.examples.schema.ClickHouseSchemaInitializer`): Demonstrates executing DDL and exporting metric points using Spring `JdbcTemplate`. + ## `HikariCP` {#hikaricp} ```java showLineNumbers diff --git a/docs/integration-client.md b/docs/integration-client.md index 05df55e08..daea5681d 100644 --- a/docs/integration-client.md +++ b/docs/integration-client.md @@ -37,6 +37,7 @@ Work through these steps in order. Each one is a decision point; the "Common Pit | 7 | [Write operations & tuning](#step-7--write-operations--tuning) | Insert pattern; heavy-ingest tuning; idempotency; write errors | | 8 | [Metadata & schema discovery](#step-8--metadata--schema-discovery) | How to obtain schemas without JDBC metadata | | 9 | [Miscellaneous features](#step-9--miscellaneous-features) | Sessions and other optional capabilities | +| 10 | [Observability & monitoring](#observability--monitoring) | Metrics, distributed tracing spans, and connection pool gauges | --- @@ -966,6 +967,172 @@ A [`Session`](../client-v2/src/main/java/com/clickhouse/client/api/Session.java) --- +## Observability & Monitoring + +The ClickHouse Java Client (V2) provides built-in observability features designed around OpenTelemetry semantic conventions and Micrometer metrics. The client enables monitoring operational durations, throughput, retries, server execution statistics, connection pool status, and distributed tracing spans. + +For a complete production-style Spring Boot application demonstrating Client V2 configuration, connection pooling, trace context propagation across `@Async` boundaries, and full telemetry integration, see the `examples/demo-spring-service` module in this repository. + +### Client Builder Observability API + +The `Client.Builder` exposes three primary methods to configure observability and metrics collection: + +- `setMetricsRecorder(MetricsRecorder recorder)` — Registers a metrics recorder for exporting operational metrics (durations, counts, retries, errors). When not explicitly set, defaults to `DefaultMetricsRecorder.NOOP`. +- `setSpanRecorder(SpanRecorder recorder)` — Registers a span recorder for distributed tracing (operation spans and transport request spans). When not explicitly set, defaults to `DefaultSpanRecorder.NOOP`. +- `registerClientMetrics(Object registry, String groupName)` — Binds Apache HttpClient connection pool metrics (such as available and leased connections, pending requests, and connection creation time) to a Micrometer `MeterRegistry`. + +**Dependencies Requirement** + +The built-in `OpenTelemetrySpanRecorder` and `MicrometerMetricsRecorder` classes are provided as part of the `client-v2` package. However, they require the OpenTelemetry API (`io.opentelemetry:opentelemetry-api`) and Micrometer (`io.micrometer:micrometer-core`) libraries to be present on the application's classpath at runtime. The main client artifact does not bundle these third-party dependencies nor pull them in transitively. Applications using OpenTelemetry tracing or Micrometer metrics must explicitly declare these dependencies in their project build configuration (`pom.xml` or `build.gradle`). + +**Builder Configuration Example** + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder; +import com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder; + +import io.micrometer.core.instrument.MeterRegistry; +import io.opentelemetry.api.OpenTelemetry; + +public Client createObservedClient(MeterRegistry meterRegistry, OpenTelemetry openTelemetry) { + return new Client.Builder() + .addEndpoint("http://localhost:8123") + .setUsername("default") + .setPassword("secret") + .setMetricsRecorder(new MicrometerMetricsRecorder(meterRegistry)) + .setSpanRecorder(new OpenTelemetrySpanRecorder(openTelemetry)) + .registerClientMetrics(meterRegistry, "clickhouse-client-pool") + .build(); +} +``` + +**Integration Approaches** + +Applications can integrate observability using two primary patterns: + +1. **OpenTelemetry Agent & Global Autoconfiguration:** + - For applications using the OpenTelemetry Java Agent or global SDK autoconfiguration, instantiate `OpenTelemetrySpanRecorder` and `MicrometerMetricsRecorder` using their default no-argument constructors. + - The default constructors automatically bind to `GlobalOpenTelemetry.get()` and Micrometer's `Metrics.globalRegistry`. +2. **Explicit Client-Level Integration:** + - Pass explicitly created `OpenTelemetry`, `Tracer`, or Micrometer `MeterRegistry` instances directly into the recorder constructors. + - Ideal when managing multiple isolated clients or multi-tenant services with independent telemetry sinks. + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder; +import com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder; + +public Client.Builder configureGlobalTelemetry(Client.Builder builder) { + return builder + .setMetricsRecorder(new MicrometerMetricsRecorder()) // uses Metrics.globalRegistry + .setSpanRecorder(new OpenTelemetrySpanRecorder()); // uses GlobalOpenTelemetry +} +``` + +### Metrics Reported + +#### Operational Metrics + +Metrics exported via `MetricsRecorder` follow the OpenTelemetry semantic conventions for database client metrics (`MetricName` and `MetricAttribute`): + +| Metric Name | Type / Unit | Description | +|-------------|-------------|-------------| +| `db.client.operation.duration` | Timer (`s`) | Duration of a ClickHouse client operation (queries and inserts), recorded for both successful and failed operations. | +| `clickhouse.client.operation.serialization.duration` | Timer (`s`) | Duration of the serialization step (e.g. POJO serialization during inserts). | +| `clickhouse.client.operation.count` | Counter (`{operation}`) | Total number of completed client operations, grouped by outcome. | +| `clickhouse.client.operation.retries` | Counter (`{retry}`) | Number of retried attempts across client operations. | + +#### Metric Attributes / Tags + +Meters created by `MicrometerMetricsRecorder` carry low-cardinality tags (`MetricAttribute`): + +| Attribute Key | Description | Example Values | +|---------------|-------------|----------------| +| `db.system.name` | Database system identifier. | `clickhouse` | +| `db.namespace` | Target database name. | `default`, `analytics` | +| `db.operation.name` | Operation type. | `query`, `insert` | +| `db.collection.name` | Target table name (recorded for inserts). | `events` | +| `db.response.status_code` | ClickHouse server error code on failure. | `60` | +| `error.type` | Exception class name on failure, or `none` on success. | `com.clickhouse.client.api.ServerException` | + +#### Connection Pool Metrics + +When calling `registerClientMetrics(meterRegistry, "groupName")`, HTTP connection pool gauges are registered with Micrometer: + +- `httpcomponents.httpclient.pool.total.max`: Configured maximum allowed persistent connections. +- `httpcomponents.httpclient.pool.total.connections` (tagged with `state="available"` or `state="leased"`): Active connections in the pool. +- `httpcomponents.httpclient.pool.total.pending`: Number of connection requests awaiting a free connection. +- `httpcomponents.httpclient.pool.route.max.default`: Configured default maximum connections per route. +- `httpcomponents.httpclient.connect.time`: Running average connection establishment time. + +#### Response Metrics API + +In addition to metrics recorders, detailed execution statistics are accessible directly on response objects (`QueryResponse`, `InsertResponse`) via `OperationMetrics`: + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.metrics.ClientMetrics; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; +import com.clickhouse.client.api.query.QueryResponse; + +public void inspectResponseMetrics(Client client) throws Exception { + try (QueryResponse response = client.query("SELECT * FROM my_table").get()) { + OperationMetrics metrics = response.getMetrics(); + long readRows = metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(); + long readBytes = metrics.getMetric(ServerMetrics.NUM_BYTES_READ).getLong(); + long durationMs = metrics.getMetric(ClientMetrics.OP_DURATION).getLong(); + } +} +``` + +### Spans Reported + +Tracing in Client V2 produces two types of spans arranged in a parent-child hierarchy: + +1. **Operation Span:** Represents a high-level client operation (`query ` or `insert .`). Started under the current thread's trace context (`Context.current()`) so that database operations join the caller's active trace span. +2. **Transport Request Span:** Represents an individual HTTP transport request attempt (`POST`). Created as a child of the corresponding operation span. Each retry attempt creates a new transport request span under the operation span. + +#### Span Attributes + +Spans are populated with standard OpenTelemetry attributes (`SpanAttribute`): + +| Span Attribute Key | Scope / Type | Recorded Moment | Description | +|--------------------|--------------|-----------------|-------------| +| `db.system.name` | All operations | Before request start | Always `clickhouse`. | +| `db.namespace` | All operations | Before request start | Target database name. | +| `db.query.text` | Query | Before request start | SQL statement text for queries or commands. | +| `db.collection.name` | Insert | Before request start | Target table name. | +| `db.operation.name` | Insert | Before request start | Operation type (`insert`, `ping`, `getTableSchema`). | +| `db.operation.batch.size` | Insert | Before request start | Number of items in insert batch. | +| `db.query.parameter.` | Query | Before request start | Statement parameter values. | +| `clickhouse.query_id` | All operations | Before request start, updated on completion | ClickHouse query ID assigned by client or server. | +| `db.response.returned_rows` | Query | On success completion | Number of rows returned by server. | +| `clickhouse.response.read_rows` | Query | On success completion | Server rows read from storage. | +| `clickhouse.response.read_bytes` | Query | On success completion | Server bytes read from storage. | +| `clickhouse.response.written_rows` | Insert | On success completion | Server rows written to storage. | +| `clickhouse.response.written_bytes` | Insert | On success completion | Server bytes written to storage. | +| `server.address` | Transport request | Before attempt | Target server hostname or IP address. | +| `server.port` | Transport request | Before attempt | Target server port. | +| `http.request.method` | Transport request | Before attempt | HTTP method (always `POST`). | +| `http.response.status_code` | Transport request | On response / failure | HTTP status code returned by server (e.g. `200`, `500`). | +| `db.response.status_code` | All & Transport request | On failure | ClickHouse server error code on failure. | +| `error.type` | All & Transport request | On failure | Exception class name on error. | + +### Developing Custom Recorders + +Applications with custom observability backends can implement custom recorders by extending `DefaultMetricsRecorder` (for metrics) or `DefaultSpanRecorder` (for tracing). + +- **`DefaultMetricsRecorder`**: Base implementation of `MetricsRecorder`. Override specific callbacks like `recordQuerySuccess` or `recordQueryFailure`. Use `MetricsSupport.DEFAULT` to derive standard metric names (`MetricName`), units, and attribute maps (`MetricAttribute`). +- **`DefaultSpanRecorder`**: Base implementation of `SpanRecorder`. Override span creation callbacks like `startQuerySpan`, `startInsertSpan`, or `startRequestSpan`. Use `SpanSupport.DEFAULT` to calculate standard OpenTelemetry span names and attributes (`SpanAttribute`). + +For production reference implementations, see the built-in client classes: +- `com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder` (OpenTelemetry tracing integration) +- `com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder` (Micrometer metrics integration) + +--- + ## Error model This is the shared exception reference used by the read ([Step 6](#step-6--read-operations--tuning)) and write ([Step 7](#step-7--write-operations--tuning)) error sections. All exceptions extend [`ClickHouseException`](../client-v2/src/main/java/com/clickhouse/client/api/ClickHouseException.java) (an unchecked `RuntimeException`). Because operations return `CompletableFuture`, a failed operation surfaces its cause wrapped in `java.util.concurrent.ExecutionException` when you call `.get()`; unwrap it with `getCause()`. @@ -1017,5 +1184,6 @@ public void executeQueryWithErrorHandling(Client client, String sql) throws Exce - [integration-index.md](integration-index.md) — choosing JDBC vs Client - [integration-jdbc.md](integration-jdbc.md) — JDBC integration path +- [integration-ops.md](integration-ops.md) — operations and observability guide - [authentication.md](authentication.md) — full authentication and TLS reference (referenced from Steps 2–3) - [features.md](features.md) — compatibility contract (referenced from Step 5) diff --git a/docs/integration-index.md b/docs/integration-index.md index d8a158067..ef2540752 100644 --- a/docs/integration-index.md +++ b/docs/integration-index.md @@ -9,6 +9,7 @@ This document is the starting point for integrating ClickHouse into a Java appli | This guide | Anyone evaluating options | this document | | Java Client path | New applications, high-throughput pipelines, custom data processing | [integration-client.md](integration-client.md) | | JDBC path | Existing JDBC-based stacks, BI tools, ORMs | [integration-jdbc.md](integration-jdbc.md) | +| Operations & Observability | SREs, monitoring setup, connection pool tuning, troubleshooting | [integration-ops.md](integration-ops.md) | Reference information should be fetched from official documentation for [Java Client](https://clickhouse.com/docs/integrations/language-clients/java/client) or [JDBC Driver](https://clickhouse.com/docs/integrations/language-clients/java/jdbc). diff --git a/docs/integration-jdbc.md b/docs/integration-jdbc.md index 4d35fa885..690808921 100644 --- a/docs/integration-jdbc.md +++ b/docs/integration-jdbc.md @@ -760,6 +760,63 @@ Key JDBC-specific properties (see [`DriverProperties`](../jdbc-v2/src/main/java/ | `jdbc_cluster_name` | — | Cluster for `KILL QUERY ON CLUSTER` | | `jdbc_type_mappings` | — | Custom ClickHouse → Java type overrides | | `default_query_settings` | — | Default settings for all queries | +| `jdbc_metrics_recorder` | — | Custom `MetricsRecorder` implementation class name | + + +## Observability & Monitoring + +The ClickHouse JDBC driver (V2) supports metrics and distributed tracing by leveraging the underlying Client V2 engine and standard JDBC application instrumentation. For detailed metric definitions, span attributes, and OpenTelemetry semantic conventions, see the [Java Client Observability Documentation](integration-client.md#observability--monitoring). + +### Distributed Tracing & Spans + +When executing JDBC statements (`executeQuery`, `executeUpdate`, `executeBatch`), distributed tracing operates across a parent-child span hierarchy: + +1. **Outer JDBC / Application Span:** Created automatically by the OpenTelemetry Java Agent or APM instrumentation when intercepting `java.sql` method calls (e.g., `PreparedStatement.executeBatch()`). +2. **Client V2 Operation Span:** Created by the underlying Client V2 engine under the current active trace context (`Context.current()`). Carries ClickHouse-specific metadata such as `clickhouse.query_id`, statement text (`db.query.text`), and server execution statistics (`clickhouse.response.read_rows`, `clickhouse.response.written_rows`). +3. **Transport Request Span:** Created per HTTP POST request attempt to the ClickHouse server, recording endpoint details (`server.address`, `server.port`), HTTP status codes (`http.response.status_code`), and retry attempts. + +```text +Application HTTP Request Span + └── JDBC Statement Span (e.g., PreparedStatement.executeQuery) + └── Client V2 Operation Span (query default) + └── Transport Request Span (POST http://localhost:8123) +``` + +Because Client V2 inherits `Context.current()`, JDBC database spans automatically join the ambient trace of the surrounding HTTP or messaging context without manual context propagation. + +### JDBC Auto-Instrumentation + +Standard JDBC operations (`Connection`, `PreparedStatement`, `ResultSet`, `executeBatch`) are automatically tracked by Java application runtime frameworks and APM agents: + +- **OpenTelemetry Java Agent:** Automatically intercepts standard JDBC method invocations, creating trace spans for database executions (`SELECT`, `INSERT`, `TRUNCATE`) with `db.system=clickhouse`, statement text, and execution timings. +- **Spring Boot & Micrometer:** Spring Data JPA repositories and `JdbcTemplate` automatically instrument database calls when Spring Boot's Micrometer Observation or Spring Actuator metrics are active. + +### Configuring Driver Metrics Recorders + +You can enable Client V2 operational metrics (durations, counts, retries) for JDBC connections using the `jdbc_metrics_recorder` connection property. The driver instantiates the specified class via its public no-argument constructor per connection. + +```java +// Configure via JDBC URL parameter +String url = "jdbc:clickhouse://localhost:8123/default?jdbc_metrics_recorder=com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder"; + +// Or set via java.util.Properties +Properties properties = new Properties(); +properties.setProperty("jdbc_metrics_recorder", "com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder"); + +try (Connection conn = DriverManager.getConnection(url, properties)) { + // JDBC operations report operation metrics to Micrometer's globalRegistry +} +``` + +*Note: `MicrometerMetricsRecorder` is shipped with the driver, but requires `io.micrometer:micrometer-core` to be available on the application's runtime classpath.* + +### Spring Boot Demo Reference + +For a complete sample demonstrating JDBC driver setup and telemetry in a Spring Boot service, see the `examples/demo-spring-service` module in this repository: + +- **JDBC Datasource Setup** (`src/main/resources/application.yml`): Configures `spring.datasource` with `com.clickhouse.jdbc.ClickHouseDriver` and driver properties (`jdbc_ignore_unsupported_values: true`). +- **Spring Data JPA Integration** (`com.clickhouse.examples.repository.SignalRepository`): Shows repository-based entity persistence over ClickHouse JDBC. +- **`JdbcTemplate` Utilities** (`com.clickhouse.examples.schema.ClickHouseSchemaInitializer`): Demonstrates executing DDL and exporting metric points using Spring `JdbcTemplate`. ## References @@ -778,6 +835,7 @@ Key JDBC-specific properties (see [`DriverProperties`](../jdbc-v2/src/main/java/ - [integration-common.md](integration-common.md) — choosing JDBC vs Client - [integration-client.md](integration-client.md) — Java Client integration path +- [integration-ops.md](integration-ops.md) — operations and observability guide - [authentication.md](authentication.md) — full authentication and TLS reference - [features.md](features.md) — compatibility contract - [type_mapping.md](../type_mapping.md) — JDBC type mapping recommendations \ No newline at end of file diff --git a/docs/integration-ops.md b/docs/integration-ops.md new file mode 100644 index 000000000..14c22cc22 --- /dev/null +++ b/docs/integration-ops.md @@ -0,0 +1,269 @@ +# ClickHouse Java Operations & Observability Guide + +This guide covers operational monitoring, connection pool management, telemetry instrumentation, and troubleshooting for Java applications using ClickHouse (`com.clickhouse:client-v2` and `com.clickhouse.clickhouse-jdbc`). + +--- + +## Overview of Workloads & Integration Layers + +The `clickhouse-java` ecosystem provides two primary integration layers built on the same underlying HTTP transport engine (`client-v2` HTTP client helper): + +1. **Java Client (`client-v2`)**: Native asynchronous and streaming API. Ideal for high-throughput microservices, event streaming consumers, and bulk ingestion/analytical workloads. +2. **JDBC Driver (`clickhouse-jdbc`)**: JDBC 4.2 compliant driver wrapping the Java Client internally. Ideal for Spring Data JPA, Hibernate, BI tools, and ORM-based applications. + +Because both layers share the same Apache HttpClient HTTP transport stack, connection pooling, metrics collection, and distributed tracing work consistently across both direct Java Client usage and JDBC driver connections. + +--- + +## Connection Pools & Resource Management + +### HTTP Connection Pooling (`client-v2`) + +The `Client` instance owns an internal Apache HttpClient 5 connection pool. It manages persistent HTTP connections to ClickHouse endpoints. + +#### Key Pool Settings + +| Setting | Builder Method | Default | Description | +|---------|----------------|---------|-------------| +| Max Connections | `setMaxConnections(int)` | `10` | Maximum open HTTP connections per server endpoint. | +| Connection TTL | `setConnectionTTL(long, TimeUnit)` | `-1` (disabled) | Time-to-live after which an active connection is closed and recreated. | +| Keep-Alive Timeout | `setKeepAliveTimeout(long, TimeUnit)` | Server default | HTTP Keep-Alive duration for idle pooled connections. | +| Connection Request Timeout | `setConnectionRequestTimeout(long, TimeUnit)` | `10000ms` | Maximum time a thread blocks waiting for an available connection from the pool. | +| Reuse Strategy | `setConnectionReuseStrategy(ConnectionReuseStrategy)` | `FIFO` | Connection pool allocation strategy (`FIFO` or `LIFO`). | + +#### Recommended Pool Configuration + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.enums.ConnectionReuseStrategy; + +import java.util.concurrent.TimeUnit; + +public Client createOptimizedClient() { + return new Client.Builder() + .addEndpoint("http://localhost:8123") + .setUsername("default") + .setPassword("secret") + .setMaxConnections(50) + .setConnectionRequestTimeout(5, TimeUnit.SECONDS) + .setKeepAliveTimeout(60, TimeUnit.SECONDS) + .setConnectionReuseStrategy(ConnectionReuseStrategy.FIFO) + .build(); +} +``` + +### JDBC Pooling Considerations (HikariCP & Frameworks) + +When using the JDBC driver (`clickhouse-jdbc`) with an external connection pooler like **HikariCP**: + +- **Layering:** HikariCP manages JDBC `Connection` instances, while each JDBC connection wraps a `Client` instance with its own internal HTTP socket pool. +- **Sizing Alignment:** Avoid over-allocating HikariCP connections. Because ClickHouse processes HTTP requests concurrently over pooled sockets, a smaller HikariCP pool (e.g., 10–20 connections) paired with a properly sized `Client` HTTP pool is typically optimal. +- **Connection Lifecycle:** Configure HikariCP's `maxLifetime` slightly shorter than any network or load-balancer idle timeout to prevent stale socket exceptions. + +--- + +## Metrics & Monitoring + +Observability in `clickhouse-java` is split into operational metrics (reported via `MetricsRecorder`), HTTP connection pool metrics (bound to Micrometer), and in-band response statistics (`OperationMetrics`). + +### Operational Metrics (`MetricsRecorder`) + +Client V2 defines standard database client metrics following OpenTelemetry semantic conventions (`MetricName` and `MetricAttribute`): + +| Metric Name | Type / Unit | Description | +|-------------|-------------|-------------| +| `db.client.operation.duration` | Timer (`s`) | Total duration of a ClickHouse client operation (queries and inserts), recorded for both successful and failed operations. | +| `clickhouse.client.operation.serialization.duration` | Timer (`s`) | Duration of client-side serialization (e.g., POJO encoding during inserts). | +| `clickhouse.client.operation.count` | Counter (`{operation}`) | Total number of completed client operations, grouped by outcome tags. | +| `clickhouse.client.operation.retries` | Counter (`{retry}`) | Number of retried attempts across client operations. | + +#### Low-Cardinality Metric Tags / Attributes + +Meters exported via `MicrometerMetricsRecorder` carry low-cardinality tags (`MetricAttribute`): + +| Tag Key | Description | Example Values | +|---------|-------------|----------------| +| `db.system.name` | Database system identifier | `clickhouse` | +| `db.namespace` | Target ClickHouse database | `default`, `analytics` | +| `db.operation.name` | Operation type | `query`, `insert` | +| `db.collection.name` | Target table name (recorded for inserts) | `events` | +| `db.response.status_code` | Server error code on failure | `60` | +| `error.type` | Exception class name on failure (`none` on success) | `com.clickhouse.client.api.ServerException` | + +### Connection Pool Gauges + +Binding connection pool metrics exposes Apache HttpClient 5 pool statistics to Micrometer via `.registerClientMetrics(meterRegistry, "groupName")`: + +| Meter Name | Tags | Description | +|------------|------|-------------| +| `httpcomponents.httpclient.pool.total.max` | `httpclient=` | Configured maximum allowed persistent connections across all routes. | +| `httpcomponents.httpclient.pool.total.connections` | `httpclient=`, `state=available` / `leased` | Number of persistent available or active leased connections. | +| `httpcomponents.httpclient.pool.total.pending` | `httpclient=` | Number of threads currently blocked awaiting a free pooled connection. | +| `httpcomponents.httpclient.connect.time` | `httpclient=` | Running average connection establishment time. | + +### In-Band Response Metrics API + +Applications can inspect execution metrics directly from response objects (`QueryResponse`, `InsertResponse`) via `OperationMetrics`: + +```java +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.metrics.ClientMetrics; +import com.clickhouse.client.api.metrics.OperationMetrics; +import com.clickhouse.client.api.metrics.ServerMetrics; +import com.clickhouse.client.api.query.QueryResponse; + +public void inspectResponseMetrics(Client client) throws Exception { + try (QueryResponse response = client.query("SELECT * FROM my_table").get()) { + OperationMetrics metrics = response.getMetrics(); + + long readRows = metrics.getMetric(ServerMetrics.NUM_ROWS_READ).getLong(); + long readBytes = metrics.getMetric(ServerMetrics.NUM_BYTES_READ).getLong(); + long writtenRows = metrics.getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong(); + long clientDurationMs = metrics.getMetric(ClientMetrics.OP_DURATION).getLong(); + + String queryId = metrics.getQueryId(); + } +} +``` + +--- + +## Distributed Tracing & Spans + +Client V2 supports OpenTelemetry distributed tracing across a structured 3-tier parent-child span hierarchy. + +### Span Hierarchy + +```text +Application HTTP / Messaging Span (e.g. Spring Controller / Kafka Consumer) + └── [JDBC Path Only] JDBC Statement Span (e.g., PreparedStatement.executeBatch) + └── Client V2 Operation Span (query or insert .) + └── Transport Request Span (POST http://localhost:8123) +``` + +1. **Operation Span**: High-level operation created under the current active trace context (`Context.current()`). +2. **Transport Request Span**: Individual HTTP transport request attempt (`POST`). Created per attempt, so retries generate separate request spans under the same operation span. + +### Span Attributes Reference + +Spans are populated with standard OpenTelemetry attributes (`SpanAttribute`): + +| Span Attribute Key | Scope / Type | Recorded Moment | Description | +|--------------------|--------------|-----------------|-------------| +| `db.system.name` | All operations | Before request start | Always `clickhouse`. | +| `db.namespace` | All operations | Before request start | Target database name. | +| `db.query.text` | Query | Before request start | SQL statement text. | +| `db.collection.name` | Insert | Before request start | Target table name. | +| `db.operation.name` | Insert | Before request start | Operation type (`insert`, `ping`, `getTableSchema`). | +| `db.operation.batch.size` | Insert | Before request start | Batch row count. | +| `db.query.parameter.` | Query | Before request start | Statement parameter values. | +| `clickhouse.query_id` | All operations | Before request start, updated on completion | ClickHouse query ID assigned by client or server. | +| `db.response.returned_rows` | Query | On success completion | Rows returned to caller. | +| `clickhouse.response.read_rows` | Query | On success completion | Rows read from storage by server. | +| `clickhouse.response.read_bytes` | Query | On success completion | Bytes read from storage by server. | +| `clickhouse.response.written_rows` | Insert | On success completion | Rows written to storage by server. | +| `clickhouse.response.written_bytes` | Insert | On success completion | Bytes written to storage by server. | +| `server.address` | Transport request | Before attempt | Target server hostname or IP address. | +| `server.port` | Transport request | Before attempt | Target server port. | +| `http.request.method` | Transport request | Before attempt | HTTP method (always `POST`). | +| `http.response.status_code` | Transport request | On response / failure | HTTP status code (e.g. `200`, `500`). | +| `db.response.status_code` | All & Transport request | On failure | ClickHouse server error code. | +| `error.type` | All & Transport request | On failure | Exception class name on error. | + +### Context Propagation across Asynchronous Boundaries + +Because Client V2 inherits `Context.current()`, operations started on threads with an active trace span automatically join the trace. When delegating tasks across thread pools (e.g., Spring `@Async` or custom `ExecutorService`), propagate the OpenTelemetry context explicitly: + +```java +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.insert.InsertResponse; + +public void processBatchAsync(Client client, String table, List data, Context parentContext) throws Exception { + try (Scope ignored = parentContext.makeCurrent()) { + try (InsertResponse response = client.insert(table, data).get()) { + // Background Client V2 span joins parentContext trace + } + } +} +``` + +--- + +## Troubleshooting & Diagnostics + +### 1. Diagnosing Connection Pool Starvation + +**Symptom:** Threads block or throw `ConnectionInitiationException` / `ConnectionRequestTimeoutException` with messages indicating connection acquisition timeout. + +**Diagnosis:** +- Monitor the `httpcomponents.httpclient.pool.total.pending` gauge in Grafana / Prometheus. A non-zero or spiking pending count indicates thread contention for HTTP sockets. +- Compare `httpcomponents.httpclient.pool.total.connections{state="leased"}` against `httpcomponents.httpclient.pool.total.max`. + +**Remediation:** +- Increase `Client.Builder.setMaxConnections(...)` to match concurrent thread demand. +- Ensure all query and insert response objects (`QueryResponse`, `InsertResponse`) are closed promptly using `try-with-resources`. + +### 2. Correlating Application Traces with ClickHouse Server Logs (`system.query_log`) + +**Symptom:** Need to trace an expensive or failing query from APM traces down to ClickHouse server execution logs. + +**Solution:** +- The client records `clickhouse.query_id` on every operation span. +- You can supply a custom query ID generator during client setup: + ```java + clientBuilder.setQueryIdGenerator(() -> UUID.randomUUID().toString()); + ``` +- Query `system.query_log` in ClickHouse using the recorded query ID: + ```sql + SELECT query_id, type, query_duration_ms, read_rows, read_bytes, memory_usage, exception + FROM system.query_log + WHERE query_id = 'your-recorded-query-id' + ORDER BY event_time DESC; + ``` + +### 3. Missing Classpath Dependencies + +**Symptom:** `NoClassDefFoundError: io/opentelemetry/api/...` or `java.lang.NoClassDefFoundError: io/micrometer/core/...` at runtime when using `OpenTelemetrySpanRecorder` or `MicrometerMetricsRecorder`. + +**Cause:** The main `client-v2` artifact includes recorder classes, but does not bundle or transitively pull in OpenTelemetry or Micrometer dependencies. + +**Solution:** Explicitly declare the required telemetry libraries in your `pom.xml` or `build.gradle`: + +```xml + + + io.opentelemetry + opentelemetry-api + 1.38.0 + + + + + io.micrometer + micrometer-core + 1.13.0 + +``` + +### 4. High Serialization Duration + +**Symptom:** `clickhouse.client.operation.serialization.duration` takes a significant portion of total operation duration (`db.client.operation.duration`). + +**Diagnosis & Remediation:** +- For POJO inserts, high serialization duration points to expensive reflection or large batch encoding. +- Consider tuning batch sizes or utilizing direct binary stream writers (`RowBinaryFormatWriter`) for ultra-high throughput paths. + +### 5. Distinguishing Transport Failures vs Server Exceptions + +- **Server Exception:** Indicated by non-null `db.response.status_code` tag (e.g. `60` for missing table, `159` for timeout). Represents ClickHouse server rejecting the query. +- **Transport / Connection Failure:** `db.response.status_code` is absent, and `error.type` indicates `ConnectionInitiationException`, `DataTransferException`, or `NoHttpResponseException`. Indicates network, proxy, or server availability issues. + +--- + +## References & Demos + +- **Spring Boot Telemetry Demo:** See `examples/demo-spring-service` in this repository for a complete implementation showing Client V2 setup, JDBC usage, connection pool metrics, and direct metrics exporting to ClickHouse. +- **Java Client Integration Guide:** [docs/integration-client.md](integration-client.md) +- **JDBC Integration Guide:** [docs/integration-jdbc.md](integration-jdbc.md) diff --git a/examples/demo-spring-service/.gitignore b/examples/demo-spring-service/.gitignore new file mode 100755 index 000000000..8a9ea1cb6 --- /dev/null +++ b/examples/demo-spring-service/.gitignore @@ -0,0 +1,16 @@ +target/ +*.class +*.log +__pycache__/ +*.pyc + +# IDE +.idea/ +*.iml +.vscode/ +.settings/ +.project +.classpath + +# OS +.DS_Store diff --git a/examples/demo-spring-service/README.md b/examples/demo-spring-service/README.md new file mode 100755 index 000000000..48f672fd8 --- /dev/null +++ b/examples/demo-spring-service/README.md @@ -0,0 +1,221 @@ +# IoT Ingest Demo Service + +A Spring Boot service demonstrating +- General integration of JDBC driver with Spring Data. +- ClickHouse Client (not JDBC) configuration and usage with Spring Boot. +- Different observability options (instrumentation vs client own observability configuration). + +--- + +## What This Demo Shows + +This sample service ingests IoT telemetry (temperature, humidity, pressure, motion, etc.) and demonstrates two complementary ways to work with ClickHouse in Spring Boot: + +1. **Spring Data JPA / JDBC** – Standard entity persistence for single-signal API requests (`POST /api/v1/signals`). +2. **Direct ClickHouse Client V2** – High-performance direct API for: + - **Async Batch Reconciliation** (`POST /api/v1/reconciliation`): High-throughput POJO batch inserts off the main HTTP thread. + - **Analytical Aggregation Queries** (`GET /api/v1/signals/slices`): Real-time 1-second location window aggregates. + +### Key Highlights +- **Zero-Setup Local Database**: Uses **Testcontainers** to automatically start a ClickHouse instance during development and testing—no external database setup required. +- **Full Observability**: Collects OpenTelemetry metrics and traces, exporting them to both ClickHouse and a local **Grafana LGTM** stack (Prometheus, Tempo, Grafana). +- **API Key Authentication**: Simple header-based (`X-API-Key`) security filter. + +--- + +## How to Run + +### Prerequisites +- **Java 17+** +- **Docker** (for the Grafana LGTM stack and Testcontainers) + +### 1. Start the Observability Stack (Optional) +Start local Prometheus, Tempo, and Grafana containers: + +```bash +docker compose up -d +``` + +### 2. Launch the Application +Run the Spring Boot service. Testcontainers will automatically launch ClickHouse: + +```bash +./mvnw spring-boot:test-run +``` + +The app starts on `http://localhost:8080`, initializes the `iot_signals` and `otel_metrics` tables in ClickHouse, and begins exporting telemetry. Press `Ctrl+C` to stop. + +> **Running against an external ClickHouse?** Run `./mvnw spring-boot:run` with `CLICKHOUSE_URL=jdbc:clickhouse://:8123/default`. + +### 3. Try the Endpoints + +#### Send a single signal (Spring Data JPA) +```bash +curl -i -X POST http://localhost:8080/api/v1/signals \ + -H 'Content-Type: application/json' \ + -H 'X-API-Key: dev-key-1' \ + -d '{"deviceId":"sensor-1","locationId":"11111111-1111-1111-1111-111111111111","type":"TEMPERATURE","value":21.5,"unit":"C"}' +``` +*Supported types: `TEMPERATURE`, `HUMIDITY`, `PRESSURE`, `MOTION`, `GAS`, `BATTERY`, `LIGHT`.* + +#### Check total stored signal count +```bash +curl -s http://localhost:8080/api/v1/signals/count -H 'X-API-Key: dev-key-1' +``` + +#### Submit a reconciliation batch (ClickHouse Client V2) +```bash +curl -i -X POST http://localhost:8080/api/v1/reconciliation \ + -H 'Content-Type: application/json' \ + -H 'X-API-Key: dev-key-1' \ + -d '[ + {"deviceId":"sensor-1","locationId":"11111111-1111-1111-1111-111111111111","signalType":"TEMPERATURE","value":21.5,"unit":"C"}, + {"deviceId":"sensor-2","locationId":"11111111-1111-1111-1111-111111111111","signalType":"HUMIDITY","value":48.0,"unit":"%"} + ]' +``` + +#### Query 1-second location aggregation slices +```bash +# Query all locations over the last 10 minutes +curl -s 'http://localhost:8080/api/v1/signals/slices?lookback=10m' \ + -H 'X-API-Key: dev-key-1' + +# Query a specific location over the last hour +curl -s 'http://localhost:8080/api/v1/signals/slices?lookback=1h&locationId=11111111-1111-1111-1111-111111111111' \ + -H 'X-API-Key: dev-key-1' +``` +*Lookback format: `` where unit is `s` (seconds), `m` (minutes), or `h` (hours). Maximum lookback is `1h`.* + +--- + +## Traffic Generators (`scripts/` Folder) + +The `scripts/` folder contains standalone Python scripts (using standard library only) to generate realistic traffic, simulate edge cases, and run load tests. + +### 1. HTTP Ingest Traffic (`signal-generator.py`) +Generates single-signal POST requests with optional authentication and payload errors. + +```bash +# Generate mixed valid and invalid traffic for 60 seconds +python3 scripts/signal-generator.py --rate 5 --duration 60 + +# Available profiles: mixed (default), valid, auth, payload, storage +python3 scripts/signal-generator.py --profile valid --rate 20 --duration 60 --quiet +``` + +### 2. Batch Reconciliation Generator (`reconciliation-generator.py`) +Sends variable-size reconciliation batches to test async ClickHouse Client V2 batch writes. + +```bash +python3 scripts/reconciliation-generator.py --rate 2 --duration 60 --min-batch 1 --max-batch 50 +``` + +### 3. Slice Query Generator (`slice-query-generator.py`) +Continuously queries the 1-second slice endpoint to generate read load and trace spans. + +```bash +python3 scripts/slice-query-generator.py --rate 5 --duration 60 --lookbacks 10s,1m,10m,1h +``` + +> **Note**: Omit `--duration` on any script to run continuously until stopped with `Ctrl+C`. + +--- + +## Architecture & Internals + +### Data Flow Overview + +``` +IoT Device / Client + │ + ├── POST /api/v1/signals ───────▶ ApiKeyAuthFilter ──▶ SignalController ──────▶ Spring Data JPA / JDBC ──▶ ClickHouse (iot_signals) + │ │ + ├── POST /api/v1/reconciliation ────┼─────────────────▶ ReconciliationController ──▶ @Async Client V2 ──────▶ ClickHouse (iot_signals) + │ │ + └── GET /api/v1/signals/slices ─────┴─────────────────▶ SignalSliceController ─────▶ Client V2 Query ────────▶ ClickHouse (iot_signals) + │ + ▼ + OpenTelemetry SDK + │ + ├─ Metrics ──────────▶ ClickHouse (otel_metrics) + └─ Metrics & Spans ──▶ Grafana LGTM (Prometheus, Tempo, Grafana) +``` + +### Component Reference + +| Layer / Concern | File / Class | Description | +| :--- | :--- | :--- | +| **HTTP Ingest API** | `web/SignalController` | Serves `POST /api/v1/signals` via Spring Data JPA repository | +| **Async Reconciliation** | `web/ReconciliationController` + `service/ReconciliationService` | Async POJO batch inserts using ClickHouse Client V2 | +| **Slice Aggregations** | `web/SignalSliceController` + `service/SignalSliceService` | Direct SQL analytical queries via Client V2 | +| **Authentication** | `web/ApiKeyAuthFilter` | Intercepts HTTP requests and validates `X-API-Key` | +| **Schema Initialization** | `schema/ClickHouseSchemaInitializer` | Creates `iot_signals` and `otel_metrics` tables on startup | +| **Telemetry & Metrics** | `telemetry/SignalMetrics` + `config/OpenTelemetryConfig` | Captures OTel metrics & traces across HTTP, JPA, and Client V2 | +| **Metrics Exporter** | `telemetry/ClickHouseMetricExporter` | Flushes OTel metrics directly to ClickHouse | + +### Inspecting Data in ClickHouse + +Connect to the running ClickHouse container (find ID via `docker ps`) to query landed signals and metrics: + +```sql +-- View stored signal breakdown by type +SELECT signal_type, count() FROM iot_signals GROUP BY signal_type; + +-- View exported OpenTelemetry metrics +SELECT name, attributes['signal.type'] AS type, value, time +FROM otel_metrics +WHERE name = 'iot.signals.received' +ORDER BY time DESC; + +-- Total JPA / ClickHouse operations by outcome +SELECT + attributes['storage.operation'] AS operation, + attributes['outcome'] AS outcome, + max(value) AS cumulative_total +FROM otel_metrics +WHERE name = 'iot.storage.operations' +GROUP BY operation, outcome; + +-- Average storage operation duration (ms) +SELECT + attributes['storage.operation'] AS operation, + attributes['outcome'] AS outcome, + maxIf(value, type = 'histogram_sum') / maxIf(value, type = 'histogram_count') AS avg_ms +FROM otel_metrics +WHERE name = 'iot.storage.duration' +GROUP BY operation, outcome; +``` + +### Observability in Grafana + +Open Grafana at **[http://localhost:3000](http://localhost:3000)** (no login required): + +- **Metrics (Prometheus)**: Inspect `iot.*` metrics or view client HTTP connection pool stats: + ```promql + httpcomponents_httpclient_pool_total_connections{httpclient="reconciliation",state="leased"} + ``` +- **Traces (Tempo)**: Filter by `service.name = iot-ingest` to trace incoming HTTP requests down to their ClickHouse database spans. + +### Integration Tests + +Run integration tests using Maven: + +```bash +./mvnw test +``` + +`SignalIngestionTests` spins up a Testcontainers ClickHouse instance to verify API authentication, validation, persistence, and telemetry exports. + +### Configuration Reference + +Configure via environment variables or `src/main/resources/application.yml`: + +| Property / Env Variable | Default | Description | +| :--- | :--- | :--- | +| `CLICKHOUSE_URL` | `jdbc:clickhouse://localhost:8123/default` | ClickHouse JDBC URL | +| `CLICKHOUSE_USER` / `CLICKHOUSE_PASSWORD` | `default` / *(empty)* | Database credentials | +| `IOT_AUTH_API_KEYS` | `dev-key-1,dev-key-2` | Allowed API keys (comma-separated) | +| `iot.telemetry.export-interval` | `15s` | Metric flush interval to ClickHouse | +| `OTEL_EXPORTER_OTLP_ENABLED` | `true` | Enable OTLP telemetry exporter | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | `http://localhost:4317` | OTLP/gRPC collector endpoint | +| `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | `http://localhost:4318/v1/metrics` | OTLP/HTTP metrics endpoint | diff --git a/examples/demo-spring-service/compose.yaml b/examples/demo-spring-service/compose.yaml new file mode 100755 index 000000000..e702d522d --- /dev/null +++ b/examples/demo-spring-service/compose.yaml @@ -0,0 +1,15 @@ +name: iot-observability + +services: + otel-lgtm: + image: grafana/otel-lgtm:latest + restart: unless-stopped + ports: + - "3000:3000" + - "4317:4317" + - "4318:4318" + volumes: + - lgtm-data:/data + +volumes: + lgtm-data: diff --git a/examples/demo-spring-service/pom.xml b/examples/demo-spring-service/pom.xml new file mode 100755 index 000000000..a48254321 --- /dev/null +++ b/examples/demo-spring-service/pom.xml @@ -0,0 +1,173 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 3.3.5 + + + + com.clickhouse.examples + iot-ingest + 0.1.0-SNAPSHOT + iot-ingest + Simple IoT signal ingestion service: authenticate signals, store them in ClickHouse, and export OpenTelemetry metrics to ClickHouse. + + + 17 + 1.43.0 + 2.30.0 + otlp + http://localhost:4318/v1/traces + 0.10.0-rc1-SNAPSHOT + + 1.21.4 + + + + + + io.opentelemetry + opentelemetry-bom + ${opentelemetry.version} + pom + import + + + + + + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-starter-actuator + + + io.micrometer + micrometer-tracing-bridge-otel + + + io.micrometer + micrometer-registry-otlp + + + + + org.springframework.boot + spring-boot-starter-data-jpa + + + com.clickhouse + clickhouse-jdbc + ${clickhouse-jdbc.version} + all + + + com.clickhouse + client-v2 + ${clickhouse-jdbc.version} + + + + + io.opentelemetry + opentelemetry-api + + + io.opentelemetry + opentelemetry-sdk + + + io.opentelemetry + opentelemetry-sdk-metrics + + + io.opentelemetry + opentelemetry-sdk-trace + + + io.opentelemetry + opentelemetry-exporter-otlp + + + + io.opentelemetry.javaagent + opentelemetry-javaagent + ${opentelemetry.javaagent.version} + provided + + + + + org.springframework.boot + spring-boot-testcontainers + test + + + org.testcontainers + clickhouse + test + + + org.testcontainers + junit-jupiter + test + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + ${settings.localRepository}/io/opentelemetry/javaagent/opentelemetry-javaagent/${opentelemetry.javaagent.version}/opentelemetry-javaagent-${opentelemetry.javaagent.version}.jar + + + + io.opentelemetry.javaagent + opentelemetry-javaagent + + + + + -Dotel.service.name=${project.artifactId} + -Dotel.traces.exporter=${opentelemetry.javaagent.traces.exporter} + -Dotel.metrics.exporter=none + -Dotel.logs.exporter=none + -Dotel.exporter.otlp.protocol=http/protobuf + -Dotel.exporter.otlp.traces.endpoint=${opentelemetry.javaagent.traces.endpoint} + -Dotel.instrumentation.clickhouse-client-v2.enabled=false + -Dotel.instrumentation.apache-httpclient.enabled=false + + + + + + diff --git a/examples/demo-spring-service/scripts/reconciliation-generator.py b/examples/demo-spring-service/scripts/reconciliation-generator.py new file mode 100755 index 000000000..e7c0399e9 --- /dev/null +++ b/examples/demo-spring-service/scripts/reconciliation-generator.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +"""Generate variable-size reconciliation batches with occasional invalid requests.""" + +from __future__ import annotations + +import argparse +import json +import random +import signal +import sys +import time +import urllib.error +import urllib.request +import uuid +from collections import Counter +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Optional + + +SIGNAL_VALUES = { + "TEMPERATURE": (lambda: round(random.uniform(-10, 45), 2), "C"), + "HUMIDITY": (lambda: round(random.uniform(10, 95), 2), "%"), + "PRESSURE": (lambda: round(random.uniform(970, 1040), 2), "hPa"), + "MOTION": (lambda: random.choice([0.0, 1.0]), ""), + "GAS": (lambda: round(random.uniform(0, 500), 2), "ppm"), + "BATTERY": (lambda: round(random.uniform(0, 100), 2), "%"), + "LIGHT": (lambda: round(random.uniform(0, 10000), 2), "lux"), +} + +LOCATION_NAMESPACE = uuid.UUID("7c3f6c4e-3cb5-4a57-a93e-8a98cc81a84c") + +ERROR_SCENARIOS = ( + "missing_key", + "invalid_key", + "malformed_json", + "missing_value", + "unknown_type", + "blank_device", + "wrong_content_type", +) + +EXPECTED_STATUS = { + "valid": 202, + "missing_key": 401, + "invalid_key": 401, + "malformed_json": 400, + "missing_value": 400, + "unknown_type": 400, + "blank_device": 400, + "wrong_content_type": 415, +} + + +@dataclass +class RequestSpec: + scenario: str + batch_size: int + body: bytes + headers: dict[str, str] + + +def valid_signal(device_count: int, location_count: int) -> dict[str, object]: + signal_type = random.choice(list(SIGNAL_VALUES)) + value_factory, unit = SIGNAL_VALUES[signal_type] + now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + device_number = random.randint(1, device_count) + location_number = (device_number - 1) % location_count + 1 + return { + "signalId": str(uuid.uuid4()), + "deviceId": f"sensor-{device_number:03d}", + "locationId": str(uuid.uuid5(LOCATION_NAMESPACE, f"location-{location_number:03d}")), + "signalType": signal_type, + "value": value_factory(), + "unit": unit, + "eventTime": now, + } + + +def request_for( + api_key: str, + device_count: int, + location_count: int, + min_batch: int, + max_batch: int, + error_rate: float, +) -> RequestSpec: + batch_size = random.randint(min_batch, max_batch) + batch = [valid_signal(device_count, location_count) for _ in range(batch_size)] + scenario = random.choice(ERROR_SCENARIOS) if random.random() < error_rate else "valid" + headers = {"Content-Type": "application/json", "X-API-Key": api_key} + + if scenario == "missing_key": + headers.pop("X-API-Key") + elif scenario == "invalid_key": + headers["X-API-Key"] = "definitely-invalid-token" + elif scenario == "malformed_json": + return RequestSpec(scenario, batch_size, b'[{"deviceId":"broken"}', headers) + elif scenario == "missing_value": + random.choice(batch).pop("value") + elif scenario == "unknown_type": + random.choice(batch)["signalType"] = "PLASMA" + elif scenario == "blank_device": + random.choice(batch)["deviceId"] = " " + elif scenario == "wrong_content_type": + headers["Content-Type"] = "text/plain" + + return RequestSpec(scenario, batch_size, json.dumps(batch).encode(), headers) + + +def send(base_url: str, spec: RequestSpec, timeout: float) -> tuple[int, float, str]: + request = urllib.request.Request( + f"{base_url.rstrip('/')}/api/v1/reconciliation", + data=spec.body, + headers=spec.headers, + method="POST", + ) + started = time.monotonic() + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + body = response.read(200).decode(errors="replace") + return response.status, (time.monotonic() - started) * 1000, body + except urllib.error.HTTPError as error: + body = error.read(200).decode(errors="replace") + return error.code, (time.monotonic() - started) * 1000, body + except urllib.error.URLError as error: + return 0, (time.monotonic() - started) * 1000, str(error.reason) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default="http://localhost:8080", help="API base URL") + parser.add_argument("--api-key", default="dev-key-1", help="valid API key") + parser.add_argument("--rate", type=float, default=1.0, help="batches per second") + parser.add_argument("--duration", type=float, default=0, help="seconds; 0 runs until Ctrl-C") + parser.add_argument("--min-batch", type=int, default=1, help="minimum signals per batch") + parser.add_argument("--max-batch", type=int, default=25, help="maximum signals per batch") + parser.add_argument( + "--error-rate", + type=float, + default=0.08, + help="fraction of requests made deliberately invalid (0.0-1.0)", + ) + parser.add_argument("--devices", type=int, default=50, help="number of emulated devices") + parser.add_argument("--locations", type=int, default=5, help="number of sensor locations") + parser.add_argument("--timeout", type=float, default=3.0, help="request timeout in seconds") + parser.add_argument("--seed", type=int, help="random seed for repeatable traffic") + parser.add_argument("--quiet", action="store_true", help="only print periodic summaries") + args = parser.parse_args() + + if args.rate <= 0 or args.devices <= 0 or args.locations <= 0 or args.duration < 0: + parser.error("rate, devices, and locations must be positive; duration must be non-negative") + if args.min_batch <= 0 or args.max_batch < args.min_batch: + parser.error("min-batch must be positive and no greater than max-batch") + if not 0 <= args.error_rate <= 1: + parser.error("error-rate must be between 0.0 and 1.0") + return args + + +def main() -> int: + args = parse_args() + random.seed(args.seed) + results: Counter[str] = Counter() + stopped = False + + def stop(_signum: int, _frame: object) -> None: + nonlocal stopped + stopped = True + + signal.signal(signal.SIGINT, stop) + signal.signal(signal.SIGTERM, stop) + + started = time.monotonic() + next_request = started + next_summary = started + 10 + print( + f"Generating reconciliation batches to {args.url} at {args.rate:g} batch/s, " + f"size {args.min_batch}-{args.max_batch}, error rate {args.error_rate:.1%}. " + f"Using {args.devices} devices across {args.locations} locations. Press Ctrl-C to stop." + ) + + while not stopped and (args.duration == 0 or time.monotonic() - started < args.duration): + spec = request_for( + args.api_key, + args.devices, + args.locations, + args.min_batch, + args.max_batch, + args.error_rate, + ) + status, latency_ms, detail = send(args.url, spec, args.timeout) + expected = EXPECTED_STATUS[spec.scenario] + matched = status == expected + results[f"status:{status}"] += 1 + results[f"scenario:{spec.scenario}"] += 1 + results["signals"] += spec.batch_size + results["matched" if matched else "unexpected"] += 1 + + if not args.quiet: + marker = "OK" if matched else "UNEXPECTED" + suffix = "" if matched else f" response={detail!r}" + print( + f"{marker:10} {spec.scenario:20} batch={spec.batch_size:<4} " + f"status={status:<3} expected={expected:<3} " + f"latency={latency_ms:7.1f}ms{suffix}" + ) + + now = time.monotonic() + if now >= next_summary: + requests = results["matched"] + results["unexpected"] + statuses = " ".join( + f"{key.removeprefix('status:')}={value}" + for key, value in sorted(results.items()) + if key.startswith("status:") + ) + print( + f"SUMMARY requests={requests} signals={results['signals']} " + f"unexpected={results['unexpected']} statuses[{statuses}]" + ) + next_summary = now + 10 + + next_request += 1 / args.rate + time.sleep(max(0, next_request - time.monotonic())) + + requests = results["matched"] + results["unexpected"] + print( + f"Stopped after {time.monotonic() - started:.1f}s, " + f"{requests} requests, and {results['signals']} generated signals." + ) + return 1 if results["unexpected"] else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/demo-spring-service/scripts/signal-generator.py b/examples/demo-spring-service/scripts/signal-generator.py new file mode 100755 index 000000000..e5b13c4d0 --- /dev/null +++ b/examples/demo-spring-service/scripts/signal-generator.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +"""Generate realistic and deliberately broken traffic for the IoT ingest API.""" + +from __future__ import annotations + +import argparse +import json +import random +import signal +import sys +import time +import urllib.error +import urllib.request +import uuid +from collections import Counter +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Optional + + +SIGNAL_VALUES = { + "TEMPERATURE": (lambda: round(random.uniform(-10, 45), 2), "C"), + "HUMIDITY": (lambda: round(random.uniform(10, 95), 2), "%"), + "PRESSURE": (lambda: round(random.uniform(970, 1040), 2), "hPa"), + "MOTION": (lambda: random.choice([0.0, 1.0]), ""), + "GAS": (lambda: round(random.uniform(0, 500), 2), "ppm"), + "BATTERY": (lambda: round(random.uniform(0, 100), 2), "%"), + "LIGHT": (lambda: round(random.uniform(0, 10000), 2), "lux"), +} + +LOCATION_NAMESPACE = uuid.UUID("7c3f6c4e-3cb5-4a57-a93e-8a98cc81a84c") + +PROFILES = { + "mixed": { + "valid": 55, + "count": 5, + "missing_key": 8, + "invalid_key": 8, + "malformed_json": 5, + "missing_value": 5, + "unknown_type": 5, + "blank_device": 3, + "wrong_content_type": 3, + "wrong_method": 3, + }, + "valid": {"valid": 100}, + "errors": { + "missing_key": 15, + "invalid_key": 15, + "malformed_json": 15, + "missing_value": 15, + "unknown_type": 15, + "blank_device": 10, + "wrong_content_type": 8, + "wrong_method": 7, + }, + "auth": {"missing_key": 50, "invalid_key": 50}, + "payload": { + "malformed_json": 25, + "missing_value": 25, + "unknown_type": 25, + "blank_device": 25, + }, + "storage": {"valid": 85, "count": 15}, +} + +EXPECTED_STATUS = { + "valid": 202, + "count": 200, + "missing_key": 401, + "invalid_key": 401, + "malformed_json": 400, + "missing_value": 400, + "unknown_type": 400, + "blank_device": 400, + "wrong_content_type": 415, + "wrong_method": 405, +} + + +@dataclass +class RequestSpec: + method: str + path: str + body: Optional[bytes] + headers: dict[str, str] + + +def valid_payload(device_count: int, location_count: int) -> dict[str, object]: + signal_type = random.choice(list(SIGNAL_VALUES)) + value_factory, unit = SIGNAL_VALUES[signal_type] + device_number = random.randint(1, device_count) + location_number = (device_number - 1) % location_count + 1 + return { + "deviceId": f"sensor-{device_number:03d}", + "locationId": str(uuid.uuid5(LOCATION_NAMESPACE, f"location-{location_number:03d}")), + "type": signal_type, + "value": value_factory(), + "unit": unit, + "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + } + + +def request_for( + scenario: str, api_key: str, device_count: int, location_count: int +) -> RequestSpec: + payload = valid_payload(device_count, location_count) + headers = {"Content-Type": "application/json", "X-API-Key": api_key} + method = "POST" + path = "/api/v1/signals" + + if scenario == "count": + return RequestSpec("GET", f"{path}/count", None, {"X-API-Key": api_key}) + if scenario == "missing_key": + headers.pop("X-API-Key") + elif scenario == "invalid_key": + headers["X-API-Key"] = "definitely-invalid-token" + elif scenario == "malformed_json": + return RequestSpec(method, path, b'{"deviceId": "broken"', headers) + elif scenario == "missing_value": + payload.pop("value") + elif scenario == "unknown_type": + payload["type"] = "PLASMA" + elif scenario == "blank_device": + payload["deviceId"] = " " + elif scenario == "wrong_content_type": + headers["Content-Type"] = "text/plain" + elif scenario == "wrong_method": + method = "PUT" + + return RequestSpec(method, path, json.dumps(payload).encode(), headers) + + +def send(base_url: str, spec: RequestSpec, timeout: float) -> tuple[int, float, str]: + request = urllib.request.Request( + f"{base_url.rstrip('/')}{spec.path}", + data=spec.body, + headers=spec.headers, + method=spec.method, + ) + started = time.monotonic() + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + body = response.read(200).decode(errors="replace") + return response.status, (time.monotonic() - started) * 1000, body + except urllib.error.HTTPError as error: + body = error.read(200).decode(errors="replace") + return error.code, (time.monotonic() - started) * 1000, body + except urllib.error.URLError as error: + return 0, (time.monotonic() - started) * 1000, str(error.reason) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default="http://localhost:8080", help="API base URL") + parser.add_argument("--api-key", default="dev-key-1", help="valid API key") + parser.add_argument("--profile", choices=PROFILES, default="mixed") + parser.add_argument("--rate", type=float, default=2.0, help="requests per second") + parser.add_argument("--duration", type=float, default=0, help="seconds; 0 runs until Ctrl-C") + parser.add_argument("--devices", type=int, default=20, help="number of emulated devices") + parser.add_argument("--locations", type=int, default=5, help="number of sensor locations") + parser.add_argument("--timeout", type=float, default=3.0, help="request timeout in seconds") + parser.add_argument("--seed", type=int, help="random seed for repeatable traffic") + parser.add_argument("--quiet", action="store_true", help="only print periodic summaries") + args = parser.parse_args() + if args.rate <= 0 or args.devices <= 0 or args.locations <= 0 or args.duration < 0: + parser.error("rate, devices, and locations must be positive; duration must be non-negative") + return args + + +def main() -> int: + args = parse_args() + random.seed(args.seed) + scenarios = list(PROFILES[args.profile]) + weights = list(PROFILES[args.profile].values()) + results: Counter[str] = Counter() + stopped = False + + def stop(_signum: int, _frame: object) -> None: + nonlocal stopped + stopped = True + + signal.signal(signal.SIGINT, stop) + signal.signal(signal.SIGTERM, stop) + + started = time.monotonic() + next_request = started + next_summary = started + 10 + print( + f"Generating {args.profile} traffic to {args.url} at {args.rate:g} req/s " + f"({args.devices} devices across {args.locations} locations). Press Ctrl-C to stop." + ) + + while not stopped and (args.duration == 0 or time.monotonic() - started < args.duration): + scenario = random.choices(scenarios, weights=weights, k=1)[0] + status, latency_ms, detail = send( + args.url, + request_for(scenario, args.api_key, args.devices, args.locations), + args.timeout, + ) + expected = EXPECTED_STATUS[scenario] + matched = status == expected + results[f"status:{status}"] += 1 + results[f"scenario:{scenario}"] += 1 + results["matched" if matched else "unexpected"] += 1 + + if not args.quiet: + marker = "OK" if matched else "UNEXPECTED" + suffix = "" if matched else f" response={detail!r}" + print( + f"{marker:10} {scenario:20} status={status:<3} " + f"expected={expected:<3} latency={latency_ms:7.1f}ms{suffix}" + ) + + now = time.monotonic() + if now >= next_summary: + total = results["matched"] + results["unexpected"] + statuses = " ".join( + f"{key.removeprefix('status:')}={value}" + for key, value in sorted(results.items()) + if key.startswith("status:") + ) + print(f"SUMMARY total={total} unexpected={results['unexpected']} statuses[{statuses}]") + next_summary = now + 10 + + next_request += 1 / args.rate + time.sleep(max(0, next_request - time.monotonic())) + + total = results["matched"] + results["unexpected"] + print(f"Stopped after {time.monotonic() - started:.1f}s and {total} requests.") + return 1 if results["unexpected"] else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/demo-spring-service/scripts/slice-query-generator.py b/examples/demo-spring-service/scripts/slice-query-generator.py new file mode 100755 index 000000000..17eb01648 --- /dev/null +++ b/examples/demo-spring-service/scripts/slice-query-generator.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +"""Continuously query recent one-second signal slices.""" + +from __future__ import annotations + +import argparse +import json +import random +import signal +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +import uuid +from collections import Counter +from dataclasses import dataclass + + +LOCATION_NAMESPACE = uuid.UUID("7c3f6c4e-3cb5-4a57-a93e-8a98cc81a84c") + + +@dataclass +class QueryResult: + status: int + latency_ms: float + row_count: int + detail: str + + +def location_id(location_number: int) -> str: + return str(uuid.uuid5(LOCATION_NAMESPACE, f"location-{location_number:03d}")) + + +def send( + base_url: str, + api_key: str, + lookback: str, + selected_location: str | None, + timeout: float, +) -> QueryResult: + parameters = {"lookback": lookback} + if selected_location is not None: + parameters["locationId"] = selected_location + url = ( + f"{base_url.rstrip('/')}/api/v1/signals/slices?" + f"{urllib.parse.urlencode(parameters)}" + ) + request = urllib.request.Request(url, headers={"X-API-Key": api_key}, method="GET") + started = time.monotonic() + + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + body = response.read() + latency_ms = (time.monotonic() - started) * 1000 + rows = json.loads(body) + if not isinstance(rows, list): + return QueryResult(response.status, latency_ms, 0, "response is not a JSON array") + return QueryResult(response.status, latency_ms, len(rows), "") + except urllib.error.HTTPError as error: + detail = error.read(200).decode(errors="replace") + return QueryResult(error.code, (time.monotonic() - started) * 1000, 0, detail) + except (urllib.error.URLError, TimeoutError) as error: + detail = str(getattr(error, "reason", error)) + return QueryResult(0, (time.monotonic() - started) * 1000, 0, detail) + except (json.JSONDecodeError, UnicodeDecodeError) as error: + return QueryResult(200, (time.monotonic() - started) * 1000, 0, str(error)) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--url", default="http://localhost:8080", help="API base URL") + parser.add_argument("--api-key", default="dev-key-1", help="valid API key") + parser.add_argument("--rate", type=float, default=2.0, help="queries per second") + parser.add_argument("--duration", type=float, default=60, help="seconds; 0 runs until Ctrl-C") + parser.add_argument( + "--lookbacks", + default="10s,1m,10m,1h", + help="comma-separated lookbacks selected randomly per query", + ) + parser.add_argument( + "--locations", + type=int, + default=5, + help="number of stable generated locations available for filtering", + ) + parser.add_argument( + "--location-filter-rate", + type=float, + default=0.7, + help="fraction of queries filtered to one location (0.0-1.0)", + ) + parser.add_argument("--timeout", type=float, default=3.0, help="request timeout in seconds") + parser.add_argument("--seed", type=int, help="random seed for repeatable queries") + parser.add_argument("--quiet", action="store_true", help="only print periodic summaries") + args = parser.parse_args() + + args.lookbacks = [value.strip() for value in args.lookbacks.split(",") if value.strip()] + if args.rate <= 0 or args.duration < 0 or args.locations <= 0: + parser.error("rate and locations must be positive; duration must be non-negative") + if not args.lookbacks: + parser.error("at least one lookback is required") + if not 0 <= args.location_filter_rate <= 1: + parser.error("location-filter-rate must be between 0.0 and 1.0") + return args + + +def main() -> int: + args = parse_args() + random.seed(args.seed) + results: Counter[str] = Counter() + stopped = False + + def stop(_signum: int, _frame: object) -> None: + nonlocal stopped + stopped = True + + signal.signal(signal.SIGINT, stop) + signal.signal(signal.SIGTERM, stop) + + started = time.monotonic() + next_query = started + next_summary = started + 10 + print( + f"Querying signal slices at {args.rate:g} req/s for " + f"{'unlimited time' if args.duration == 0 else f'{args.duration:g}s'}, " + f"using lookbacks {','.join(args.lookbacks)}. Press Ctrl-C to stop." + ) + + while not stopped and (args.duration == 0 or time.monotonic() - started < args.duration): + lookback = random.choice(args.lookbacks) + selected_location = None + if random.random() < args.location_filter_rate: + selected_location = location_id(random.randint(1, args.locations)) + + result = send(args.url, args.api_key, lookback, selected_location, args.timeout) + matched = result.status == 200 and not result.detail + results["requests"] += 1 + results["rows"] += result.row_count + results["latency_micros"] += round(result.latency_ms * 1000) + results["matched" if matched else "unexpected"] += 1 + results[f"status:{result.status}"] += 1 + + if not args.quiet: + marker = "OK" if matched else "UNEXPECTED" + scope = selected_location or "all" + suffix = "" if matched else f" response={result.detail!r}" + print( + f"{marker:10} lookback={lookback:<4} location={scope:<36} " + f"status={result.status:<3} rows={result.row_count:<5} " + f"latency={result.latency_ms:7.1f}ms{suffix}" + ) + + now = time.monotonic() + if now >= next_summary: + average_ms = results["latency_micros"] / results["requests"] / 1000 + print( + f"SUMMARY requests={results['requests']} rows={results['rows']} " + f"unexpected={results['unexpected']} avg_latency={average_ms:.1f}ms" + ) + next_summary = now + 10 + + next_query += 1 / args.rate + time.sleep(max(0, next_query - time.monotonic())) + + average_ms = ( + results["latency_micros"] / results["requests"] / 1000 + if results["requests"] + else 0 + ) + print( + f"Stopped after {time.monotonic() - started:.1f}s: requests={results['requests']} " + f"rows={results['rows']} unexpected={results['unexpected']} " + f"avg_latency={average_ms:.1f}ms." + ) + return 1 if results["unexpected"] else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/IotIngestApplication.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/IotIngestApplication.java new file mode 100755 index 000000000..e52f7ddb6 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/IotIngestApplication.java @@ -0,0 +1,21 @@ +package com.clickhouse.examples; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.properties.ConfigurationPropertiesScan; + +/** + * Entry point for the IoT signal ingestion service. + * + *

The service exposes an authenticated HTTP endpoint that accepts IoT signals, + * persists them to ClickHouse, and records OpenTelemetry metrics that are themselves + * exported to ClickHouse. + */ +@SpringBootApplication +@ConfigurationPropertiesScan +public class IotIngestApplication { + + public static void main(String[] args) { + SpringApplication.run(IotIngestApplication.class, args); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/AuthProperties.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/AuthProperties.java new file mode 100755 index 000000000..6556ba208 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/AuthProperties.java @@ -0,0 +1,29 @@ +package com.clickhouse.examples.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.util.Set; + +/** + * API-key authentication settings. + * + *

Bound from the {@code iot.auth} prefix. A request is authenticated when it presents + * one of the configured API keys in the {@code X-API-Key} header. + * + * @param headerName header carrying the API key + * @param apiKeys the set of accepted API keys + */ +@ConfigurationProperties(prefix = "iot.auth") +public record AuthProperties(String headerName, Set apiKeys) { + + public AuthProperties { + if (headerName == null || headerName.isBlank()) { + headerName = "X-API-Key"; + } + apiKeys = apiKeys == null ? Set.of() : Set.copyOf(apiKeys); + } + + public boolean isValid(String candidate) { + return candidate != null && apiKeys.contains(candidate); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ClickHouseDialect.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ClickHouseDialect.java new file mode 100755 index 000000000..05d158778 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ClickHouseDialect.java @@ -0,0 +1,18 @@ +package com.clickhouse.examples.config; + +import org.hibernate.dialect.DatabaseVersion; +import org.hibernate.dialect.Dialect; + +/** + * Minimal Hibernate dialect for the JPA operations used by this service. + * + *

ClickHouse has no official Hibernate dialect. The base Hibernate 6 SQL rendering is + * sufficient here because the application only performs inserts and aggregate reads, while + * table creation remains under {@code ClickHouseSchemaInitializer}. + */ +public class ClickHouseDialect extends Dialect { + + public ClickHouseDialect() { + super(DatabaseVersion.make(24, 3)); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/OpenTelemetryConfig.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/OpenTelemetryConfig.java new file mode 100755 index 000000000..6d26f0667 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/OpenTelemetryConfig.java @@ -0,0 +1,107 @@ +package com.clickhouse.examples.config; + + +import com.clickhouse.client.api.observability.MetricsRecorder; +import com.clickhouse.client.api.observability.SpanRecorder; +import com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder; +import com.clickhouse.client.api.observability.otel.OpenTelemetrySpanRecorder; +import com.clickhouse.examples.telemetry.ClickHouseMetricExporter; +import io.micrometer.core.instrument.MeterRegistry; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder; +import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; +import io.opentelemetry.sdk.resources.Resource; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.time.Duration; + +/** + * Builds the application OpenTelemetry SDK for custom metrics. Metrics continue to be stored + * in ClickHouse and, when OTLP export is enabled, are also sent to the configured collector. + * The Java agent owns HTTP and ClickHouse tracing so both spans share one context. + */ +@Configuration +public class OpenTelemetryConfig { + + @Bean + public ClickHouseMetricExporter clickHouseMetricExporter(JdbcTemplate jdbc) { + return new ClickHouseMetricExporter(jdbc); + } + + @Bean(destroyMethod = "close") + public OpenTelemetrySdk openTelemetrySdk( + ClickHouseMetricExporter exporter, + @Value("${iot.telemetry.export-interval:15s}") Duration exportInterval, + @Value("${iot.telemetry.otlp.enabled:true}") boolean otlpEnabled, + @Value("${iot.telemetry.otlp.endpoint:http://localhost:4317}") String otlpEndpoint, + @Value("${spring.application.name:iot-ingest}") String serviceName) { + + Resource resource = Resource.getDefault().merge(Resource.create( + Attributes.of(AttributeKey.stringKey("service.name"), serviceName))); + + SdkMeterProviderBuilder meterProviderBuilder = SdkMeterProvider.builder() + .setResource(resource) + .registerMetricReader(PeriodicMetricReader.builder(exporter) + .setInterval(exportInterval) + .build()); + + if (otlpEnabled) { + OtlpGrpcMetricExporter metricExporter = OtlpGrpcMetricExporter.builder() + .setEndpoint(otlpEndpoint) + .build(); + meterProviderBuilder.registerMetricReader(PeriodicMetricReader.builder(metricExporter) + .setInterval(exportInterval) + .build()); + } + + SdkMeterProvider meterProvider = meterProviderBuilder.build(); + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + .setResource(resource) + .build(); + + return OpenTelemetrySdk.builder() + .setMeterProvider(meterProvider) + .setTracerProvider(tracerProvider) + .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance())) + .build(); + } + + @Bean + public OpenTelemetry openTelemetry(OpenTelemetrySdk sdk) { + return sdk; + } + + /** + * Lets the ClickHouse client report its own query and insert spans. + * + *

The spans must be created through the globally registered instance rather than through the + * SDK above: the agent owns tracing, and the SDK built here has no span processor because it + * only exports metrics. Without an agent the global instance is a no-op and the client records + * nothing. + */ + @Bean + public SpanRecorder clickHouseSpanRecorder() { + return new OpenTelemetrySpanRecorder(GlobalOpenTelemetry.get()); + } + + /** + * Lets the ClickHouse client record operation metrics (duration, serialization, count, retries) + * using the Micrometer MetricsRecorder SPI. + */ + @Bean + public MetricsRecorder clickHouseMetricsRecorder(MeterRegistry meterRegistry) { + return new MicrometerMetricsRecorder(meterRegistry); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ReconciliationConfig.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ReconciliationConfig.java new file mode 100755 index 000000000..47bf6d6e9 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/config/ReconciliationConfig.java @@ -0,0 +1,109 @@ +package com.clickhouse.examples.config; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.observability.MetricsRecorder; +import com.clickhouse.client.api.observability.SpanRecorder; +import com.clickhouse.examples.model.ReconciliationSignal; +import io.opentelemetry.context.Context; +import io.micrometer.core.instrument.MeterRegistry; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.jdbc.JdbcConnectionDetails; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.DependsOn; +import org.springframework.core.task.TaskDecorator; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +import java.net.URI; +import java.util.UUID; +import java.util.concurrent.Executor; + +/** + * Configures the direct ClickHouse client and the Spring-owned reconciliation executor. + */ +@Configuration +@EnableAsync +public class ReconciliationConfig { + + @Bean(destroyMethod = "close") + @DependsOn("clickHouseSchemaInitializer") + public Client clickHouseClient( + JdbcConnectionDetails connectionDetails, + MeterRegistry meterRegistry, + SpanRecorder spanRecorder, + MetricsRecorder metricsRecorder, + @Value("${iot.telemetry.clickhouse-client.metrics-group:reconciliation}") String metricsGroup) { + ClickHouseAddress address = ClickHouseAddress.fromJdbcUrl(connectionDetails.getJdbcUrl()); + + Client client = new Client.Builder() + .addEndpoint(address.endpoint()) + .setDefaultDatabase(address.database()) + .setUsername(connectionDetails.getUsername()) + .setPassword(connectionDetails.getPassword()) + // Spring owns the asynchronous boundary; the client performs work on that thread. + // Running on the caller also lets client spans join the trace already in progress. + .useAsyncRequests(false) + // Publishes Apache HttpClient 5 connection-pool gauges to Spring's registry. + .registerClientMetrics(meterRegistry, metricsGroup) + // Reports a span per query and insert, plus one per HTTP attempt below it. + .setSpanRecorder(spanRecorder) + // Records client operation metrics (duration, serialization, count, retries) via Micrometer. + .setMetricsRecorder(metricsRecorder) + // Puts the identifier on the span as db.query.id, so a trace can be looked up + // in ClickHouse's system.query_log. + .setQueryIdGenerator(() -> UUID.randomUUID().toString()) + .build(); + + try { + client.register(ReconciliationSignal.class, client.getTableSchema("iot_signals")); + return client; + } catch (RuntimeException ex) { + client.close(); + throw ex; + } + } + + @Bean + @Qualifier("reconciliationExecutor") + public Executor reconciliationExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setThreadNamePrefix("reconciliation-"); + executor.setCorePoolSize(2); + executor.setMaxPoolSize(4); + executor.setQueueCapacity(100); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setTaskDecorator(openTelemetryContext()); + return executor; + } + + private TaskDecorator openTelemetryContext() { + return task -> Context.current().wrap(task); + } + + private record ClickHouseAddress(String endpoint, String database) { + + private static ClickHouseAddress fromJdbcUrl(String jdbcUrl) { + String uriValue; + if (jdbcUrl.startsWith("jdbc:clickhouse:http://") + || jdbcUrl.startsWith("jdbc:clickhouse:https://")) { + uriValue = jdbcUrl.substring("jdbc:clickhouse:".length()); + } else if (jdbcUrl.startsWith("jdbc:clickhouse://")) { + uriValue = "http:" + jdbcUrl.substring("jdbc:clickhouse:".length()); + } else if (jdbcUrl.startsWith("jdbc:ch://")) { + uriValue = "http:" + jdbcUrl.substring("jdbc:ch:".length()); + } else { + throw new IllegalArgumentException("Unsupported ClickHouse JDBC URL: " + jdbcUrl); + } + + URI uri = URI.create(uriValue); + String path = uri.getPath(); + String database = path == null || path.length() <= 1 + ? "default" + : path.substring(1).split("/", 2)[0]; + String endpoint = uri.getScheme() + "://" + uri.getRawAuthority(); + return new ClickHouseAddress(endpoint, database); + } + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/ReconciliationSignal.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/ReconciliationSignal.java new file mode 100755 index 000000000..7179ace33 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/ReconciliationSignal.java @@ -0,0 +1,135 @@ +package com.clickhouse.examples.model; + +import com.fasterxml.jackson.annotation.JsonFormat; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; + +import java.time.Instant; +import java.util.UUID; + +/** + * JavaBean used by both Jackson request binding and ClickHouse Client V2 POJO serialization. + */ +public class ReconciliationSignal { + + private UUID signalId; + + @NotBlank + private String deviceId; + + @NotNull + private UUID locationId; + + @NotNull + private SignalType signalType; + + @NotNull + private Double value; + + private String unit; + + @JsonFormat(shape = JsonFormat.Shape.STRING) + private Instant eventTime; + + @JsonFormat(shape = JsonFormat.Shape.STRING) + private Instant receivedAt; + + public ReconciliationSignal() { + // Required by Jackson and the ClickHouse POJO serializer. + } + + public ReconciliationSignal( + UUID signalId, + String deviceId, + UUID locationId, + SignalType signalType, + Double value, + String unit, + Instant eventTime, + Instant receivedAt) { + this.signalId = signalId; + this.deviceId = deviceId; + this.locationId = locationId; + this.signalType = signalType; + this.value = value; + this.unit = unit; + this.eventTime = eventTime; + this.receivedAt = receivedAt; + } + + public ReconciliationSignal withDefaults(Instant fallbackTime) { + return new ReconciliationSignal( + signalId == null ? UUID.randomUUID() : signalId, + deviceId, + locationId, + signalType, + value, + unit == null ? "" : unit, + eventTime == null ? fallbackTime : eventTime, + receivedAt == null ? fallbackTime : receivedAt); + } + + public UUID getSignalId() { + return signalId; + } + + public void setSignalId(UUID signalId) { + this.signalId = signalId; + } + + public String getDeviceId() { + return deviceId; + } + + public void setDeviceId(String deviceId) { + this.deviceId = deviceId; + } + + public UUID getLocationId() { + return locationId; + } + + public void setLocationId(UUID locationId) { + this.locationId = locationId; + } + + public SignalType getSignalType() { + return signalType; + } + + public void setSignalType(SignalType signalType) { + this.signalType = signalType; + } + + public Double getValue() { + return value; + } + + public void setValue(Double value) { + this.value = value; + } + + public String getUnit() { + return unit; + } + + public void setUnit(String unit) { + this.unit = unit; + } + + public Instant getEventTime() { + return eventTime; + } + + public void setEventTime(Instant eventTime) { + this.eventTime = eventTime; + } + + public Instant getReceivedAt() { + return receivedAt; + } + + public void setReceivedAt(Instant receivedAt) { + this.receivedAt = receivedAt; + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/Signal.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/Signal.java new file mode 100755 index 000000000..a9f0ddc1d --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/Signal.java @@ -0,0 +1,32 @@ +package com.clickhouse.examples.model; + +import com.fasterxml.jackson.annotation.JsonFormat; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; + +import java.time.Instant; +import java.util.UUID; + +/** + * A single reading emitted by an IoT device, as received over HTTP. + * + * @param deviceId identifier of the emitting device (required) + * @param locationId identifier of the device location (required) + * @param type kind of measurement (required, must be a known {@link SignalType}) + * @param value numeric reading + * @param unit unit of the reading, e.g. "C", "%", "hPa" (optional) + * @param timestamp when the reading was taken on the device; defaults to ingestion time when absent + */ +public record Signal( + @NotBlank String deviceId, + @NotNull UUID locationId, + @NotNull SignalType type, + @NotNull Double value, + String unit, + @JsonFormat(shape = JsonFormat.Shape.STRING) Instant timestamp +) { + /** Returns a copy with {@code timestamp} filled in from {@code fallback} when the client omitted it. */ + public Signal withTimestampOrDefault(Instant fallback) { + return timestamp != null ? this : new Signal(deviceId, locationId, type, value, unit, fallback); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalEntity.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalEntity.java new file mode 100755 index 000000000..588dcbe07 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalEntity.java @@ -0,0 +1,85 @@ +package com.clickhouse.examples.model; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import org.hibernate.annotations.UuidGenerator; +import org.hibernate.annotations.JdbcTypeCode; +import org.hibernate.type.SqlTypes; + +import java.time.Instant; +import java.util.UUID; + +/** + * JPA persistence model for the {@code iot_signals} ClickHouse table. + * + *

The HTTP model remains separate so persistence details do not leak into the API. + */ +@Entity +@Table(name = "iot_signals") +public class SignalEntity { + + @Id + @GeneratedValue + @UuidGenerator + @JdbcTypeCode(SqlTypes.VARCHAR) + @Column(name = "signal_id", nullable = false, updatable = false) + private UUID id; + + @Column(name = "device_id", nullable = false) + private String deviceId; + + @JdbcTypeCode(SqlTypes.VARCHAR) + @Column(name = "location_id", nullable = false) + private UUID locationId; + + @Enumerated(EnumType.STRING) + @Column(name = "signal_type", nullable = false) + private SignalType type; + + @Column(name = "value", nullable = false) + private double value; + + @Column(name = "unit", nullable = false) + private String unit; + + @Column(name = "event_time", nullable = false) + private Instant eventTime; + + protected SignalEntity() { + // Required by JPA. + } + + private SignalEntity( + String deviceId, + UUID locationId, + SignalType type, + double value, + String unit, + Instant eventTime) { + this.deviceId = deviceId; + this.locationId = locationId; + this.type = type; + this.value = value; + this.unit = unit; + this.eventTime = eventTime; + } + + public static SignalEntity from(Signal signal) { + return new SignalEntity( + signal.deviceId(), + signal.locationId(), + signal.type(), + signal.value(), + signal.unit() == null ? "" : signal.unit(), + signal.timestamp()); + } + + public UUID getId() { + return id; + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalSlice.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalSlice.java new file mode 100755 index 000000000..1311795d1 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalSlice.java @@ -0,0 +1,20 @@ +package com.clickhouse.examples.model; + +import java.time.Instant; +import java.util.UUID; + +/** + * One-second view of mean signal values reported by devices in a location. + */ +public record SignalSlice( + UUID locationId, + Instant timestamp, + Double temperature, + Double humidity, + Double pressure, + Double motion, + Double gas, + Double battery, + Double light +) { +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalType.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalType.java new file mode 100755 index 000000000..5ed40182f --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/model/SignalType.java @@ -0,0 +1,17 @@ +package com.clickhouse.examples.model; + +/** + * The kinds of signal an IoT device may emit. + * + *

Kept as a closed enum so that unknown signal types are rejected at the edge + * and so that per-type metrics have a bounded, low-cardinality set of values. + */ +public enum SignalType { + TEMPERATURE, + HUMIDITY, + PRESSURE, + MOTION, + GAS, + BATTERY, + LIGHT +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/repository/SignalRepository.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/repository/SignalRepository.java new file mode 100755 index 000000000..d8ef5662f --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/repository/SignalRepository.java @@ -0,0 +1,12 @@ +package com.clickhouse.examples.repository; + +import com.clickhouse.examples.model.SignalEntity; +import org.springframework.data.jpa.repository.JpaRepository; + +import java.util.UUID; + +/** + * Spring Data JPA repository for signals stored in ClickHouse. + */ +public interface SignalRepository extends JpaRepository { +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/schema/ClickHouseSchemaInitializer.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/schema/ClickHouseSchemaInitializer.java new file mode 100755 index 000000000..fd1c38a61 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/schema/ClickHouseSchemaInitializer.java @@ -0,0 +1,66 @@ +package com.clickhouse.examples.schema; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +/** + * Creates the ClickHouse tables the service needs, if they do not already exist. + * + *

Runs during bean initialization, before the application starts accepting traffic, so + * the first signal insert and metric export both find their tables in place. + */ +@Component +public class ClickHouseSchemaInitializer implements InitializingBean { + + private static final Logger log = LoggerFactory.getLogger(ClickHouseSchemaInitializer.class); + + private static final String SIGNALS_DDL = """ + CREATE TABLE IF NOT EXISTS iot_signals ( + signal_id UUID DEFAULT generateUUIDv4(), + device_id String, + location_id UUID, + signal_type LowCardinality(String), + value Float64, + unit String, + event_time DateTime64(3), + received_at DateTime64(3) DEFAULT now64(3) + ) ENGINE = MergeTree + ORDER BY (location_id, event_time, signal_type) + """; + + private static final String LOCATION_MIGRATION_DDL = """ + ALTER TABLE iot_signals + ADD COLUMN IF NOT EXISTS location_id UUID AFTER device_id + """; + + private static final String METRICS_DDL = """ + CREATE TABLE IF NOT EXISTS otel_metrics ( + name String, + description String, + unit String, + type LowCardinality(String), + value Float64, + attributes Map(String, String), + start_time DateTime64(9), + time DateTime64(9) + ) ENGINE = MergeTree + ORDER BY (name, time) + """; + + private final JdbcTemplate jdbc; + + public ClickHouseSchemaInitializer(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + @Override + public void afterPropertiesSet() { + jdbc.execute(SIGNALS_DDL); + jdbc.execute(LOCATION_MIGRATION_DDL); + jdbc.execute(METRICS_DDL); + log.info("ClickHouse schema ready (tables: iot_signals, otel_metrics)"); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/ReconciliationService.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/ReconciliationService.java new file mode 100755 index 000000000..b2f56cd17 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/ReconciliationService.java @@ -0,0 +1,53 @@ +package com.clickhouse.examples.service; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.insert.InsertResponse; +import com.clickhouse.examples.model.ReconciliationSignal; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Service; + +import java.util.List; +import java.util.concurrent.ExecutionException; + +/** + * Persists reconciliation batches through the direct ClickHouse Client V2 API. + */ +@Service +public class ReconciliationService { + + private static final Logger log = LoggerFactory.getLogger(ReconciliationService.class); + + private final Client client; + + public ReconciliationService(Client client) { + this.client = client; + } + + /** + * Runs after the controller has handed the batch to Spring's task executor. + * + *

The captured request context is restored on the worker, so the span the client reports for + * the insert remains a child of the HTTP span even though the response is already complete. + */ + @Async("reconciliationExecutor") + public void reconcile(List signals, Context requestContext) { + try (Scope ignored = requestContext.makeCurrent()) { + try (InsertResponse response = client.insert("iot_signals", signals).get()) { + log.debug( + "Reconciled batch size={} writtenRows={} queryId={}", + signals.size(), + response.getWrittenRows(), + response.getQueryId()); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + log.error("Reconciliation interrupted for batch size={}", signals.size(), ex); + } catch (ExecutionException | RuntimeException ex) { + log.error("Reconciliation failed for batch size={}", signals.size(), ex); + } + } + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/SignalSliceService.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/SignalSliceService.java new file mode 100755 index 000000000..d32062f21 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/service/SignalSliceService.java @@ -0,0 +1,84 @@ +package com.clickhouse.examples.service; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.query.GenericRecord; +import com.clickhouse.examples.model.SignalSlice; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** + * Builds a row-oriented, one-second view of signal means by location. + */ +@Service +public class SignalSliceService { + + private static final String SELECT_SLICES = """ + SELECT + location_id, + toStartOfSecond(event_time) AS window_start, + if(countIf(signal_type = 'TEMPERATURE') = 0, NULL, + avgIf(value, signal_type = 'TEMPERATURE')) AS temperature, + if(countIf(signal_type = 'HUMIDITY') = 0, NULL, + avgIf(value, signal_type = 'HUMIDITY')) AS humidity, + if(countIf(signal_type = 'PRESSURE') = 0, NULL, + avgIf(value, signal_type = 'PRESSURE')) AS pressure, + if(countIf(signal_type = 'MOTION') = 0, NULL, + avgIf(value, signal_type = 'MOTION')) AS motion, + if(countIf(signal_type = 'GAS') = 0, NULL, + avgIf(value, signal_type = 'GAS')) AS gas, + if(countIf(signal_type = 'BATTERY') = 0, NULL, + avgIf(value, signal_type = 'BATTERY')) AS battery, + if(countIf(signal_type = 'LIGHT') = 0, NULL, + avgIf(value, signal_type = 'LIGHT')) AS light + FROM iot_signals + WHERE event_time >= now64(3) - toIntervalMillisecond({lookbackMillis:UInt64}) + AND event_time < now64(3) + %s + GROUP BY location_id, window_start + ORDER BY window_start DESC, location_id + """; + + private final Client client; + + public SignalSliceService(Client client) { + this.client = client; + } + + public List findSlices(Duration lookback, UUID locationId) { + Map parameters; + String locationClause = ""; + if (locationId != null) { + locationClause = "AND location_id = {locationId:UUID}"; + parameters = Map.of( + "lookbackMillis", lookback.toMillis(), + "locationId", locationId); + } else { + parameters = Map.of("lookbackMillis", lookback.toMillis()); + } + + return client.queryAll(SELECT_SLICES.formatted(locationClause), parameters).stream() + .map(this::toSignalSlice) + .toList(); + } + + private SignalSlice toSignalSlice(GenericRecord record) { + return new SignalSlice( + record.getUUID("location_id"), + record.getInstant("window_start"), + nullableDouble(record, "temperature"), + nullableDouble(record, "humidity"), + nullableDouble(record, "pressure"), + nullableDouble(record, "motion"), + nullableDouble(record, "gas"), + nullableDouble(record, "battery"), + nullableDouble(record, "light")); + } + + private Double nullableDouble(GenericRecord record, String column) { + return record.hasValue(column) ? record.getDouble(column) : null; + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/ClickHouseMetricExporter.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/ClickHouseMetricExporter.java new file mode 100755 index 000000000..3f5d5b326 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/ClickHouseMetricExporter.java @@ -0,0 +1,122 @@ +package com.clickhouse.examples.telemetry; + +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.metrics.InstrumentType; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; +import io.opentelemetry.sdk.metrics.data.DoublePointData; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; +import io.opentelemetry.sdk.metrics.data.LongPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.PointData; +import io.opentelemetry.sdk.metrics.export.MetricExporter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.time.Instant; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * An OpenTelemetry {@link MetricExporter} that writes metric points into a ClickHouse table. + * + *

Each exported point becomes one row in {@code otel_metrics}, keeping the metric name, + * its attributes (e.g. {@code signal.type}) and the aggregated value. This lets the same + * ClickHouse instance that stores raw signals also serve as the metrics backend, without + * requiring a separate OpenTelemetry Collector. + */ +public class ClickHouseMetricExporter implements MetricExporter { + + private static final Logger log = LoggerFactory.getLogger(ClickHouseMetricExporter.class); + + private static final String INSERT_SQL = """ + INSERT INTO otel_metrics + (name, description, unit, type, value, attributes, start_time, time) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """; + + private final JdbcTemplate jdbc; + + public ClickHouseMetricExporter(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + /** Counters report cumulative totals, which is what we want to persist as a time series. */ + @Override + public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { + return AggregationTemporality.CUMULATIVE; + } + + @Override + public CompletableResultCode export(Collection metrics) { + try { + for (MetricData metric : metrics) { + switch (metric.getType()) { + case LONG_SUM -> exportLong(metric, metric.getLongSumData().getPoints(), "sum"); + case DOUBLE_SUM -> exportDouble(metric, metric.getDoubleSumData().getPoints(), "sum"); + case LONG_GAUGE -> exportLong(metric, metric.getLongGaugeData().getPoints(), "gauge"); + case DOUBLE_GAUGE -> exportDouble(metric, metric.getDoubleGaugeData().getPoints(), "gauge"); + case HISTOGRAM -> exportHistogram(metric, metric.getHistogramData().getPoints()); + default -> log.debug("Skipping unsupported metric type {} for {}", metric.getType(), metric.getName()); + } + } + return CompletableResultCode.ofSuccess(); + } catch (RuntimeException e) { + log.warn("Failed to export metrics to ClickHouse", e); + return CompletableResultCode.ofFailure(); + } + } + + private void exportLong(MetricData metric, Collection points, String type) { + for (LongPointData p : points) { + insert(metric, type, (double) p.getValue(), p); + } + } + + private void exportDouble(MetricData metric, Collection points, String type) { + for (DoublePointData p : points) { + insert(metric, type, p.getValue(), p); + } + } + + private void exportHistogram(MetricData metric, Collection points) { + for (HistogramPointData p : points) { + insert(metric, "histogram_count", p.getCount(), p); + insert(metric, "histogram_sum", p.getSum(), p); + } + } + + private void insert(MetricData metric, String type, double value, PointData point) { + jdbc.update(INSERT_SQL, + metric.getName(), + metric.getDescription(), + metric.getUnit(), + type, + value, + attributesOf(point), + nanosToInstant(point.getStartEpochNanos()), + nanosToInstant(point.getEpochNanos())); + } + + /** Point attributes as a plain {@link Map}; the ClickHouse driver maps this to Map(String,String). */ + private Map attributesOf(PointData point) { + Map attrs = new LinkedHashMap<>(); + point.getAttributes().forEach((key, val) -> attrs.put(key.getKey(), String.valueOf(val))); + return attrs; + } + + private static Instant nanosToInstant(long epochNanos) { + return Instant.ofEpochSecond(epochNanos / 1_000_000_000L, epochNanos % 1_000_000_000L); + } + + @Override + public CompletableResultCode flush() { + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + return CompletableResultCode.ofSuccess(); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/SignalMetrics.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/SignalMetrics.java new file mode 100755 index 000000000..1e27bfe98 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/telemetry/SignalMetrics.java @@ -0,0 +1,79 @@ +package com.clickhouse.examples.telemetry; + +import com.clickhouse.examples.model.SignalType; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.Meter; +import org.springframework.stereotype.Component; + +import java.time.Duration; + +/** + * Application metrics recorded through the OpenTelemetry API. + * + *

These instruments feed the configured metric reader, which exports the aggregated + * values to ClickHouse (see {@link ClickHouseMetricExporter}). The primary metric is the + * number of signals received, broken down by {@code signal.type}. + */ +@Component +public class SignalMetrics { + + private static final AttributeKey SIGNAL_TYPE = AttributeKey.stringKey("signal.type"); + private static final AttributeKey STORAGE_OPERATION = AttributeKey.stringKey("storage.operation"); + private static final AttributeKey OUTCOME = AttributeKey.stringKey("outcome"); + + private final LongCounter signalsReceived; + private final LongCounter signalsRejected; + private final LongCounter authFailures; + private final LongCounter storageOperations; + private final DoubleHistogram storageDuration; + + public SignalMetrics(OpenTelemetry openTelemetry) { + Meter meter = openTelemetry.getMeter("com.clickhouse.examples"); + + this.signalsReceived = meter.counterBuilder("iot.signals.received") + .setDescription("Number of IoT signals accepted and stored, by signal type") + .setUnit("{signal}") + .build(); + this.signalsRejected = meter.counterBuilder("iot.signals.rejected") + .setDescription("Number of IoT signals rejected as invalid, by signal type") + .setUnit("{signal}") + .build(); + this.authFailures = meter.counterBuilder("iot.auth.failures") + .setDescription("Number of requests rejected due to a missing or invalid API key") + .setUnit("{request}") + .build(); + this.storageOperations = meter.counterBuilder("iot.storage.operations") + .setDescription("Number of JPA/ClickHouse operations by operation and outcome") + .setUnit("{operation}") + .build(); + this.storageDuration = meter.histogramBuilder("iot.storage.duration") + .setDescription("JPA/ClickHouse operation duration by operation and outcome") + .setUnit("ms") + .build(); + } + + public void recordReceived(SignalType type) { + signalsReceived.add(1, Attributes.of(SIGNAL_TYPE, type.name())); + } + + public void recordRejected(SignalType type) { + String label = type == null ? "UNKNOWN" : type.name(); + signalsRejected.add(1, Attributes.of(SIGNAL_TYPE, label)); + } + + public void recordAuthFailure() { + authFailures.add(1); + } + + public void recordStorageOperation(String operation, boolean success, Duration duration) { + Attributes attributes = Attributes.of( + STORAGE_OPERATION, operation, + OUTCOME, success ? "success" : "failure"); + storageOperations.add(1, attributes); + storageDuration.record(duration.toNanos() / 1_000_000.0, attributes); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ApiKeyAuthFilter.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ApiKeyAuthFilter.java new file mode 100755 index 000000000..8f65c8958 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ApiKeyAuthFilter.java @@ -0,0 +1,55 @@ +package com.clickhouse.examples.web; + +import com.clickhouse.examples.config.AuthProperties; +import com.clickhouse.examples.telemetry.SignalMetrics; +import jakarta.servlet.FilterChain; +import jakarta.servlet.ServletException; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import org.springframework.core.annotation.Order; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.stereotype.Component; +import org.springframework.web.filter.OncePerRequestFilter; + +import java.io.IOException; + +/** + * Rejects requests to protected endpoints that do not carry a valid API key. + * + *

Only paths under {@code /api/} are guarded; actuator/health style endpoints and the + * root are left open. Rejected requests are counted as an OpenTelemetry metric. + */ +@Component +@Order(1) +public class ApiKeyAuthFilter extends OncePerRequestFilter { + + private final AuthProperties auth; + private final SignalMetrics metrics; + + public ApiKeyAuthFilter(AuthProperties auth, SignalMetrics metrics) { + this.auth = auth; + this.metrics = metrics; + } + + @Override + protected boolean shouldNotFilter(HttpServletRequest request) { + return !request.getRequestURI().startsWith("/api/"); + } + + @Override + protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain chain) + throws ServletException, IOException { + + String presented = request.getHeader(auth.headerName()); + if (!auth.isValid(presented)) { + metrics.recordAuthFailure(); + response.setStatus(HttpServletResponse.SC_UNAUTHORIZED); + response.setContentType(MediaType.APPLICATION_JSON_VALUE); + response.setHeader(HttpHeaders.WWW_AUTHENTICATE, "ApiKey header=\"" + auth.headerName() + "\""); + response.getWriter().write("{\"error\":\"invalid or missing API key\"}"); + return; + } + chain.doFilter(request, response); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/GlobalExceptionHandler.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/GlobalExceptionHandler.java new file mode 100755 index 000000000..28e8127e5 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/GlobalExceptionHandler.java @@ -0,0 +1,68 @@ +package com.clickhouse.examples.web; + +import com.clickhouse.examples.telemetry.SignalMetrics; +import org.springframework.dao.DataAccessException; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.http.converter.HttpMessageNotReadableException; +import org.springframework.web.HttpMediaTypeNotSupportedException; +import org.springframework.web.HttpRequestMethodNotSupportedException; +import org.springframework.web.bind.MethodArgumentNotValidException; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.RestControllerAdvice; + +import java.util.Map; + +/** + * Turns malformed or invalid signal payloads into 400 responses and counts them as rejected. + */ +@RestControllerAdvice +public class GlobalExceptionHandler { + + private final SignalMetrics metrics; + + public GlobalExceptionHandler(SignalMetrics metrics) { + this.metrics = metrics; + } + + @ExceptionHandler(MethodArgumentNotValidException.class) + public ResponseEntity> onValidation(MethodArgumentNotValidException ex) { + metrics.recordRejected(null); + String detail = ex.getBindingResult().getFieldErrors().stream() + .findFirst() + .map(e -> e.getField() + " " + e.getDefaultMessage()) + .orElse("invalid payload"); + return badRequest(detail); + } + + @ExceptionHandler(HttpMessageNotReadableException.class) + public ResponseEntity> onUnreadable(HttpMessageNotReadableException ex) { + metrics.recordRejected(null); + return badRequest("malformed or unrecognized signal payload"); + } + + @ExceptionHandler(HttpMediaTypeNotSupportedException.class) + public ResponseEntity> onUnsupportedMediaType(HttpMediaTypeNotSupportedException ex) { + metrics.recordRejected(null); + return ResponseEntity.status(HttpStatus.UNSUPPORTED_MEDIA_TYPE) + .body(Map.of("error", "unsupported media type")); + } + + @ExceptionHandler(HttpRequestMethodNotSupportedException.class) + public ResponseEntity> onUnsupportedMethod(HttpRequestMethodNotSupportedException ex) { + metrics.recordRejected(null); + return ResponseEntity.status(HttpStatus.METHOD_NOT_ALLOWED) + .body(Map.of("error", "method not allowed")); + } + + @ExceptionHandler(DataAccessException.class) + public ResponseEntity> onStorageFailure(DataAccessException ex) { + return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE) + .body(Map.of("error", "signal storage unavailable")); + } + + private ResponseEntity> badRequest(String detail) { + return ResponseEntity.status(HttpStatus.BAD_REQUEST) + .body(Map.of("error", "bad request", "detail", detail)); + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ReconciliationController.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ReconciliationController.java new file mode 100755 index 000000000..9b39a30e9 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/ReconciliationController.java @@ -0,0 +1,49 @@ +package com.clickhouse.examples.web; + +import com.clickhouse.examples.model.ReconciliationSignal; +import com.clickhouse.examples.service.ReconciliationService; +import io.opentelemetry.context.Context; +import jakarta.validation.Valid; +import jakarta.validation.constraints.NotEmpty; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.context.request.async.DeferredResult; + +import java.time.Instant; +import java.util.List; +import java.util.Map; + +/** + * Accepts batches for fire-and-forget reconciliation through the direct ClickHouse client. + */ +@RestController +@RequestMapping("/api/v1/reconciliation") +public class ReconciliationController { + + private final ReconciliationService reconciliationService; + + public ReconciliationController(ReconciliationService reconciliationService) { + this.reconciliationService = reconciliationService; + } + + @PostMapping + public DeferredResult>> reconcile( + @Valid @NotEmpty @RequestBody List<@Valid ReconciliationSignal> signals) { + Instant receivedAt = Instant.now(); + List batch = signals.stream() + .map(signal -> signal.withDefaults(receivedAt)) + .toList(); + Context requestContext = Context.current(); + + DeferredResult>> result = new DeferredResult<>(); + result.onCompletion(() -> reconciliationService.reconcile(batch, requestContext)); + result.setResult(ResponseEntity.status(HttpStatus.ACCEPTED) + .body(Map.of("status", "accepted", "count", batch.size()))); + + return result; + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalController.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalController.java new file mode 100755 index 000000000..cf7771fbc --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalController.java @@ -0,0 +1,77 @@ +package com.clickhouse.examples.web; + +import com.clickhouse.examples.model.Signal; +import com.clickhouse.examples.model.SignalEntity; +import com.clickhouse.examples.repository.SignalRepository; +import com.clickhouse.examples.telemetry.SignalMetrics; +import jakarta.validation.Valid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import java.time.Clock; +import java.time.Duration; +import java.util.Map; +import java.util.function.Supplier; + +/** + * Ingestion endpoint for IoT signals. All routes here are guarded by {@link ApiKeyAuthFilter}. + */ +@RestController +@RequestMapping("/api/v1/signals") +public class SignalController { + + private static final Logger log = LoggerFactory.getLogger(SignalController.class); + + private final SignalRepository repository; + private final SignalMetrics metrics; + private final Clock clock; + + @Autowired + public SignalController(SignalRepository repository, SignalMetrics metrics) { + this(repository, metrics, Clock.systemUTC()); + } + + SignalController(SignalRepository repository, SignalMetrics metrics, Clock clock) { + this.repository = repository; + this.metrics = metrics; + this.clock = clock; + } + + @PostMapping + public ResponseEntity> ingest(@Valid @RequestBody Signal signal) { + Signal stored = signal.withTimestampOrDefault(clock.instant()); + measureStorage("insert", () -> repository.save(SignalEntity.from(stored))); + metrics.recordReceived(stored.type()); + log.debug("Stored signal type={} device={}", stored.type(), stored.deviceId()); + return ResponseEntity.status(HttpStatus.ACCEPTED) + .body(Map.of("status", "accepted", "type", stored.type().name())); + } + + @GetMapping("/count") + public Map count() { + return Map.of("count", measureStorage("count", repository::count)); + } + + private T measureStorage(String operationName, Supplier operation) { + long startNanos = System.nanoTime(); + boolean success = false; + try { + T result = operation.get(); + success = true; + return result; + } finally { + metrics.recordStorageOperation( + operationName, + success, + Duration.ofNanos(System.nanoTime() - startNanos)); + } + } +} diff --git a/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalSliceController.java b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalSliceController.java new file mode 100755 index 000000000..cdd1ebee9 --- /dev/null +++ b/examples/demo-spring-service/src/main/java/com/clickhouse/examples/web/SignalSliceController.java @@ -0,0 +1,71 @@ +package com.clickhouse.examples.web; + +import com.clickhouse.examples.model.SignalSlice; +import com.clickhouse.examples.service.SignalSliceService; +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.server.ResponseStatusException; + +import java.time.Duration; +import java.util.List; +import java.util.UUID; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Returns one-second mean-value rows for recently reported location signals. + */ +@RestController +@RequestMapping("/api/v1/signals/slices") +public class SignalSliceController { + + private static final Duration MAX_LOOKBACK = Duration.ofHours(1); + private static final Pattern LOOKBACK_PATTERN = Pattern.compile("([1-9][0-9]*)(s|m|h)"); + + private final SignalSliceService signalSliceService; + + public SignalSliceController(SignalSliceService signalSliceService) { + this.signalSliceService = signalSliceService; + } + + @GetMapping + public List slices( + @RequestParam(defaultValue = "10s") String lookback, + @RequestParam(required = false) UUID locationId) { + return signalSliceService.findSlices(parseLookback(lookback), locationId); + } + + private Duration parseLookback(String value) { + Matcher matcher = LOOKBACK_PATTERN.matcher(value); + if (!matcher.matches()) { + throw invalidLookback(); + } + + long amount; + try { + amount = Long.parseLong(matcher.group(1)); + } catch (NumberFormatException ex) { + throw invalidLookback(); + } + + Duration duration = switch (matcher.group(2)) { + case "s" -> Duration.ofSeconds(amount); + case "m" -> Duration.ofMinutes(amount); + case "h" -> Duration.ofHours(amount); + default -> throw invalidLookback(); + }; + if (duration.compareTo(MAX_LOOKBACK) > 0) { + throw invalidLookback(); + } + return duration; + } + + private ResponseStatusException invalidLookback() { + return new ResponseStatusException( + HttpStatus.BAD_REQUEST, + "lookback must use s, m, or h and cannot exceed 1h"); + } +} diff --git a/examples/demo-spring-service/src/main/resources/application.yml b/examples/demo-spring-service/src/main/resources/application.yml new file mode 100755 index 000000000..4453d6e39 --- /dev/null +++ b/examples/demo-spring-service/src/main/resources/application.yml @@ -0,0 +1,63 @@ +spring: + application: + name: iot-ingest + datasource: + # Used when running against a real/standalone ClickHouse. During local development and + # tests these values are overridden by the Testcontainers-managed instance (@ServiceConnection). + url: ${CLICKHOUSE_URL:jdbc:clickhouse://localhost:8123/default} + username: ${CLICKHOUSE_USER:default} + password: ${CLICKHOUSE_PASSWORD:} + driver-class-name: com.clickhouse.jdbc.ClickHouseDriver + hikari: + connection-test-query: SELECT 1 + maximum-pool-size: 8 + data-source-properties: + # Hibernate calls optional JDBC transaction/LOB methods that ClickHouse does not + # implement. The driver safely treats them as no-ops in this compatibility mode. + jdbc_ignore_unsupported_values: true + jpa: + database-platform: com.clickhouse.examples.config.ClickHouseDialect + open-in-view: false + hibernate: + # ClickHouse-specific MergeTree DDL is managed by ClickHouseSchemaInitializer. + ddl-auto: none + properties: + hibernate: + jdbc: + time_zone: UTC + +server: + port: 8080 + +management: + tracing: + # The Java agent owns HTTP and database spans so they use one SDK/context. + # Application metrics still use the SDK in OpenTelemetryConfig. + enabled: false + otlp: + metrics: + export: + # Micrometer uses OTLP/HTTP; the existing application SDK uses OTLP/gRPC. + enabled: ${OTEL_EXPORTER_OTLP_ENABLED:true} + url: ${OTEL_EXPORTER_OTLP_METRICS_ENDPOINT:http://localhost:4318/v1/metrics} + step: ${OTEL_METRIC_EXPORT_INTERVAL:15s} + +iot: + auth: + header-name: X-API-Key + # Demo keys. Override in real deployments via IOT_AUTH_API_KEYS (comma-separated). + api-keys: ${IOT_AUTH_API_KEYS:dev-key-1,dev-key-2} + telemetry: + # How often aggregated OpenTelemetry metrics are flushed. + export-interval: 15s + clickhouse-client: + # Becomes the "httpclient" tag on Apache HttpClient connection-pool gauges. + metrics-group: reconciliation + otlp: + # The local Grafana LGTM stack accepts OTLP/gRPC on this endpoint. + enabled: ${OTEL_EXPORTER_OTLP_ENABLED:true} + endpoint: ${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4317} + +logging: + level: + com.clickhouse.examples: DEBUG diff --git a/examples/demo-spring-service/src/test/java/com/clickhouse/examples/SignalIngestionTests.java b/examples/demo-spring-service/src/test/java/com/clickhouse/examples/SignalIngestionTests.java new file mode 100755 index 000000000..fc1b15600 --- /dev/null +++ b/examples/demo-spring-service/src/test/java/com/clickhouse/examples/SignalIngestionTests.java @@ -0,0 +1,341 @@ +package com.clickhouse.examples; + +import com.clickhouse.client.api.observability.MetricsRecorder; +import com.clickhouse.client.api.observability.micrometer.MicrometerMetricsRecorder; +import com.clickhouse.examples.repository.SignalRepository; +import io.micrometer.core.instrument.MeterRegistry; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.context.annotation.Import; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.jdbc.core.JdbcTemplate; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.time.Duration; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * End-to-end ingestion test running against a real ClickHouse container. + */ +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = { + "iot.telemetry.otlp.enabled=false", + "management.otlp.metrics.export.enabled=false" + }) +@Import(TestcontainersConfiguration.class) +@Testcontainers(disabledWithoutDocker = true) +class SignalIngestionTests { + + @Autowired + TestRestTemplate rest; + + @Autowired + SignalRepository repository; + + @Autowired + OpenTelemetrySdk openTelemetry; + + @Autowired + JdbcTemplate jdbc; + + @Autowired + MeterRegistry meterRegistry; + + @Autowired + MetricsRecorder metricsRecorder; + + @Test + void configuresMicrometerMetricsRecorder() { + assertThat(metricsRecorder).isInstanceOf(MicrometerMetricsRecorder.class); + } + + @Test + void registersClickHouseHttpConnectionPoolMetrics() { + assertThat(meterRegistry + .find("httpcomponents.httpclient.pool.total.connections") + .tags("httpclient", "reconciliation", "state", "leased") + .gauge()) + .isNotNull(); + assertThat(meterRegistry + .find("httpcomponents.httpclient.pool.total.pending") + .tag("httpclient", "reconciliation") + .gauge()) + .isNotNull(); + assertThat(meterRegistry + .find("httpcomponents.httpclient.pool.total.max") + .tag("httpclient", "reconciliation") + .gauge()) + .isNotNull() + .extracting(gauge -> gauge.value()) + .isEqualTo(10.0); + } + + @Test + void rejectsRequestsWithoutApiKey() { + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + HttpEntity request = new HttpEntity<>( + "{\"deviceId\":\"sensor-1\",\"type\":\"TEMPERATURE\",\"value\":21.5}", headers); + + ResponseEntity response = rest.postForEntity("/api/v1/signals", request, String.class); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.UNAUTHORIZED); + } + + @Test + void acceptsAuthenticatedSignalAndStoresIt() { + long before = repository.count(); + UUID locationId = UUID.randomUUID(); + + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.set("X-API-Key", "dev-key-1"); + HttpEntity request = new HttpEntity<>( + """ + {"deviceId":"sensor-1","locationId":"%s","type":"TEMPERATURE","value":21.5,"unit":"C"} + """.formatted(locationId), + headers); + + ResponseEntity> response = rest.exchange( + "/api/v1/signals", + HttpMethod.POST, + request, + new ParameterizedTypeReference<>() {}); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(response.getBody()).containsEntry("status", "accepted"); + assertThat(repository.count()).isEqualTo(before + 1); + + assertThat(openTelemetry.getSdkMeterProvider().forceFlush().join(10, TimeUnit.SECONDS).isSuccess()) + .isTrue(); + Double exportedCount = jdbc.queryForObject(""" + SELECT max(value) + FROM otel_metrics + WHERE name = 'iot.signals.received' + AND attributes['signal.type'] = 'TEMPERATURE' + """, Double.class); + assertThat(exportedCount).isNotNull().isGreaterThanOrEqualTo(1.0); + + Double storageWrites = jdbc.queryForObject(""" + SELECT max(value) + FROM otel_metrics + WHERE name = 'iot.storage.operations' + AND attributes['storage.operation'] = 'insert' + AND attributes['outcome'] = 'success' + """, Double.class); + assertThat(storageWrites).isNotNull().isGreaterThanOrEqualTo(1.0); + + Double durationSamples = jdbc.queryForObject(""" + SELECT max(value) + FROM otel_metrics + WHERE name = 'iot.storage.duration' + AND type = 'histogram_count' + AND attributes['storage.operation'] = 'insert' + AND attributes['outcome'] = 'success' + """, Double.class); + assertThat(durationSamples).isNotNull().isGreaterThanOrEqualTo(1.0); + } + + @Test + void acceptsReconciliationBatchAndStoresItAsynchronously() throws InterruptedException { + UUID firstId = UUID.randomUUID(); + UUID secondId = UUID.randomUUID(); + UUID locationId = UUID.randomUUID(); + + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.set("X-API-Key", "dev-key-1"); + HttpEntity request = new HttpEntity<>(""" + [ + { + "signalId": "%s", + "deviceId": "reconciliation-1", + "locationId": "%s", + "signalType": "TEMPERATURE", + "value": 20.5, + "unit": "C" + }, + { + "signalId": "%s", + "deviceId": "reconciliation-2", + "locationId": "%s", + "signalType": "HUMIDITY", + "value": 51.0, + "unit": "%%" + } + ] + """.formatted(firstId, locationId, secondId, locationId), headers); + + ResponseEntity> response = rest.exchange( + "/api/v1/reconciliation", + HttpMethod.POST, + request, + new ParameterizedTypeReference<>() {}); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + assertThat(response.getBody()) + .containsEntry("status", "accepted") + .containsEntry("count", 2); + awaitReconciledSignals(firstId, secondId); + + assertThat(meterRegistry + .find("db.client.operation.duration") + .tag("db.system.name", "clickhouse") + .tag("db.operation.name", "insert") + .tag("db.collection.name", "iot_signals") + .timer()) + .isNotNull() + .satisfies(timer -> assertThat(timer.count()).isGreaterThanOrEqualTo(1)); + assertThat(meterRegistry + .find("clickhouse.client.operation.count") + .tag("db.system.name", "clickhouse") + .tag("db.operation.name", "insert") + .tag("db.collection.name", "iot_signals") + .counter()) + .isNotNull() + .satisfies(counter -> assertThat(counter.count()).isGreaterThanOrEqualTo(1.0)); + } + + @Test + void returnsOneSecondMeanRowsFilteredByLocation() { + UUID locationId = UUID.randomUUID(); + Instant eventTime = Instant.now() + .minusSeconds(1) + .truncatedTo(ChronoUnit.SECONDS) + .plusMillis(100); + + postSignal(locationId, "slice-device-1", "TEMPERATURE", 20.0, "C", eventTime); + postSignal(locationId, "slice-device-2", "TEMPERATURE", 24.0, "C", eventTime.plusMillis(100)); + postSignal(locationId, "slice-device-3", "HUMIDITY", 50.0, "%", eventTime.plusMillis(200)); + + ResponseEntity>> response = rest.exchange( + "/api/v1/signals/slices?lookback=10s&locationId=" + locationId, + HttpMethod.GET, + new HttpEntity<>(authenticatedHeaders()), + new ParameterizedTypeReference<>() {}); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(response.getBody()).singleElement().satisfies(row -> { + assertThat(row).containsEntry("locationId", locationId.toString()); + assertThat(((Number) row.get("temperature")).doubleValue()).isEqualTo(22.0); + assertThat(((Number) row.get("humidity")).doubleValue()).isEqualTo(50.0); + assertThat(row.get("pressure")).isNull(); + }); + + assertThat(meterRegistry + .find("db.client.operation.duration") + .tag("db.system.name", "clickhouse") + .tag("db.operation.name", "query") + .timer()) + .isNotNull() + .satisfies(timer -> assertThat(timer.count()).isGreaterThanOrEqualTo(1)); + assertThat(meterRegistry + .find("clickhouse.client.operation.count") + .tag("db.system.name", "clickhouse") + .tag("db.operation.name", "query") + .counter()) + .isNotNull() + .satisfies(counter -> assertThat(counter.count()).isGreaterThanOrEqualTo(1.0)); + } + + @Test + void rejectsSliceLookbackLongerThanOneHour() { + ResponseEntity response = rest.exchange( + "/api/v1/signals/slices?lookback=61m", + HttpMethod.GET, + new HttpEntity<>(authenticatedHeaders()), + String.class); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + } + + @Test + void rejectsUnknownSignalType() { + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.set("X-API-Key", "dev-key-1"); + HttpEntity request = new HttpEntity<>( + "{\"deviceId\":\"sensor-1\",\"type\":\"PLASMA\",\"value\":1.0}", headers); + + ResponseEntity response = rest.postForEntity("/api/v1/signals", request, String.class); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + } + + @Test + void rejectsSignalWithoutValue() { + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.set("X-API-Key", "dev-key-1"); + HttpEntity request = new HttpEntity<>( + "{\"deviceId\":\"sensor-1\",\"type\":\"TEMPERATURE\"}", headers); + + ResponseEntity response = rest.postForEntity("/api/v1/signals", request, String.class); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST); + } + + private void postSignal( + UUID locationId, + String deviceId, + String type, + double value, + String unit, + Instant eventTime) { + HttpHeaders headers = authenticatedHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + HttpEntity request = new HttpEntity<>(""" + { + "deviceId": "%s", + "locationId": "%s", + "type": "%s", + "value": %s, + "unit": "%s", + "timestamp": "%s" + } + """.formatted(deviceId, locationId, type, value, unit, eventTime), headers); + + ResponseEntity response = rest.postForEntity("/api/v1/signals", request, String.class); + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED); + } + + private HttpHeaders authenticatedHeaders() { + HttpHeaders headers = new HttpHeaders(); + headers.set("X-API-Key", "dev-key-1"); + return headers; + } + + private void awaitReconciledSignals(UUID firstId, UUID secondId) throws InterruptedException { + Instant deadline = Instant.now().plus(Duration.ofSeconds(10)); + long stored; + do { + stored = jdbc.queryForObject( + "SELECT count() FROM iot_signals WHERE signal_id IN (?, ?)", + Long.class, + firstId, + secondId); + if (stored == 2) { + return; + } + Thread.sleep(50); + } while (Instant.now().isBefore(deadline)); + + assertThat(stored).as("asynchronously reconciled signals").isEqualTo(2); + } +} diff --git a/examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestIotIngestApplication.java b/examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestIotIngestApplication.java new file mode 100755 index 000000000..8028513ee --- /dev/null +++ b/examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestIotIngestApplication.java @@ -0,0 +1,18 @@ +package com.clickhouse.examples; + +import org.springframework.boot.SpringApplication; + +/** + * Local development launcher: starts the application with a Testcontainers-managed ClickHouse. + * + *

Run with {@code ./mvnw spring-boot:test-run} (or from your IDE). A ClickHouse container + * is started automatically and the app connects to it; stop the process to tear it down. + */ +public class TestIotIngestApplication { + + public static void main(String[] args) { + SpringApplication.from(IotIngestApplication::main) + .with(TestcontainersConfiguration.class) + .run(args); + } +} diff --git a/examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestcontainersConfiguration.java b/examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestcontainersConfiguration.java new file mode 100755 index 000000000..c7c837e58 --- /dev/null +++ b/examples/demo-spring-service/src/test/java/com/clickhouse/examples/TestcontainersConfiguration.java @@ -0,0 +1,22 @@ +package com.clickhouse.examples; + +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.context.annotation.Bean; +import org.testcontainers.clickhouse.ClickHouseContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * Spins up a real ClickHouse instance in a container for local development and integration + * tests. {@link ServiceConnection} wires the container's JDBC coordinates straight into + * Spring's {@code DataSource}, so no manual configuration is needed. + */ +@TestConfiguration(proxyBeanMethods = false) +public class TestcontainersConfiguration { + + @Bean + @ServiceConnection + ClickHouseContainer clickHouseContainer() { + return new ClickHouseContainer(DockerImageName.parse("clickhouse/clickhouse-server:24.3")); + } +}