From f43e14506c14fdc7e426e39a981f1f337477c9bf Mon Sep 17 00:00:00 2001 From: Leon van Zantvoort Date: Sun, 12 Jul 2026 01:18:48 +0200 Subject: [PATCH] feat(micrometer): observe transactions as storm.transaction Observations Queries are observed since the Micrometer binding landed, but the transaction boundaries around them were not: long-running transactions, rollback rates, and per-domain transaction behavior were invisible. - The QueryObserver SPI gains a default no-op onTransaction hook, invoked when the pending transaction scope materializes a physical transaction (an outermost block or REQUIRES_NEW; joined blocks and savepoint scopes run inside an existing transaction and are not double-counted) and completed with the outcome. The hook sits in the shared scope engine, so blocking, suspend, and Spring-bridged transactions are covered identically. Observer failures are contained and never affect the transaction. - MicrometerQueryObserver reports storm.transaction observations with storm.tx.outcome (committed/rolled_back), storm.tx.propagation, storm.tx.read_only, and the observer's extra key values (storm.database in multi-database setups). - No new wiring: the starters, the Ktor plugin, and builder-composed templates pick the signal up through their existing query observer. Fixes #235 --- CHANGELOG.md | 1 + docs/ktor-integration.md | 2 +- docs/spring-integration.md | 2 + .../java/st/orm/core/spi/QueryObserver.java | 55 ++++++++ .../st/orm/core/spi/TransactionScope.java | 85 +++++++++++- .../impl/PreparedStatementTemplateImpl.java | 3 +- .../micrometer/MicrometerQueryObserver.java | 28 ++++ .../TransactionObservationTest.java | 125 ++++++++++++++++++ 8 files changed, 297 insertions(+), 4 deletions(-) create mode 100644 storm-micrometer/src/test/java/st/orm/micrometer/TransactionObservationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 60c336087..7d239ee1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,7 @@ Framework integration points are now instance-scoped (#198), and the Ktor integr - The Spring Boot starters bind query observations automatically: with an `ObservationRegistry` bean present (Actuator provides one), every query executed by the auto-configured template reports as a `storm.query` Micrometer Observation via the shared `StormObservationAutoConfiguration`. The starters ship storm-micrometer; override the convention with an `ObservationConvention` bean, replace the binding with a `QueryObserver` bean, or disable via `management.observations.enable.storm.query=false`. - `OtelDatabaseObservationConvention` (storm-micrometer): opt-in OpenTelemetry database client semantic conventions on query observations — `db.system.name` (derived from the JDBC URL), `db.operation.name`, and `db.query.text` (placeholders only) alongside the `storm.*` key values, so Storm queries surface in the database tooling of OTLP-capable observability backends. Activated with `storm.observations.semantic-conventions=otel` in the Spring Boot starters (database product detected from the DataSource) or by registering the convention in the Ktor dependency container. - Trace context in SQL statements: the `SqlCommenter` hook (template builder, all language stacks) appends per-execution comment content to statements after all processing and caching; `TraceContextSqlCommenter` (storm-micrometer) renders the current span as a sqlcommenter-style W3C `traceparent` comment, correlating database-side diagnostics such as slow query logs with traces. Opt-in everywhere — `storm.tracing.sql-comments=true` in the Spring Boot starters (with a `Tracer` present), the `sqlCommenter` slot in the Ktor plugin — since a per-execution comment defeats prepared statement caching. Hostile content is rejected (comment terminator, semicolons), and the emitted comment is padded so MySQL/MariaDB executable-comment and optimizer-hint markers are never interpreted. +- Transactions are observed alongside queries: every physical transaction (outermost block or `REQUIRES_NEW`, blocking, suspend, and Spring-bridged alike) reports as a `storm.transaction` Micrometer Observation with duration, outcome (`committed`/`rolled_back`), propagation, and read-only key values; joined blocks are not double-counted. The `QueryObserver` SPI gains a default no-op transaction hook, so existing observers and wiring are unaffected. - `@EnableStormRepositories(basePackages = ...)` (storm-spring, usable from both language stacks): switches on repository scanning in plain Spring applications, mirroring `@EnableJpaRepositories`; without packages, the annotated class's package is scanned. In Spring Boot the annotation doubles as the explicit override: the auto-configured scanning backs off. The `RepositoryBeanFactoryPostProcessor` adapters also gain configuring constructors (base packages, template bean name, prefix), so multi-template applications define one bean per repository set instead of one subclass. - `@DataStormTest` test slice in the new storm-spring-boot-test-autoconfigure module, the counterpart of `@DataJpaTest`: starts only the `DataSource`, Storm's auto-configuration, and SQL initialization (plus Flyway/Liquibase when present), keeps regular components out, replaces the data source with an embedded database on Spring Boot 3 (opt out via `spring.test.database.replace=none` for Testcontainers `@ServiceConnection` setups), and rolls back a per-test transaction. One module serves both starters and both Spring Boot 3 and 4: the slice imports the starters' identically named auto-configuration classes, lists relocated platform classes as optional imports, and registers its type exclusion through a context customizer instead of the relocated `@TypeExcludeFilters`. diff --git a/docs/ktor-integration.md b/docs/ktor-integration.md index d515a7db8..a9eb08b45 100644 --- a/docs/ktor-integration.md +++ b/docs/ktor-integration.md @@ -725,7 +725,7 @@ install(Storm) { ``` What each observation produces depends on the handlers attached to the registry: timing metrics, tracing spans, or both. Storm spans nest under the current trace context, so with context propagation in place (for example the OpenTelemetry agent, or micrometer-tracing with a `TracingObservationHandler`) queries appear under the active request span. -Observations are named `storm.query` and carry the following key values: +Physical transactions report as `storm.transaction` observations with their duration, outcome (`committed` or `rolled_back`), propagation, and read-only flag; joined blocks are not double-counted. Query observations are named `storm.query` and carry the following key values: | Key | Cardinality | Value | |-----|-------------|-------| diff --git a/docs/spring-integration.md b/docs/spring-integration.md index e41aad8c8..d51fe06dc 100644 --- a/docs/spring-integration.md +++ b/docs/spring-integration.md @@ -624,6 +624,8 @@ A generic DataSource proxy can only time statements. Storm knows the entity and - **Low-cardinality key values** (become metric tags): the SQL operation (`SELECT`, `INSERT`, ...), the execution kind (query, update, batch), and the entity or projection type. - **High-cardinality key values** (visible to trace handlers only): the SQL statement with parameter placeholders. Parameter values are never reported. +Transactions are observed alongside the queries: every physical transaction — an outermost `transaction` block or a `REQUIRES_NEW` block, in any language stack and through the Spring bridge alike — reports as a `storm.transaction` observation with its duration, `storm.tx.outcome` (`committed` or `rolled_back`), `storm.tx.propagation`, and `storm.tx.read_only`. Joined blocks run inside an existing transaction and are deliberately not double-counted. Long-running transactions and rollback rates become queryable per application, and per domain with the multi-template tagging. + Observability backends key their database tooling — latency panels, service-map database nodes, query views — on the OpenTelemetry database client semantic conventions. Set `storm.observations.semantic-conventions: otel` and every observation additionally carries the standard attributes (`db.system.name` detected from the DataSource, `db.operation.name`, and `db.query.text` on spans), so Storm queries surface in the database UX of any OTLP-capable backend, from Elastic to Grafana to Datadog, while the `storm.*` key values remain for custom dashboards: ```yaml diff --git a/storm-core/src/main/java/st/orm/core/spi/QueryObserver.java b/storm-core/src/main/java/st/orm/core/spi/QueryObserver.java index 59fe99277..59a45fb5a 100644 --- a/storm-core/src/main/java/st/orm/core/spi/QueryObserver.java +++ b/storm-core/src/main/java/st/orm/core/spi/QueryObserver.java @@ -43,6 +43,61 @@ public interface QueryObserver { */ Observation onExecute(@Nonnull QueryContext context); + /** + * Called when a physical transaction opens: an outermost transaction block, or a {@code REQUIRES_NEW} + * block. Joined blocks and savepoint scopes are not physical transactions and are not observed. + * + *

The default implementation ignores transactions.

+ * + * @param options the options the transaction was opened with; never {@code null}. + * @return the observation tracking this transaction; never {@code null}. + * @since 1.13 + */ + default TransactionObservation onTransaction(@Nonnull TransactionScope.Options options) { + return TransactionObservation.NOOP; + } + + /** + * Tracks a single observed physical transaction. + * + * @since 1.13 + */ + interface TransactionObservation { + + /** + * An observation that ignores all signals. + */ + TransactionObservation NOOP = new TransactionObservation() { + @Override + public void error(@Nonnull Throwable throwable) { + // Ignore. + } + + @Override + public void close(boolean rolledBack) { + // Ignore. + } + }; + + /** + * Signals that completing the observed transaction failed. + * + *

Invoked at most once, before {@link #close(boolean)}.

+ * + * @param throwable the failure. + */ + void error(@Nonnull Throwable throwable); + + /** + * Signals that the observed transaction has completed. + * + *

Invoked exactly once per observation.

+ * + * @param rolledBack whether the transaction rolled back rather than committed. + */ + void close(boolean rolledBack); + } + /** * Tracks a single observed statement execution. * diff --git a/storm-core/src/main/java/st/orm/core/spi/TransactionScope.java b/storm-core/src/main/java/st/orm/core/spi/TransactionScope.java index 6bd1f6ebe..6749112a1 100644 --- a/storm-core/src/main/java/st/orm/core/spi/TransactionScope.java +++ b/storm-core/src/main/java/st/orm/core/spi/TransactionScope.java @@ -67,6 +67,9 @@ public record Options(@Nullable TransactionPropagation propagation, private static final ThreadLocal CURRENT = new ThreadLocal<>(); + private static final java.util.logging.Logger OBSERVER_LOGGER = + java.util.logging.Logger.getLogger("st.orm.transaction"); + /** * Returns the thread local that holds the current transaction scope. * @@ -127,6 +130,19 @@ public static TransactionScope create(@Nonnull Options options, @Nullable Transa * @return the transaction context to execute under, or {@code null} when no transaction is active. * @throws PersistenceException if the active scope is owned by a different provider. */ + public static @Nullable TransactionContext resolveContext(@Nonnull TransactionTemplateProvider provider, + @Nullable QueryObserver queryObserver) { + var scope = current(); + if (scope != null) { + return scope.getOrMaterializeContext(provider, queryObserver); + } + return provider.getTransactionTemplate().currentContext().orElse(null); + } + + /** + * Resolves the transaction context without a query observer; physical transactions opened through this + * path are not observed. + */ public static @Nullable TransactionContext resolveContext(@Nonnull TransactionTemplateProvider provider) { var scope = CURRENT.get(); if (scope != null) { @@ -165,6 +181,7 @@ public static TransactionScope create(@Nonnull Options options, @Nullable Transa private volatile @Nullable TransactionTemplateProvider owner; private volatile @Nullable TransactionHandle handle; + private @Nullable QueryObserver.TransactionObservation observation; private boolean rollbackOnly; // Buffered until materialization; guarded by this. private boolean rollbackInherited; // Set when the mark came from a joined inner scope; guarded by this. @@ -232,6 +249,21 @@ public boolean isMaterialized() { * @throws PersistenceException if this scope, or an outer scope it must join, is owned by a different provider. */ public synchronized TransactionContext getOrMaterializeContext(@Nonnull TransactionTemplateProvider provider) { + return getOrMaterializeContext(provider, null); + } + + /** + * Returns the transaction context of this scope, materializing it through the given provider if needed and + * reporting a newly opened physical transaction to the given query observer. + * + * @param provider the transaction template provider of the executing template. + * @param queryObserver the query observer of the executing template, or {@code null} to observe nothing. + * @return the transaction context of this scope. + * @throws PersistenceException if this scope, or an outer scope it must join, is owned by a different provider. + * @since 1.13 + */ + public synchronized TransactionContext getOrMaterializeContext(@Nonnull TransactionTemplateProvider provider, + @Nullable QueryObserver queryObserver) { var existingHandle = handle; if (existingHandle != null) { if (owner != provider) { @@ -243,7 +275,7 @@ public synchronized TransactionContext getOrMaterializeContext(@Nonnull Transact // place before this scope opens. This preserves the semantics of eagerly entered transaction blocks: joining // propagations share the outer transaction, MANDATORY finds it, NEVER detects it, and REQUIRES_NEW suspends // it. The provider-identity check happens naturally in the ancestor's own materialization. - TransactionContext outerContext = parent != null ? parent.getOrMaterializeContext(provider) : null; + TransactionContext outerContext = parent != null ? parent.getOrMaterializeContext(provider, queryObserver) : null; var template = provider.getTransactionTemplate(); if (options.propagation() != null) { template = template.propagation(options.propagation()); @@ -264,9 +296,33 @@ public synchronized TransactionContext getOrMaterializeContext(@Nonnull Transact } this.owner = provider; this.handle = newHandle; + if (queryObserver != null && isPhysicalTransaction(outerContext)) { + try { + this.observation = queryObserver.onTransaction(options); + } catch (Throwable observerFailure) { + OBSERVER_LOGGER.log(java.util.logging.Level.WARNING, + "Transaction observer failed on open; transaction unaffected.", observerFailure); + } + } return newHandle.context(); } + /** + * Returns whether this scope opens a physical transaction of its own: an outermost transactional block, + * or a block that suspends the enclosing transaction with {@code REQUIRES_NEW}. Joined blocks and + * savepoint scopes run inside an existing physical transaction. + */ + private boolean isPhysicalTransaction(@Nullable TransactionContext outerContext) { + var propagation = options.propagation(); + if (propagation == TransactionPropagation.REQUIRES_NEW) { + return true; + } + return outerContext == null + && (propagation == null + || propagation == TransactionPropagation.REQUIRED + || propagation == TransactionPropagation.MANDATORY); + } + /** * Returns the number of seconds remaining until the scope's deadline, rounded up; {@code 0} when the deadline * has already passed. @@ -370,7 +426,15 @@ public synchronized boolean isRollbackOnly() { public synchronized void complete(boolean rollback) { var currentHandle = handle; if (currentHandle != null) { - currentHandle.complete(rollback); + var currentObservation = observation; + observation = null; + try { + currentHandle.complete(rollback); + } catch (Throwable completionFailure) { + closeObservation(currentObservation, true, completionFailure); + throw completionFailure; + } + closeObservation(currentObservation, rollback || rollbackOnly, null); return; } if ((rollback || rollbackOnly) && isJoining() && parent != null) { @@ -378,6 +442,23 @@ public synchronized void complete(boolean rollback) { } } + private static void closeObservation(@Nullable QueryObserver.TransactionObservation observation, + boolean rolledBack, + @Nullable Throwable error) { + if (observation == null) { + return; + } + try { + if (error != null) { + observation.error(error); + } + observation.close(rolledBack); + } catch (Throwable observerFailure) { + OBSERVER_LOGGER.log(java.util.logging.Level.WARNING, + "Transaction observer failed on completion; transaction unaffected.", observerFailure); + } + } + /** * Uninstalls this scope from the current thread, restoring its parent as the current scope. * diff --git a/storm-core/src/main/java/st/orm/core/template/impl/PreparedStatementTemplateImpl.java b/storm-core/src/main/java/st/orm/core/template/impl/PreparedStatementTemplateImpl.java index 22b119f3e..cb85fc287 100644 --- a/storm-core/src/main/java/st/orm/core/template/impl/PreparedStatementTemplateImpl.java +++ b/storm-core/src/main/java/st/orm/core/template/impl/PreparedStatementTemplateImpl.java @@ -308,7 +308,8 @@ private static TemplateProcessor createDataSourceProcessor(@Nonnull DataSource d var parameters = sql.parameters(); var bindVariables = sql.bindVariables().orElse(null); var generatedKeys = sql.generatedKeys(); - var transactionContext = TransactionScope.resolveContext(transactionTemplateProvider); + var transactionContext = TransactionScope.resolveContext(transactionTemplateProvider, + strategies.queryObserver()); Connection connection = connectionProvider.getConnection(dataSource, transactionContext); PreparedStatement preparedStatement = null; boolean success = false; diff --git a/storm-micrometer/src/main/java/st/orm/micrometer/MicrometerQueryObserver.java b/storm-micrometer/src/main/java/st/orm/micrometer/MicrometerQueryObserver.java index c285158d7..af894a9c8 100644 --- a/storm-micrometer/src/main/java/st/orm/micrometer/MicrometerQueryObserver.java +++ b/storm-micrometer/src/main/java/st/orm/micrometer/MicrometerQueryObserver.java @@ -23,6 +23,7 @@ import jakarta.annotation.Nonnull; import st.orm.core.spi.QueryContext; import st.orm.core.spi.QueryObserver; +import st.orm.core.spi.TransactionScope; /** * {@link QueryObserver} that reports Storm query executions as Micrometer @@ -84,6 +85,33 @@ public MicrometerQueryObserver(@Nonnull ObservationRegistry observationRegistry, this.extraLowCardinalityKeyValues = requireNonNull(extraLowCardinalityKeyValues, "extraLowCardinalityKeyValues"); } + @Override + public TransactionObservation onTransaction(@Nonnull TransactionScope.Options options) { + var observation = io.micrometer.observation.Observation + .createNotStarted("storm.transaction", observationRegistry) + .contextualName("transaction") + .lowCardinalityKeyValue("storm.tx.propagation", + options.propagation() != null ? options.propagation().name() : "REQUIRED") + .lowCardinalityKeyValue("storm.tx.read_only", + String.valueOf(Boolean.TRUE.equals(options.readOnly()))); + for (var keyValue : extraLowCardinalityKeyValues) { + observation = observation.lowCardinalityKeyValue(keyValue.getKey(), keyValue.getValue()); + } + var started = observation.start(); + return new TransactionObservation() { + @Override + public void error(@Nonnull Throwable throwable) { + started.error(throwable); + } + + @Override + public void close(boolean rolledBack) { + started.lowCardinalityKeyValue("storm.tx.outcome", rolledBack ? "rolled_back" : "committed"); + started.stop(); + } + }; + } + @Override public Observation onExecute(@Nonnull QueryContext context) { var observationContext = new StormQueryObservationContext(context, extraLowCardinalityKeyValues); diff --git a/storm-micrometer/src/test/java/st/orm/micrometer/TransactionObservationTest.java b/storm-micrometer/src/test/java/st/orm/micrometer/TransactionObservationTest.java new file mode 100644 index 000000000..ba070f3d0 --- /dev/null +++ b/storm-micrometer/src/test/java/st/orm/micrometer/TransactionObservationTest.java @@ -0,0 +1,125 @@ +/* + * Copyright 2024 - 2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package st.orm.micrometer; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.micrometer.observation.tck.TestObservationRegistry; +import java.util.List; +import org.junit.jupiter.api.Test; +import st.orm.TransactionPropagation; +import st.orm.core.spi.TransactionScope; +import st.orm.core.template.ORMTemplate; + +/** + * Tests for the {@code storm.transaction} observation: physical transactions are observed with their + * outcome, joined blocks are not, and {@code REQUIRES_NEW} opens its own observation. + */ +public class TransactionObservationTest { + + private static TransactionScope.Options options(TransactionPropagation propagation) { + return new TransactionScope.Options(propagation, null, null, null, false); + } + + private ORMTemplate template(TestObservationRegistry registry, String database) { + return ORMTemplate.builder(new SimpleDataSource("jdbc:h2:mem:" + database + ";DB_CLOSE_DELAY=-1")) + .queryObserver(new MicrometerQueryObserver(registry)) + .build(); + } + + private List transactionContexts(TestObservationRegistry registry) { + var contexts = new java.util.ArrayList(); + io.micrometer.observation.tck.TestObservationRegistryAssert.assertThat(registry) + .hasHandledContextsThatSatisfy(handled -> handled.stream() + .filter(context -> "storm.transaction".equals(context.getName())) + .forEach(contexts::add)); + return contexts; + } + + @Test + public void committedTransactionIsObservedWithOutcome() { + var registry = TestObservationRegistry.create(); + var orm = template(registry, "txobserved"); + var scope = TransactionScope.open(options(null)); + try { + orm.query("SELECT 1").getSingleResult(Integer.class); + scope.complete(false); + } finally { + scope.close(); + } + var contexts = transactionContexts(registry); + assertEquals(1, contexts.size()); + assertEquals("committed", keyValue(contexts.get(0), "storm.tx.outcome")); + assertEquals("REQUIRED", keyValue(contexts.get(0), "storm.tx.propagation")); + } + + @Test + public void rolledBackTransactionCarriesTheOutcome() { + var registry = TestObservationRegistry.create(); + var orm = template(registry, "txrollback"); + var scope = TransactionScope.open(options(null)); + try { + orm.query("SELECT 1").getSingleResult(Integer.class); + scope.complete(true); + } finally { + scope.close(); + } + var contexts = transactionContexts(registry); + assertEquals(1, contexts.size()); + assertEquals("rolled_back", keyValue(contexts.get(0), "storm.tx.outcome")); + } + + @Test + public void joinedBlocksAreNotObservedButRequiresNewIs() { + var registry = TestObservationRegistry.create(); + var orm = template(registry, "txnesting"); + var outer = TransactionScope.open(options(null)); + try { + orm.query("SELECT 1").getSingleResult(Integer.class); + // Joined inner block: same physical transaction, no observation. + var joined = TransactionScope.open(options(TransactionPropagation.REQUIRED)); + try { + orm.query("SELECT 2").getSingleResult(Integer.class); + joined.complete(false); + } finally { + joined.close(); + } + // REQUIRES_NEW: its own physical transaction, its own observation. + var independent = TransactionScope.open(options(TransactionPropagation.REQUIRES_NEW)); + try { + orm.query("SELECT 3").getSingleResult(Integer.class); + independent.complete(false); + } finally { + independent.close(); + } + outer.complete(false); + } finally { + outer.close(); + } + var contexts = transactionContexts(registry); + assertEquals(2, contexts.size()); + var propagations = contexts.stream() + .map(context -> keyValue(context, "storm.tx.propagation")) + .sorted() + .toList(); + assertEquals(List.of("REQUIRED", "REQUIRES_NEW"), propagations); + } + + private static String keyValue(io.micrometer.observation.Observation.Context context, String key) { + var keyValue = context.getLowCardinalityKeyValue(key); + return keyValue != null ? keyValue.getValue() : null; + } +}