diff --git a/spring-data-cassandra-distribution/pom.xml b/spring-data-cassandra-distribution/pom.xml index c08ddfa12..23a8c748b 100644 --- a/spring-data-cassandra-distribution/pom.xml +++ b/spring-data-cassandra-distribution/pom.xml @@ -42,48 +42,30 @@ exec-maven-plugin - generate-metrics-metadata + generate-docs generate-resources java - io.micrometer.docs.metrics.DocsFromSources - - - - generate-tracing-metadata - generate-resources - - java - - - io.micrometer.docs.spans.DocsFromSources + io.micrometer.docs.DocsGeneratorCommand + true + + ${micrometer-docs-generator.inputPath} + ${micrometer-docs-generator.inclusionPattern} + ${micrometer-docs-generator.outputPath} + io.micrometer - micrometer-docs-generator-spans - ${micrometer-docs-generator} - jar - - - io.micrometer - micrometer-docs-generator-metrics + micrometer-docs-generator ${micrometer-docs-generator} jar - - true - - ${micrometer-docs-generator.inputPath} - ${micrometer-docs-generator.inclusionPattern} - ${micrometer-docs-generator.outputPath} - - org.asciidoctor diff --git a/spring-data-cassandra/pom.xml b/spring-data-cassandra/pom.xml index 8f39acbd0..6d22e8a50 100644 --- a/spring-data-cassandra/pom.xml +++ b/spring-data-cassandra/pom.xml @@ -77,6 +77,7 @@ micrometer-observation true + io.micrometer micrometer-tracing @@ -87,6 +88,12 @@ io.micrometer micrometer-test test + + + com.github.tomakehurst + wiremock-jre8-standalone + + diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java index bc145260e..4d8787490 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java @@ -256,13 +256,14 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { } /** - * Returns the list of CQL scripts to be run on startup after {@link #getKeyspaceCreations() Keyspace creations} - * and after initialization of the {@literal System} Keyspace. + * Returns the list of CQL scripts to be run on startup after {@link #getKeyspaceCreations() Keyspace creations} and + * after initialization of the {@literal System} Keyspace. return super.getSessionBuilderConfigurer(); * - * @return the list of CQL scripts to be run on startup; may be {@link Collections#emptyList() empty} - * but never {@literal null}. + * @return the list of CQL scripts to be run on startup; may be {@link Collections#emptyList() empty} but never + * {@literal null}. * @deprecated since 3.0; Declare a - * {@link org.springframework.data.cassandra.core.cql.session.init.SessionFactoryInitializer} bean instead. + * {@link org.springframework.data.cassandra.core.cql.session.init.SessionFactoryInitializer} bean + * instead. */ @Deprecated protected List getStartupScripts() { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservation.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservation.java index 7db40d305..d7821c0cf 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservation.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservation.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2022 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. @@ -17,6 +17,7 @@ package org.springframework.data.cassandra.observability; import io.micrometer.common.docs.KeyName; import io.micrometer.observation.docs.ObservationDocumentation; +import io.micrometer.tracing.docs.EventValue; /** * Cassandra-based implementation of {@link ObservationDocumentation}. @@ -24,7 +25,7 @@ import io.micrometer.observation.docs.ObservationDocumentation; * @author Mark Paluch * @author Marcin Grzejszczak * @author Greg Turnquist - * @since 4.0.0 + * @since 4.0 */ enum CassandraObservation implements ObservationDocumentation { @@ -101,15 +102,6 @@ enum CassandraObservation implements ObservationDocumentation { } }, - /** - * A tag containing error that occurred for the given node. - */ - NODE_ERROR_TAG { - @Override - public String asString() { - return "spring.data.cassandra.node[%s].error"; - } - } } enum HighCardinalityKeyNames implements KeyName { @@ -122,6 +114,41 @@ enum CassandraObservation implements ObservationDocumentation { public String asString() { return "spring.data.cassandra.cql"; } + }, + + /** + * A tag containing error that occurred for the given node. + */ + NODE_ERROR_TAG { + @Override + public String asString() { + return "spring.data.cassandra.node[%s].error"; + } } + + } + + enum Events implements EventValue { + + /** + * Set whenever an error occurred for the given node. + */ + NODE_ERROR { + @Override + public String getValue() { + return "cassandra.node.error"; + } + }, + + /** + * Set when a success occurred for the session processing. + */ + NODE_SUCCESS { + @Override + public String getValue() { + return "cassandra.node.success"; + } + } + } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConfiguration.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConfiguration.java deleted file mode 100644 index e0723b510..000000000 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConfiguration.java +++ /dev/null @@ -1,52 +0,0 @@ -/* - * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; - -import io.micrometer.observation.ObservationRegistry; -import io.micrometer.tracing.Tracer; - -import java.util.Optional; - -import org.springframework.context.annotation.Bean; - -/** - * Set of beans enable observability with Spring Data Cassandra. - * - * @author Greg Turnquist - * @since 4.0.0 - */ -public class CassandraObservationConfiguration { - - @Bean - CqlSessionObservationConvention observationConvention() { - return new DefaultCassandraObservationConvention(); - } - - @Bean - CqlSessionTracingObservationHandler cqlSessionTracingObservationHandler( - Optional observationRegistry, Tracer tracer) { - - CqlSessionTracingObservationHandler observationHandler = new CqlSessionTracingObservationHandler(tracer); - observationRegistry.ifPresent(registry -> registry.observationConfig().observationHandler(observationHandler)); - return observationHandler; - } - - @Bean - CqlSessionTracingBeanPostProcessor traceCqlSessionBeanPostProcessor(ObservationRegistry observationRegistry, - CqlSessionObservationConvention observationConvention) { - return new CqlSessionTracingBeanPostProcessor(observationRegistry, observationConvention); - } -} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionContext.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationContext.java similarity index 54% rename from spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionContext.java rename to spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationContext.java index f4852ec28..bc966ba64 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionContext.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationContext.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2022 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. @@ -15,43 +15,60 @@ */ package org.springframework.data.cassandra.observability; -import io.micrometer.observation.Observation; - -import org.springframework.lang.Nullable; - import com.datastax.oss.driver.api.core.CqlSession; import com.datastax.oss.driver.api.core.cql.Statement; +import io.micrometer.observation.Observation; +import io.micrometer.observation.transport.Kind; +import io.micrometer.observation.transport.SenderContext; + /** * A {@link Observation.Context} for {@link CqlSession}. * * @author Greg Turnquist - * @since 4.0.0 + * @author Mark Paluch + * @since 4.0 */ -public class CqlSessionContext extends Observation.Context { +public class CassandraObservationContext extends SenderContext { - private final @Nullable Statement statement; + private final Statement statement; + + private final boolean prepare; private final String methodName; - private final @Nullable CqlSession delegateSession; + private final String sessionName; + private final String keyspaceName; - public CqlSessionContext(@Nullable Statement statement, String methodName, @Nullable CqlSession delegateSession) { + public CassandraObservationContext(Statement statement, String remoteServiceName, boolean prepare, + String methodName, String sessionName, String keyspaceName) { + + super((carrier, key, value) -> {}, Kind.CLIENT); this.statement = statement; + this.prepare = prepare; this.methodName = methodName; - this.delegateSession = delegateSession; + this.sessionName = sessionName; + this.keyspaceName = keyspaceName; + + setRemoteServiceName(remoteServiceName); } - @Nullable public Statement getStatement() { return statement; } + public boolean isPrepare() { + return prepare; + } + public String getMethodName() { return methodName; } - @Nullable - public CqlSession getDelegateSession() { - return delegateSession; + public String getSessionName() { + return sessionName; + } + + public String getKeyspaceName() { + return keyspaceName; } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationConvention.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConvention.java similarity index 79% rename from spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationConvention.java rename to spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConvention.java index da3c12a10..b02af625a 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationConvention.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConvention.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2022 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. @@ -22,12 +22,12 @@ import io.micrometer.observation.ObservationConvention; * {@link ObservationConvention} for Cassandra. * * @author Greg Turnquist - * @since 4.0.0 + * @since 4.0 */ -public interface CqlSessionObservationConvention extends ObservationConvention { +public interface CassandraObservationConvention extends ObservationConvention { @Override default boolean supportsContext(Observation.Context context) { - return context instanceof CqlSessionContext; + return context instanceof CassandraObservationContext; } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/EnableCassandraObservability.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationSupplier.java similarity index 60% rename from spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/EnableCassandraObservability.java rename to spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationSupplier.java index 4f9685cb5..8a03cb4b3 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/EnableCassandraObservability.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationSupplier.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2022 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. @@ -15,20 +15,20 @@ */ package org.springframework.data.cassandra.observability; -import java.lang.annotation.*; - -import org.springframework.context.annotation.Import; +import io.micrometer.observation.Observation; /** - * Annotation to enable Cassandra observability. + * Returns the Cassandra Observation. Used internally - do not implement. * - * @author Greg Turnquist - * @since 4.0.0 + * @author Mark Paluch + * @author Marcin Grzejszczak + * @since 4.0 */ -@Inherited -@Documented -@Target(ElementType.TYPE) -@Retention(RetentionPolicy.RUNTIME) -@Import(CassandraObservationConfiguration.class) -public @interface EnableCassandraObservability { +interface CassandraObservationSupplier { + + /** + * @return the observation + */ + Observation getObservation(); + } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.java new file mode 100644 index 000000000..dc7539fbe --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.java @@ -0,0 +1,186 @@ +/* + * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; + +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Map; +import java.util.function.Function; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; + +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import com.datastax.oss.driver.api.core.cql.Statement; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; + +/** + * A {@link MethodInterceptor} that wraps calls around {@link CqlSession} in a trace representation. This interceptor + * wraps statements for {@code execute} and {@code prepare} (including their asynchronous variants) only. Graph and + * reactive {@link CqlSession} method are called as-is. + * + * @author Mark Paluch + * @author Marcin Grzejszczak + * @author Greg Turnquist + * @since 4.0.0 + */ +final class CqlSessionObservationInterceptor implements MethodInterceptor { + + private final CqlSession delegate; + + private final String remoteServiceName; + + private final ObservationRegistry observationRegistry; + + private final CassandraObservationConvention observationConvention = new DefaultCassandraObservationConvention(); + + CqlSessionObservationInterceptor(CqlSession delegate, String remoteServiceName, + ObservationRegistry observationRegistry) { + + this.delegate = delegate; + this.remoteServiceName = remoteServiceName; + this.observationRegistry = observationRegistry; + } + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + + Method method = invocation.getMethod(); + Object[] args = invocation.getArguments(); + + if (method.getName().equals("execute") && args.length > 0) { + return observe(createStatement(args), method.getName(), this.delegate::execute); + } + + if (method.getName().equals("executeAsync") && args.length > 0) { + return observe(createStatement(args), method.getName(), this.delegate::executeAsync); + } + + // prepare calls do not notify RequestTracker so we need to stop the observation ourselves + if (method.getName().equals("prepare") && args.length > 0) { + + Statement statement = createStatement(args); + + if (ObservationStatement.isObservationStatement(statement)) { + return this.delegate.prepareAsync((SimpleStatement) statement); + } + + Observation observation = startObservation(statement, true, "prepare"); + + try { + return this.delegate.prepare((SimpleStatement) ObservationStatement.createProxy(observation, statement)); + } catch (RuntimeException e) { + + observation.error(e); + throw e; + } finally { + observation.stop(); + } + } + + if (method.getName().equals("prepareAsync") && args.length > 0) { + + Statement statement = createStatement(args); + + if (ObservationStatement.isObservationStatement(statement)) { + return this.delegate.prepareAsync((SimpleStatement) statement); + } + + Observation observation = startObservation(statement, true, "prepareAsync"); + + return this.delegate.prepareAsync((SimpleStatement) ObservationStatement.createProxy(observation, statement)) + .whenComplete((preparedStatement, throwable) -> { + + if (throwable != null) { + observation.error(throwable); + } + + observation.stop(); + }); + } + + return invocation.proceed(); + } + + /** + * Convert list of arguments into a {@link Statement}. + * + * @param args + * @return CQL statement + */ + private static Statement createStatement(Object[] args) { + + if (args[0] instanceof Statement) { + return (Statement) args[0]; + } + + if (args[0] instanceof String & args.length == 1) { + return SimpleStatement.newInstance((String) args[0]); + } + + if (args[0]instanceof String query && args.length == 2) { + return args[1] instanceof Map // + ? SimpleStatement.newInstance(query, (Map) args[1]) // + : SimpleStatement.newInstance(query, (Object[]) args[1]); + } + + throw new IllegalArgumentException(String.format("Unsupported arguments %s", Arrays.toString(args))); + } + + /** + * Observe a {@link Statement}. + * + * @param statement original CQL {@link Statement} + * @param statementExecutor function that transforms a {@link Statement} into a resulting {@link Object} + * @return the statement execution result. + */ + private Object observe(Statement statement, String methodName, Function, Object> statementExecutor) { + + // avoid duplicate statement wrapping + if (ObservationStatement.isObservationStatement(statement)) { + return statementExecutor.apply(statement); + } + + Statement observableStatement = ObservationStatement.createProxy(startObservation(statement, false, methodName), + statement); + + return statementExecutor.apply(observableStatement); + } + + private Observation startObservation(Statement statement, boolean prepare, String methodName) { + + Observation currentObservation = observationRegistry.getCurrentObservation(); + + Observation observation = Observation + .createNotStarted(methodName, + () -> new CassandraObservationContext(statement, remoteServiceName, prepare, methodName, + delegate.getContext().getSessionName(), + delegate.getKeyspace().map(CqlIdentifier::asInternal).orElse("system")), + observationRegistry) + .observationConvention(observationConvention); + + if (currentObservation != null) { + observation.parentObservation(currentObservation); + } + + return observation.start(); + } + +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingInterceptor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingInterceptor.java deleted file mode 100644 index d6c1f4462..000000000 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingInterceptor.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; - -import io.micrometer.observation.Observation; -import io.micrometer.observation.ObservationRegistry; - -import java.lang.reflect.Method; -import java.util.Arrays; -import java.util.Map; -import java.util.function.Function; - -import org.aopalliance.intercept.MethodInterceptor; -import org.aopalliance.intercept.MethodInvocation; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; - -import com.datastax.oss.driver.api.core.CqlSession; -import com.datastax.oss.driver.api.core.cql.ResultSet; -import com.datastax.oss.driver.api.core.cql.SimpleStatement; -import com.datastax.oss.driver.api.core.cql.Statement; - -/** - * A {@link MethodInterceptor} that wraps calls around {@link CqlSession} in a trace representation. This interceptor - * wraps statements for {@code execute} and {@code prepare} (including their asynchronous variants) only. Graph and - * reactive {@link CqlSession} method are called as-is. - * - * @author Mark Paluch - * @author Marcin Grzejszczak - * @author Greg Turnquist - * @since 4.0.0 - */ -final class CqlSessionTracingInterceptor implements MethodInterceptor { - - private static final Log log = LogFactory.getLog(CqlSessionTracingInterceptor.class); - - private final CqlSession delegateSession; - - private final ObservationRegistry observationRegistry; - - private CqlSessionObservationConvention observationConvention; - - CqlSessionTracingInterceptor(CqlSession delegateSession, ObservationRegistry observationRegistry, - CqlSessionObservationConvention observationConvention) { - - this.delegateSession = delegateSession; - this.observationRegistry = observationRegistry; - this.observationConvention = observationConvention; - } - - @Nullable - @Override - public Object invoke(@NotNull MethodInvocation invocation) throws Throwable { - - Method method = invocation.getMethod(); - Object[] args = invocation.getArguments(); - - if (method.getName().equals("execute") && args.length > 0) { - return tracedCall(createStatement(args), method.getName(), this.delegateSession::execute); - } - - if (method.getName().equals("executeAsync") && args.length > 0) { - return tracedCall(createStatement(args), method.getName(), this.delegateSession::executeAsync); - } - - if (method.getName().equals("prepare") && args.length > 0) { - return tracedCall(createStatement(args), method.getName(), - statement -> this.delegateSession.prepare((SimpleStatement) statement)); - } - - if (method.getName().equals("prepareAsync") && args.length > 0) { - return tracedCall(createStatement(args), method.getName(), - statement -> this.delegateSession.prepareAsync((SimpleStatement) statement)); - } - - return invocation.proceed(); - } - - /** - * Apply tracing to a {@link Statement}. - * - * @param statement original CQL {@link Statement} - * @param statementExecutor function that transforms a {@link Statement} into a resulting {@link Object} - * @return {@link ResultSet} - */ - private Object tracedCall(Statement statement, String methodName, - Function, Object> statementExecutor) { - - if (this.observationRegistry.getCurrentObservation() == null) { - return statementExecutor.apply(statement); - } - - Observation observation = childObservation(statement, methodName, this.delegateSession); - - if (log.isDebugEnabled()) { - log.debug("Created a new child observation before query [" + observation + "]"); - } - - return observation.observe(() -> statementExecutor.apply(statement)); - } - - /** - * Convert list of arguments into a {@link Statement}. - * - * @param args - * @return CQL statement - */ - private static Statement createStatement(Object[] args) { - - if (args[0] instanceof Statement) { - return (Statement) args[0]; - } - - if (args[0] instanceof String & args.length == 1) { - return SimpleStatement.newInstance((String) args[0]); - } - - if (args[0]instanceof String query && args.length == 2) { - return args[1] instanceof Map // - ? SimpleStatement.newInstance(query, (Map) args[1]) // - : SimpleStatement.newInstance(query, (Object[]) args[1]); - } - - throw new IllegalArgumentException(String.format("Unsupported arguments %s", Arrays.toString(args))); - } - - private Observation childObservation(Statement statement, String methodName, CqlSession delegateSession) { - - CqlSessionContext observationContext = new CqlSessionContext(statement, methodName, delegateSession); - - return CassandraObservation.CASSANDRA_QUERY_OBSERVATION // - .observation(this.observationRegistry, () -> observationContext) // - .contextualName(CassandraObservation.CASSANDRA_QUERY_OBSERVATION.getContextualName()) // - .highCardinalityKeyValues(this.observationConvention.getHighCardinalityKeyValues(observationContext)) // - .lowCardinalityKeyValues(this.observationConvention.getLowCardinalityKeyValues(observationContext)); - // .start(); - } -} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingObservationHandler.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingObservationHandler.java deleted file mode 100644 index 725639f55..000000000 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingObservationHandler.java +++ /dev/null @@ -1,102 +0,0 @@ -/* - * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; - -import io.micrometer.common.KeyValue; -import io.micrometer.observation.Observation; -import io.micrometer.tracing.Span; -import io.micrometer.tracing.Tracer; -import io.micrometer.tracing.handler.TracingObservationHandler; - -import java.net.URI; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -/** - * A {@link TracingObservationHandler} for {@link CqlSessionContext}. - * - * @author Greg Turnquist - * @since 4.0.0 - */ -public class CqlSessionTracingObservationHandler implements TracingObservationHandler { - - private static final Log log = LogFactory.getLog(CqlSessionTracingObservationHandler.class); - - private final Tracer tracer; - - public CqlSessionTracingObservationHandler(Tracer tracer) { - this.tracer = tracer; - } - - @Override - public void onStart(CqlSessionContext context) { - - Span.Builder builder = this.tracer.spanBuilder() // - .name(context.getContextualName()) // - .kind(Span.Kind.CLIENT); - - getTracingContext(context).setSpan(builder.start()); - } - - @Override - public void onStop(CqlSessionContext context) { - - Span span = getRequiredSpan(context); - tagSpan(context, span); - - String sessionName = null; - String url = null; - - for (KeyValue keyValue : context.getLowCardinalityKeyValues()) { - - if (keyValue.getKey().equals(CassandraObservation.LowCardinalityKeyNames.SESSION_NAME.asString())) { - sessionName = keyValue.getValue(); - } - - if (keyValue.getKey().equals(CassandraObservation.LowCardinalityKeyNames.URL.asString())) { - url = keyValue.getValue(); - } - } - - if (sessionName != null) { - span.remoteServiceName("cassandra-" + sessionName); - } - - if (url != null) { - try { - URI uri = URI.create(url); - span.remoteIpAndPort(uri.getHost(), uri.getPort()); - } catch (Exception e) { - if (log.isDebugEnabled()) { - log.debug("Failed to parse the url [" + url + "]; Won't set this value on the span"); - } - } - } - - span.end(); - } - - @Override - public boolean supportsContext(Observation.Context context) { - return context instanceof CqlSessionContext; - } - - @Override - public Tracer getTracer() { - return this.tracer; - } -} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraObservationConvention.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraObservationConvention.java index 43919353b..b567ce7dc 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraObservationConvention.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraObservationConvention.java @@ -15,33 +15,34 @@ */ package org.springframework.data.cassandra.observability; -import io.micrometer.common.KeyValues; - -import java.util.Optional; import java.util.StringJoiner; import org.springframework.data.cassandra.observability.CassandraObservation.HighCardinalityKeyNames; import org.springframework.data.cassandra.observability.CassandraObservation.LowCardinalityKeyNames; +import org.springframework.util.StringUtils; -import com.datastax.oss.driver.api.core.CqlIdentifier; -import com.datastax.oss.driver.api.core.cql.*; +import com.datastax.oss.driver.api.core.cql.BatchStatement; +import com.datastax.oss.driver.api.core.cql.BatchableStatement; +import com.datastax.oss.driver.api.core.cql.BoundStatement; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import com.datastax.oss.driver.api.core.cql.Statement; + +import io.micrometer.common.KeyValues; /** - * Default {@link CqlSessionObservationConvention} implementation. + * Default {@link CassandraObservationConvention} implementation. * * @author Greg Turnquist - * @since 4.0.0 + * @author Mark Paluch + * @since 4.0 */ -public class DefaultCassandraObservationConvention implements CqlSessionObservationConvention { +class DefaultCassandraObservationConvention implements CassandraObservationConvention { @Override - public KeyValues getLowCardinalityKeyValues(CqlSessionContext context) { + public KeyValues getLowCardinalityKeyValues(CassandraObservationContext context) { - KeyValues keyValues = KeyValues.of( - LowCardinalityKeyNames.SESSION_NAME - .withValue(Optional.ofNullable(context.getDelegateSession().getName()).orElse("unknown")), - LowCardinalityKeyNames.KEYSPACE_NAME - .withValue(context.getDelegateSession().getKeyspace().map(CqlIdentifier::asInternal).orElse("unknown")), + KeyValues keyValues = KeyValues.of(LowCardinalityKeyNames.SESSION_NAME.withValue(context.getSessionName()), + LowCardinalityKeyNames.KEYSPACE_NAME.withValue(context.getKeyspaceName()), LowCardinalityKeyNames.METHOD_NAME.withValue(context.getMethodName())); if (context.getStatement().getNode() != null) { @@ -53,10 +54,15 @@ public class DefaultCassandraObservationConvention implements CqlSessionObservat } @Override - public KeyValues getHighCardinalityKeyValues(CqlSessionContext context) { + public KeyValues getHighCardinalityKeyValues(CassandraObservationContext context) { return KeyValues.of(HighCardinalityKeyNames.CQL_TAG.withValue(getCql(context.getStatement()))); } + @Override + public String getContextualName(CassandraObservationContext context) { + return (context.isPrepare() ? "PREPARE: " : "") + getSpanName(getCql(context.getStatement()), ""); + } + /** * Extract the CQL query from the delegate {@link Statement}. * @@ -67,7 +73,7 @@ public class DefaultCassandraObservationConvention implements CqlSessionObservat String query = ""; - if (statement instanceof SimpleStatement) { + if (statement instanceof SimpleStatement || statement instanceof BoundStatement) { query = getQuery(statement); } @@ -103,4 +109,19 @@ public class DefaultCassandraObservationConvention implements CqlSessionObservat return ""; } + + /** + * Tries to parse the CQL query or provides the default name. + * + * @param defaultName if there's no query + * @return span name + */ + public String getSpanName(String cql, String defaultName) { + + if (StringUtils.hasText(cql) && cql.indexOf(' ') > -1) { + return cql.substring(0, cql.indexOf(' ')); + } + + return defaultName; + } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingFactory.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactory.java similarity index 51% rename from spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingFactory.java rename to spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactory.java index 071324b8c..b87d82e12 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingFactory.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactory.java @@ -15,40 +15,55 @@ */ package org.springframework.data.cassandra.observability; -import io.micrometer.observation.ObservationRegistry; - import org.springframework.aop.framework.ProxyFactory; +import org.springframework.util.Assert; import com.datastax.oss.driver.api.core.CqlSession; +import io.micrometer.observation.ObservationRegistry; + /** - * Factory to wrap a {@link CqlSession} with a {@link CqlSessionTracingInterceptor}. + * Factory to wrap a {@link CqlSession} with a {@link CqlSessionObservationInterceptor}. * * @author Mark Paluch * @author Greg Turnquist - * @since 4.0.0 + * @since 4.0 */ -public final class CqlSessionTracingFactory { +public final class ObservableCqlSessionFactory { - private CqlSessionTracingFactory() { - throw new IllegalStateException("Can't instantiate a utility class"); + private ObservableCqlSessionFactory() { + throw new UnsupportedOperationException("Can't instantiate a utility class"); } /** - * Wrap the {@link CqlSession} with a {@link CqlSessionTracingInterceptor}. + * Wrap the {@link CqlSession} with a {@link CqlSessionObservationInterceptor}. * - * @param session - * @param observationRegistry - * @param observationConvention + * @param session must not be {@literal null}. + * @param observationRegistry must not be {@literal null}. * @return */ - public static CqlSession wrap(CqlSession session, ObservationRegistry observationRegistry, - CqlSessionObservationConvention observationConvention) { + public static CqlSession wrap(CqlSession session, ObservationRegistry observationRegistry) { + return wrap(session, "Cassandra", observationRegistry); + } + + /** + * Wrap the {@link CqlSession} with a {@link CqlSessionObservationInterceptor}. + * + * @param session must not be {@literal null}. + * @param remoteServiceName must not be {@literal null}. + * @param observationRegistry must not be {@literal null}. + * @return + */ + public static CqlSession wrap(CqlSession session, String remoteServiceName, ObservationRegistry observationRegistry) { + + Assert.notNull(session, "CqlSession must not be null"); + Assert.notNull(remoteServiceName, "CqlSessionObservationConvention must not be null"); + Assert.notNull(observationRegistry, "ObservationRegistry must not be null"); ProxyFactory proxyFactory = new ProxyFactory(); proxyFactory.setTarget(session); - proxyFactory.addAdvice(new CqlSessionTracingInterceptor(session, observationRegistry, observationConvention)); + proxyFactory.addAdvice(new CqlSessionObservationInterceptor(session, remoteServiceName, observationRegistry)); proxyFactory.addInterface(CqlSession.class); return (CqlSession) proxyFactory.getProxy(); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSession.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSession.java new file mode 100644 index 000000000..276716913 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSession.java @@ -0,0 +1,183 @@ +/* + * Copyright 2022 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 org.springframework.data.cassandra.observability; + +import java.util.Map; +import java.util.Optional; + +import org.springframework.data.cassandra.ReactiveResultSet; +import org.springframework.data.cassandra.ReactiveSession; +import org.springframework.lang.Nullable; + +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.context.DriverContext; +import com.datastax.oss.driver.api.core.cql.PreparedStatement; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import com.datastax.oss.driver.api.core.cql.Statement; +import com.datastax.oss.driver.api.core.metadata.Metadata; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor; +import reactor.core.publisher.Mono; +import reactor.util.context.ContextView; + +/** + * Instrumented {@link ReactiveSession} for observability. + * + * @author Mark Paluch + * @author Marcin Grzejszczak + * @since 4.0 + */ +public class ObservableReactiveSession implements ReactiveSession { + + private final ReactiveSession delegate; + + private final String remoteServiceName; + + private final ObservationRegistry observationRegistry; + + private final CassandraObservationConvention convention = new DefaultCassandraObservationConvention(); + + ObservableReactiveSession(ReactiveSession delegate, String remoteServiceName, + ObservationRegistry observationRegistry) { + this.delegate = delegate; + this.remoteServiceName = remoteServiceName; + this.observationRegistry = observationRegistry; + } + + /** + * Factory method for creation of a {@link ObservableReactiveSession}. + * + * @param session reactive session. + * @param observationRegistry observation registry. + * @return traced representation of a {@link ReactiveSession}. + */ + public static ReactiveSession create(ReactiveSession session, ObservationRegistry observationRegistry) { + return new ObservableReactiveSession(session, "Cassandra", observationRegistry); + } + + /** + * Factory method for creation of a {@link ObservableReactiveSession}. + * + * @param session reactive session. + * @param observationRegistry observation registry. + * @return traced representation of a {@link ReactiveSession}. + */ + public static ReactiveSession create(ReactiveSession session, String remoteServiceName, + ObservationRegistry observationRegistry) { + return new ObservableReactiveSession(session, remoteServiceName, observationRegistry); + } + + @Override + public boolean isClosed() { + return this.delegate.isClosed(); + } + + @Override + public DriverContext getContext() { + return this.delegate.getContext(); + } + + @Override + public Optional getKeyspace() { + return this.delegate.getKeyspace(); + } + + @Override + public Metadata getMetadata() { + return this.delegate.getMetadata(); + } + + @Override + public Mono execute(String cql) { + return execute(SimpleStatement.newInstance(cql)); + } + + @Override + public Mono execute(String cql, Object... objects) { + return execute(SimpleStatement.newInstance(cql, objects)); + } + + @Override + public Mono execute(String cql, Map map) { + return execute(SimpleStatement.newInstance(cql, map)); + } + + @Override + public Mono execute(Statement statement) { + + if (ObservationStatement.isObservationStatement(statement)) { + return this.delegate.execute(statement); + } + + return Mono.deferContextual(contextView -> { + + Observation observation = startObservation(getParentObservation(contextView), statement, false, "execute"); + return this.delegate.execute(ObservationStatement.createProxy(observation, statement)); + }); + } + + @Override + public Mono prepare(String cql) { + return prepare(SimpleStatement.newInstance(cql)); + } + + @Override + public Mono prepare(SimpleStatement statement) { + + if (ObservationStatement.isObservationStatement(statement)) { + return this.delegate.prepare(statement); + } + + // prepare calls do not notify RequestTracker so we need to stop the observation ourselves + return Mono.deferContextual(contextView -> { + + Observation observation = startObservation(getParentObservation(contextView), statement, true, "prepare"); + return this.delegate.prepare(ObservationStatement.createProxy(observation, statement)) // + .doOnError(observation::error) // + .doFinally(ignore -> observation.stop()); + }); + } + + @Override + public void close() { + this.delegate.close(); + } + + private Observation startObservation(@Nullable Observation parent, Statement statement, boolean prepare, + String methodName) { + + Observation observation = Observation + .createNotStarted(methodName, + () -> new CassandraObservationContext(statement, remoteServiceName, prepare, methodName, + delegate.getContext().getSessionName(), + delegate.getKeyspace().map(CqlIdentifier::asInternal).orElse("system")), + observationRegistry) + .observationConvention(convention); + + if (parent != null) { + observation.parentObservation(parent); + } + + return observation.start(); + } + + @Nullable + private static Observation getParentObservation(ContextView contextView) { + return contextView.getOrDefault(ObservationThreadLocalAccessor.KEY, null); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactory.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactory.java new file mode 100644 index 000000000..64f3e926d --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactory.java @@ -0,0 +1,65 @@ +/* + * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; + +import org.springframework.data.cassandra.ReactiveSession; +import org.springframework.util.Assert; + +import com.datastax.oss.driver.api.core.CqlSession; + +import io.micrometer.observation.ObservationRegistry; + +/** + * Factory to wrap a {@link org.springframework.data.cassandra.ReactiveSession} with {@link ObservableReactiveSession}. + * + * @author Mark Paluch + * @since 4.0 + */ +public final class ObservableReactiveSessionFactory { + + private ObservableReactiveSessionFactory() { + throw new UnsupportedOperationException("Can't instantiate a utility class"); + } + + /** + * Wrap the {@link CqlSession} with a {@link CqlSessionObservationInterceptor}. + * + * @param session must not be {@literal null}. + * @param observationRegistry must not be {@literal null}. + * @return + */ + public static ReactiveSession wrap(ReactiveSession session, ObservationRegistry observationRegistry) { + return wrap(session, "Cassandra", observationRegistry); + } + + /** + * Wrap the {@link CqlSession} with a {@link CqlSessionObservationInterceptor}. + * + * @param session must not be {@literal null}. + * @param remoteServiceName must not be {@literal null}. + * @param observationRegistry must not be {@literal null}. + * @return + */ + public static ReactiveSession wrap(ReactiveSession session, String remoteServiceName, + ObservationRegistry observationRegistry) { + + Assert.notNull(session, "CqlSession must not be null"); + Assert.notNull(remoteServiceName, "CqlSessionObservationConvention must not be null"); + Assert.notNull(observationRegistry, "ObservationRegistry must not be null"); + + return ObservableReactiveSession.create(session, remoteServiceName, observationRegistry); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationBeanPostProcessor.java similarity index 56% rename from spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessor.java rename to spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationBeanPostProcessor.java index 6614fec20..0ddd10f94 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessor.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationBeanPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2022 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. @@ -15,41 +15,33 @@ */ package org.springframework.data.cassandra.observability; -import io.micrometer.observation.ObservationRegistry; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.data.cassandra.ReactiveSession; import com.datastax.oss.driver.api.core.CqlSession; -/** - * {@link BeanPostProcessor} to automatically wrap all {@link CqlSession}s with a {@link CqlSessionTracingInterceptor}. - * - * @author Marcin Grzejszczak - * @author Mark Paluch - * @author Greg Turnquist - * @since 4.0.0 - */ -public class CqlSessionTracingBeanPostProcessor implements BeanPostProcessor { +import io.micrometer.observation.ObservationRegistry; - private final ObservationRegistry observationRegistry; +class ObservationBeanPostProcessor implements BeanPostProcessor { - private final CqlSessionObservationConvention observationConvention; - - public CqlSessionTracingBeanPostProcessor(ObservationRegistry observationRegistry, - CqlSessionObservationConvention observationConvention) { + public final ObservationRegistry observationRegistry; + public ObservationBeanPostProcessor(ObservationRegistry observationRegistry) { this.observationRegistry = observationRegistry; - this.observationConvention = observationConvention; } @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { if (bean instanceof CqlSession) { - return CqlSessionTracingFactory.wrap((CqlSession) bean, this.observationRegistry, this.observationConvention); + return ObservableCqlSessionFactory.wrap((CqlSession) bean, observationRegistry); } - return bean; + if (bean instanceof ReactiveSession) { + return ObservableReactiveSessionFactory.wrap((ReactiveSession) bean, observationRegistry); + } + + return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationRequestTracker.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationRequestTracker.java new file mode 100644 index 000000000..738f87a57 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationRequestTracker.java @@ -0,0 +1,145 @@ +/* + * Copyright 2022 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 org.springframework.data.cassandra.observability; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.data.cassandra.observability.CassandraObservation.Events; +import org.springframework.data.cassandra.observability.CassandraObservation.HighCardinalityKeyNames; +import org.springframework.lang.Nullable; + +import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; +import com.datastax.oss.driver.api.core.metadata.Node; +import com.datastax.oss.driver.api.core.session.Request; +import com.datastax.oss.driver.api.core.tracker.RequestTracker; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.Observation.Event; + +/** + * Trace implementation of the {@link RequestTracker}. + * + * @author Mark Paluch + * @author Marcin Grzejszczak + * @since 4.0 + */ +public enum ObservationRequestTracker implements RequestTracker { + + /** + * Singleton instance of the {@link RequestTracker} to complete {@link Observation}s. + */ + INSTANCE; + + private static final Log log = LogFactory.getLog(ObservationRequestTracker.class); + + @Override + public void onSuccess(Request request, long latencyNanos, DriverExecutionProfile executionProfile, Node node, + String requestLogPrefix) { + + if (request instanceof CassandraObservationSupplier) { + + Observation observation = ((CassandraObservationSupplier) request).getObservation(); + + if (log.isDebugEnabled()) { + log.debug("Closing observation [" + observation + "]"); + } + + observation.stop(); + } + } + + @Override + public void onError(Request request, Throwable error, long latencyNanos, DriverExecutionProfile executionProfile, + @Nullable Node node, String requestLogPrefix) { + + if (request instanceof CassandraObservationSupplier) { + + Observation observation = ((CassandraObservationSupplier) request).getObservation(); + observation.error(error); + + if (log.isDebugEnabled()) { + log.debug("Closing observation [" + observation + "]"); + } + + observation.stop(); + } + } + + @Override + public void onNodeError(Request request, Throwable error, long latencyNanos, DriverExecutionProfile executionProfile, + Node node, String requestLogPrefix) { + + if (request instanceof CassandraObservationSupplier) { + + Observation observation = ((CassandraObservationSupplier) request).getObservation(); + + observation.event(Event.of(Events.NODE_SUCCESS.getValue())); + observation.highCardinalityKeyValue( + String.format(HighCardinalityKeyNames.NODE_ERROR_TAG.asString(), node.getEndPoint()), error.toString()); + + tryAddingRemoteIpAndPort(node, observation); + if (log.isDebugEnabled()) { + log.debug("Marking node error for [" + observation + "]"); + } + } + } + + @Override + public void onNodeSuccess(Request request, long latencyNanos, DriverExecutionProfile executionProfile, Node node, + String requestLogPrefix) { + + if (request instanceof CassandraObservationSupplier) { + + Observation observation = ((CassandraObservationSupplier) request).getObservation(); + + observation.event(Event.of(Events.NODE_SUCCESS.getValue())); + tryAddingRemoteIpAndPort(node, observation); + + if (log.isDebugEnabled()) { + log.debug("Marking node success for [" + observation + "]"); + } + } + } + + @Override + public void close() throws Exception { + + } + + private void tryAddingRemoteIpAndPort(Node node, Observation observation) { + try { + SocketAddress socketAddress = node.getEndPoint().resolve(); + String host; + int port; + if (socketAddress instanceof InetSocketAddress) { + InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress; + host = inetSocketAddress.getHostString(); + port = inetSocketAddress.getPort(); + } else { + host = socketAddress.toString(); + port = 0; + } + + // TODO observation.remoteIpAndPort(host, port); + } catch (Exception e) { + log.debug("Exception occurred while trying to set ip and port", e); + } + } + +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationStatement.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationStatement.java new file mode 100644 index 000000000..c53eb44ec --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationStatement.java @@ -0,0 +1,91 @@ +/* + * Copyright 2022 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 org.springframework.data.cassandra.observability; + +import javax.annotation.Nonnull; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.lang.Nullable; +import org.springframework.util.ClassUtils; + +import com.datastax.oss.driver.api.core.cql.Statement; + +import io.micrometer.observation.Observation; + +/** + * Trace implementation of a {@link Statement}. + * + * @author Mark Paluch + * @author Marcin Grzejszczak + * @since 4.0 + */ +final class ObservationStatement implements MethodInterceptor { + + private final Observation observation; + + private Statement delegate; + + private ObservationStatement(Observation observation, Statement delegate) { + this.observation = observation; + this.delegate = delegate; + } + + /** + * Creates a proxy with {@link ObservationStatement} attached to it. + * + * @param observation current observation + * @param target target to proxy + * @param type of target + * @return proxied object with trace advice + */ + @SuppressWarnings("unchecked") + public static T createProxy(Observation observation, T target) { + + ProxyFactory factory = new ProxyFactory(ClassUtils.getAllInterfaces(target)); + + factory.addInterface(CassandraObservationSupplier.class); + factory.setTarget(target); + factory.addAdvice(new ObservationStatement(observation, (Statement) target)); + + return (T) factory.getProxy(); + } + + /** + * @param statement target statement to inspect + * @return whether the given {@code statement} is a traced statement wrapper + */ + public static boolean isObservationStatement(Statement statement) { + return statement instanceof CassandraObservationSupplier; + } + + @Nullable + @Override + public Object invoke(@Nonnull MethodInvocation invocation) throws Throwable { + + if (invocation.getMethod().getName().equals("getObservation")) { + return this.observation; + } + + Object result = invocation.proceed(); + if (result instanceof Statement) { + this.delegate = (Statement) result; + } + return result; + } + +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/package-info.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/package-info.java new file mode 100644 index 000000000..327c03529 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/package-info.java @@ -0,0 +1,5 @@ +/** + * Infrastructure to provide driver observability using Micrometer. + */ +@org.springframework.lang.NonNullApi +package org.springframework.data.cassandra.observability; diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessorTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessorTests.java deleted file mode 100644 index dedbc853e..000000000 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessorTests.java +++ /dev/null @@ -1,105 +0,0 @@ -/* - * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; - -import io.micrometer.observation.ObservationHandler; -import io.micrometer.observation.ObservationRegistry; -import io.micrometer.tracing.Tracer; -import io.micrometer.tracing.test.simple.SimpleTracer; - -import java.lang.reflect.Method; -import java.util.Collection; - -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.springframework.aop.Advisor; -import org.springframework.aop.framework.Advised; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit.jupiter.SpringExtension; -import org.springframework.util.ReflectionUtils; - -import com.datastax.oss.driver.api.core.CqlSession; - -/** - * Verify that {@link CqlSessionTracingBeanPostProcessor} properly wraps {@link CqlSession} beans registered in the app - * context. - * - * @author Greg Turnquist - * @since 4.0.0 - */ -@ExtendWith(SpringExtension.class) -@ContextConfiguration -public class CqlSessionTracingBeanPostProcessorTests { - - @Autowired CqlSession session; - @Autowired ObservationRegistry registry; - @Autowired CqlSessionTracingObservationHandler handler; - - @Test - void injectedCqlSessionShouldBeWrapped() throws Exception { - - assertThat(session).isInstanceOf(CqlSession.class); - assertThat(session).isInstanceOf(Advised.class); - - Advised advised = (Advised) session; - - assertThat(advised.getAdvisors()).extracting(Advisor::getAdvice) - .satisfies(advice -> advice.getClass().equals(CqlSessionTracingInterceptor.class)); - assertThat(advised.getTargetSource().getTarget()).isEqualTo(TestConfig.originalSession); - } - - @Test - void injectedObservationHandlerIsRegisteredWithRegistry() { - - ObservationRegistry.ObservationConfig config = registry.observationConfig(); - - Method getObservationHandlers = ReflectionUtils.findMethod(ObservationRegistry.ObservationConfig.class, - "getObservationHandlers"); - ReflectionUtils.makeAccessible(getObservationHandlers); - Collection> handlers = (Collection>) ReflectionUtils - .invokeMethod(getObservationHandlers, config); - - assertThat(handlers).contains(handler); - } - - @Configuration - @EnableCassandraObservability - static class TestConfig { - - static CqlSession originalSession = mock(CqlSession.class); - - @Bean - CqlSession originalSession() { - return originalSession; - } - - @Bean - ObservationRegistry meterRegistry() { - return ObservationRegistry.create(); - } - - @Bean - Tracer tracer() { - return new SimpleTracer(); - } - } -} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingTests.java index e28ae6dca..16c1702e9 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingTests.java @@ -15,9 +15,27 @@ */ package org.springframework.data.cassandra.observability; -import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.*; import static org.springframework.data.cassandra.observability.CassandraObservation.*; +import java.util.Deque; +import java.util.Map; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.cassandra.config.AbstractSessionConfiguration; +import org.springframework.data.cassandra.test.util.CassandraExtension; +import org.springframework.data.cassandra.test.util.IntegrationTestsSupport; +import org.springframework.test.context.junit.jupiter.SpringExtension; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.cql.Statement; + import io.micrometer.common.KeyValue; import io.micrometer.common.KeyValues; import io.micrometer.common.docs.KeyName; @@ -32,33 +50,24 @@ import io.micrometer.tracing.test.simple.SimpleSpan; import io.micrometer.tracing.test.simple.SimpleTracer; import io.micrometer.tracing.test.simple.SpanAssert; -import java.util.Deque; -import java.util.Map; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Configuration; -import org.springframework.data.cassandra.config.AbstractSessionConfiguration; -import org.springframework.data.cassandra.test.util.CassandraExtension; -import org.springframework.data.cassandra.test.util.IntegrationTestsSupport; - -import com.datastax.oss.driver.api.core.CqlSession; -import com.datastax.oss.driver.api.core.cql.Statement; - /** - * Verify that {@link CqlSessionTracingInterceptor} properly wraps {@link Statement} object with tracing. + * Verify that {@link CqlSessionObservationInterceptor} properly wraps {@link Statement} object with tracing. * * @author Greg Turnquist * @since 4.0.0 */ +@ExtendWith({ SpringExtension.class, CassandraExtension.class }) public class CqlSessionTracingTests extends IntegrationTestsSupport { private static final String CREATE_KEYSPACE = "CREATE KEYSPACE ConfigTest " + "WITH " + "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };"; + private final ObservationRegistry observationRegistry = ObservationRegistry.create(); + + private final MeterRegistry meterRegistry = new SimpleMeterRegistry(); + + private final SimpleTracer tracer = new SimpleTracer(); + private ConfigurableApplicationContext context; private CqlSession session; @@ -67,7 +76,9 @@ public class CqlSessionTracingTests extends IntegrationTestsSupport { void setUp() { this.context = new AnnotationConfigApplicationContext(Config.class); - this.session = context.getBean(CqlSession.class); + this.session = ObservableCqlSessionFactory.wrap(context.getBean(CqlSession.class), "my-cassandra", + observationRegistry); + this.observationRegistry.observationConfig().observationHandler(new DefaultMeterObservationHandler(meterRegistry)); } @AfterEach @@ -75,52 +86,15 @@ public class CqlSessionTracingTests extends IntegrationTestsSupport { this.context.close(); } - @Test - void shouldNotCreateAnyMetricsWhenThereIsNoObservation() { - - MeterRegistry meterRegistry = new SimpleMeterRegistry(); - ObservationRegistry observationRegistry = ObservationRegistry.create(); - observationRegistry.observationConfig().observationHandler(new DefaultMeterObservationHandler(meterRegistry)); - - MeterRegistryAssert.then(meterRegistry).hasNoMetrics(); - } - - @Test - void tracingNoStatementsShouldProduceNoMetrics() { - - MeterRegistry meterRegistry = new SimpleMeterRegistry(); - ObservationRegistry observationRegistry = ObservationRegistry.create(); - observationRegistry.observationConfig().observationHandler(new DefaultMeterObservationHandler(meterRegistry)); - - CqlSessionObservationConvention observationContention = new DefaultCassandraObservationConvention(); - - SimpleTracer tracer = new SimpleTracer(); - observationRegistry.observationConfig().observationHandler(new CqlSessionTracingObservationHandler(tracer)); - - CqlSessionTracingFactory.wrap(session, observationRegistry, observationContention); - - MeterRegistryAssert.then(meterRegistry).hasNoMetrics(); - } - @Test void shouldCreateObservationForCqlSessionOperations() { - MeterRegistry meterRegistry = new SimpleMeterRegistry(); - ObservationRegistry observationRegistry = ObservationRegistry.create(); - observationRegistry.observationConfig().observationHandler(new DefaultMeterObservationHandler(meterRegistry)); - - CqlSessionObservationConvention observationContention = new DefaultCassandraObservationConvention(); - SimpleTracer tracer = new SimpleTracer(); - observationRegistry.observationConfig().observationHandler(new CqlSessionTracingObservationHandler(tracer)); - Observation.start("test", observationRegistry).scoped(() -> { - CqlSession traceSession = CqlSessionTracingFactory.wrap(session, observationRegistry, observationContention); - - traceSession.execute(CREATE_KEYSPACE); - traceSession.executeAsync(CREATE_KEYSPACE); - traceSession.prepare(CREATE_KEYSPACE); - traceSession.prepareAsync(CREATE_KEYSPACE); + session.execute(CREATE_KEYSPACE); + session.executeAsync(CREATE_KEYSPACE); + session.prepare(CREATE_KEYSPACE); + session.prepareAsync(CREATE_KEYSPACE); }); MeterRegistryAssert.then(meterRegistry).hasTimerWithNameAndTags(CASSANDRA_QUERY_OBSERVATION.getName(), KeyValues.of( // diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ImperativeIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ImperativeIntegrationTests.java new file mode 100644 index 000000000..9a40add18 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ImperativeIntegrationTests.java @@ -0,0 +1,83 @@ +/* + * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.cassandra.core.cql.CqlTemplate; +import org.springframework.data.cassandra.test.util.CassandraExtension; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit.jupiter.SpringExtension; + +import com.datastax.oss.driver.api.core.CqlSession; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.tracing.test.SampleTestRunner; + +/** + * Collection of tests that log metrics and tracing. + * + * @author Greg Turnquist + * @author Mark Paluch + */ +@ExtendWith({ SpringExtension.class, CassandraExtension.class }) +@ContextConfiguration(classes = TestConfig.class) +public class ImperativeIntegrationTests extends SampleTestRunner { + + @Autowired CqlSession session; + + ImperativeIntegrationTests() { + super(SampleRunnerConfig.builder().build()); + } + + @Override + protected MeterRegistry createMeterRegistry() { + return TestConfig.METER_REGISTRY; + } + + @Override + protected ObservationRegistry createObservationRegistry() { + return TestConfig.OBSERVATION_REGISTRY; + } + + @Override + public SampleTestRunnerConsumer yourCode() { + + return (tracer, meterRegistry) -> { + + CqlSession observableSession = ObservableCqlSessionFactory.wrap(session, createObservationRegistry()); + + observableSession.execute("DROP KEYSPACE IF EXISTS ObservationTest"); + observableSession.execute("CREATE KEYSPACE ObservationTest " + "WITH " + + "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };"); + observableSession.execute("USE ObservationTest"); + observableSession + .execute("CREATE TABLE IF NOT EXISTS person (id int, firstName text, lastName text, PRIMARY KEY(id));"); + + CqlTemplate template = new CqlTemplate(observableSession); + + template.execute("INSERT INTO person (id,firstName,lastName) VALUES(?,?,?)", 1, "Walter", "White"); + + System.out.println(((SimpleMeterRegistry) meterRegistry).getMetersAsString()); + + assertThat(tracer.getFinishedSpans()).hasSize(6); + }; + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ReactiveIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ReactiveIntegrationTests.java new file mode 100644 index 000000000..f6071b72b --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ReactiveIntegrationTests.java @@ -0,0 +1,94 @@ +/* + * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.cassandra.ReactiveResultSet; +import org.springframework.data.cassandra.ReactiveSession; +import org.springframework.data.cassandra.core.cql.ReactiveCqlTemplate; +import org.springframework.data.cassandra.test.util.CassandraExtension; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit.jupiter.SpringExtension; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor; +import io.micrometer.tracing.test.SampleTestRunner; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; +import reactor.util.context.Context; + +/** + * Collection of tests that log metrics and tracing. + * + * @author Mark Paluch + */ +@ExtendWith({ SpringExtension.class, CassandraExtension.class }) +@ContextConfiguration(classes = TestConfig.class) +public class ReactiveIntegrationTests extends SampleTestRunner { + + @Autowired ReactiveSession session; + + ReactiveIntegrationTests() { + super(SampleRunnerConfig.builder().build()); + } + + @Override + protected MeterRegistry createMeterRegistry() { + return TestConfig.METER_REGISTRY; + } + + @Override + protected ObservationRegistry createObservationRegistry() { + return TestConfig.OBSERVATION_REGISTRY; + } + + @Override + public SampleTestRunnerConsumer yourCode() { + + return (tracer, meterRegistry) -> { + + Observation intermediate = Observation.start("intermediate", createObservationRegistry()); + + ReactiveSession observableSession = ObservableReactiveSessionFactory.wrap(session, createObservationRegistry()); + + Mono drop = observableSession.execute("DROP KEYSPACE IF EXISTS ObservationTest"); + Mono create = observableSession.execute("CREATE KEYSPACE ObservationTest " + "WITH " + + "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };"); + Mono use = observableSession.execute("USE ObservationTest"); + Mono createTable = observableSession + .execute("CREATE TABLE IF NOT EXISTS person (id int, firstName text, lastName text, PRIMARY KEY(id));"); + + ReactiveCqlTemplate template = new ReactiveCqlTemplate(observableSession); + + intermediate.observe(() -> { + drop.then(create).then(use).then(createTable) + .then(template.execute("INSERT INTO person (id,firstName,lastName) VALUES(?,?,?)", 1, "Walter", "White")) + .contextWrite(Context.of(ObservationThreadLocalAccessor.KEY, intermediate)).as(StepVerifier::create) + .expectNextCount(1).verifyComplete(); + }); + + System.out.println(((SimpleMeterRegistry) meterRegistry).getMetersAsString()); + + assertThat(tracer.getFinishedSpans()).hasSize(7); + }; + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/TestConfig.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/TestConfig.java new file mode 100644 index 000000000..f5f113912 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/TestConfig.java @@ -0,0 +1,69 @@ +/* + * Copyright 2022 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 org.springframework.data.cassandra.observability; + +import org.springframework.context.annotation.Bean; +import org.springframework.data.cassandra.ReactiveSession; +import org.springframework.data.cassandra.config.SessionBuilderConfigurer; +import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; +import org.springframework.data.cassandra.support.AbstractTestJavaConfig; +import org.springframework.lang.Nullable; + +import com.datastax.oss.driver.api.core.CqlSession; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.observation.ObservationRegistry; + +/** + * @author Mark Paluch + */ +class TestConfig extends AbstractTestJavaConfig { + + static final MeterRegistry METER_REGISTRY = new SimpleMeterRegistry(); + static final ObservationRegistry OBSERVATION_REGISTRY = ObservationRegistry.create(); + + static { + OBSERVATION_REGISTRY.observationConfig().observationHandler(new DefaultMeterObservationHandler(METER_REGISTRY)); + } + + @Override + protected String getKeyspaceName() { + return "system"; + } + + @Bean + ObservationRegistry registry() { + return OBSERVATION_REGISTRY; + } + + @Bean + CassandraObservationConvention observationContention() { + return new DefaultCassandraObservationConvention(); + } + + @Nullable + @Override + protected SessionBuilderConfigurer getSessionBuilderConfigurer() { + return sessionBuilder -> sessionBuilder.addRequestTracker(ObservationRequestTracker.INSTANCE); + } + + @Bean + ReactiveSession reactiveSession(CqlSession session) { + return new DefaultBridgedReactiveSession(session); + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ZipkinIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ZipkinIntegrationTests.java deleted file mode 100644 index 9018143b3..000000000 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ZipkinIntegrationTests.java +++ /dev/null @@ -1,130 +0,0 @@ -/* - * Copyright 2013-2022 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 org.springframework.data.cassandra.observability; - -import io.micrometer.core.instrument.MeterRegistry; -import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler; -import io.micrometer.core.instrument.simple.SimpleMeterRegistry; -import io.micrometer.observation.Observation; -import io.micrometer.observation.ObservationHandler; -import io.micrometer.observation.ObservationRegistry; -import io.micrometer.tracing.test.SampleTestRunner; -import io.micrometer.tracing.test.reporter.BuildingBlocks; - -import java.util.Deque; -import java.util.function.BiConsumer; - -import org.junit.jupiter.api.extension.ExtendWith; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.annotation.Bean; -import org.springframework.data.cassandra.support.AbstractTestJavaConfig; -import org.springframework.data.cassandra.test.util.CassandraExtension; -import org.springframework.data.cassandra.test.util.TestKeyspace; -import org.springframework.data.cassandra.test.util.TestKeyspaceName; -import org.springframework.test.context.junit.jupiter.SpringExtension; - -import com.datastax.oss.driver.api.core.CqlSession; - -/** - * Collection of tests that log metrics and tracing with an external tracing tool. Since this external tool must be up - * and running after the test is completed, this test is ONLY run manually. - * - * @author Greg Turnquist - * @since 4.0.0 - */ -@ExtendWith({ SpringExtension.class, CassandraExtension.class }) -@TestKeyspaceName -public class ZipkinIntegrationTests extends SampleTestRunner { - - private static final MeterRegistry METER_REGISTRY = new SimpleMeterRegistry(); - private static final ObservationRegistry OBSERVATION_REGISTRY = ObservationRegistry.create(); - - static { - OBSERVATION_REGISTRY.observationConfig().observationHandler(new DefaultMeterObservationHandler(METER_REGISTRY)); - } - - @Autowired CqlSession session; - - ZipkinIntegrationTests() { - super(SampleRunnerConfig.builder().build()); - } - - @Override - protected MeterRegistry createMeterRegistry() { - return METER_REGISTRY; - } - - @Override - protected ObservationRegistry createObservationRegistry() { - return OBSERVATION_REGISTRY; - } - - @Override - public BiConsumer>> customizeObservationHandlers() { - - return (buildingBlocks, observationHandlers) -> observationHandlers - .addLast(new CqlSessionTracingObservationHandler(buildingBlocks.getTracer())); - } - - @Override - public TracingSetup[] getTracingSetup() { - return new TracingSetup[] { TracingSetup.ZIPKIN_BRAVE }; - } - - @Override - public SampleTestRunnerConsumer yourCode() { - - return (tracer, meterRegistry) -> { - - OBSERVATION_REGISTRY.observationConfig() - .observationHandler(new CqlSessionTracingObservationHandler(tracer.getTracer())); - - session.execute("DROP KEYSPACE IF EXISTS ConfigTest"); - session.execute("CREATE KEYSPACE ConfigTest " + "WITH " - + "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };"); - session.execute("USE ConfigTest"); - session.execute("CREATE TABLE IF NOT EXISTS person (id int, firstName text, lastName text, PRIMARY KEY(id));"); - - System.out.println(((SimpleMeterRegistry) meterRegistry).getMetersAsString()); - }; - } - - static class TestConfig extends AbstractTestJavaConfig { - - @TestKeyspace CqlSession session; - - @Override - protected String getKeyspaceName() { - return "system"; - } - - @Bean - ObservationRegistry registry() { - return OBSERVATION_REGISTRY; - } - - @Bean - CqlSessionObservationConvention observationContention() { - return new DefaultCassandraObservationConvention(); - } - - @Bean - CqlSessionTracingBeanPostProcessor traceCqlSessionBeanPostProcessor(ObservationRegistry observationRegistry, - CqlSessionObservationConvention observationConvention) { - return new CqlSessionTracingBeanPostProcessor(observationRegistry, observationConvention); - } - } -} diff --git a/src/main/asciidoc/index.adoc b/src/main/asciidoc/index.adoc index 7b94208a6..39a99bc39 100644 --- a/src/main/asciidoc/index.adoc +++ b/src/main/asciidoc/index.adoc @@ -23,6 +23,7 @@ include::{spring-data-commons-docs}/repositories.adoc[leveloffset=+1] include::reference/introduction.adoc[leveloffset=+1] include::reference/cassandra.adoc[leveloffset=+1] +include::reference/observability.adoc[leveloffset=+1] include::reference/reactive-cassandra.adoc[leveloffset=+1] include::reference/cassandra-repositories.adoc[leveloffset=+1] include::reference/reactive-cassandra-repositories.adoc[leveloffset=+1] @@ -40,4 +41,3 @@ include::{spring-data-commons-docs}/repository-populator-namespace-reference.ado include::{spring-data-commons-docs}/repository-query-keywords-reference.adoc[leveloffset=+1] include::{spring-data-commons-docs}/repository-query-return-types-reference.adoc[leveloffset=+1] include::reference/migration-guides.adoc[leveloffset=+1] -include::reference/observability.adoc[leveloffset=+1] diff --git a/src/main/asciidoc/reference/observability.adoc b/src/main/asciidoc/reference/observability.adoc index 62c85fb47..b21ef81ba 100644 --- a/src/main/asciidoc/reference/observability.adoc +++ b/src/main/asciidoc/reference/observability.adoc @@ -1,84 +1,60 @@ -:root-target: ../../../../../target/ +[[cassandra.observability]] +== Observability -[[observability]] -= Observability metadata +Getting insights from an application component about its operations, timing and relation to application code is crucial to understand latency. +Spring Data Cassandra ships with a Micrometer instrumentation through the Cassandra driver to collect observations during Cassandra interaction. +Once the integration is set up, Micrometer will create meters and spans (for distributed tracing) for each Cassandra statement. -include::{root-target}_conventions.adoc[] +To enable the instrumentation, apply the following configuration to your application: -include::{root-target}_metrics.adoc[] - -include::{root-target}_spans.adoc[] - -[[observability.registration]] -== Observability Registration - -Spring Data Cassandra is not yet supported in Spring Boot to automatically enable observability. -Don't worry, we've got you covered! -Simply add the `@EnableCassandraObservability` annotation to your Spring Boot application, and you'll be all set. - -.Activating observability for your Spring Boot application ==== [source,java] ---- -@SpringBootApplication -@EnableCassandraObservability // <1> -public class SpringDataCassandraObservabilityApplication { +@Configuration +class ObservabilityConfiguration { - public static void main(String[] args) { - SpringApplication.run(SpringDataCassandraObservabilityApplication.class, args); - } + @Bean + public ObservationBeanPostProcessor observationBeanPostProcessor(ObservationRegistry observationRegistry) { + return new ObservationBeanPostProcessor(observationRegistry); <1> + } + + @Bean + public SessionBuilderConfigurer getSessionBuilderConfigurer() { + return sessionBuilder -> sessionBuilder.addRequestTracker(ObservationRequestTracker.INSTANCE); <2> + } + + class ObservationBeanPostProcessor implements BeanPostProcessor { + + public final ObservationRegistry observationRegistry; + + public ObservationBeanPostProcessor(ObservationRegistry observationRegistry) { + this.observationRegistry = observationRegistry; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + + if (bean instanceof CqlSession) { + return ObservableCqlSessionFactory.wrap((CqlSession) bean, observationRegistry); + } + + if (bean instanceof ReactiveSession) { + return ObservableReactiveSessionFactory.wrap((ReactiveSession) bean, observationRegistry); + } + + return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName); + } + } } ---- -<1> This annotation will activate the bits needed start wrapping CQL calls and register them with your tracer of choice. +<1> Wraps all CQL session objects (imperative/reactive Session API) to observe Cassandra statement execution. +<2> Integrate with the Cassandra driver to obtain success/error callbacks. ==== -By the way, Spring Boot DOES have autoconfigured hooks into various parts of the system. -For example, there is an observation filter that Spring Boot will apply to Spring MVC ensuring all your calls are wrapped properly. -And if anywhere in the midst of that web call, you invoke Spring Data Cassandra (either through `CqlTemplate` or a custom repository), it will get captured properly. +include::../../../../target/_conventions.adoc[] -Something that is NOT covered are situations where your code runs independently. -For example, if you have some block that is run during startup inside a `CommandLineRunner`, there is no way for Spring Boot to know that this should observed. -Check out the code block below: +include::../../../../target/_metrics.adoc[] -.Loading data for a sample application -==== -[source,java] ----- -@Bean -CommandLineRunner initData(EmployeeRepository repository) { - return args -> { - repository.save(new Employee("1", "Frodo", "ring bearer")); - repository.save(new Employee("2", "Bilbo", "burglar")); - }; -} ----- -This tactic is used all the time in demos. These calls to a Spring Data Cassandra repository will NOT be observed. -==== +include::../../../../target/_spans.adoc[] -If you DO want to observe such a code block, you must wrap it yourself, as shown below: - -.Observing the loading of data in a sample application -==== -[source,java] ----- -@Bean -CommandLineRunner initData(EmployeeRepository repository, ObservationRegistry registry) { // <1> - return args -> { - Observation.createNotStarted("init-database", registry).observe(() -> { // <2> - repository.save(new Employee("1", "Frodo", "ring bearer")); - repository.save(new Employee("2", "Bilbo", "burglar")); - }); - }; -} ----- -<1> Your `CommandLineRunner` requires access to the app context's `ObservationRegistry` -<2> You need to create your own `Observation` using the `createNotStarted()` method. Give it any contextual name you like, but be sure to include the `registry`. -==== - -The `observe()` method takes a Java 8 lambda function which is invoked inside a common Micrometer pattern of: - -* Starting the observation. -* Invoking your callback. -* Properly ending the observation by either reporting an error if it fails or stopping the observation if it succeeds. - -This will allow you to observe chunks of code that may fall outside of currently autoconfigured operations. +See also https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/database/#Cassandra[OpenTelemetry Semantic Conventions] for further reference.