Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing.

Browse Source 
Add request tracker integration to capture completion/error responses. Add reactive integration. Properly observe prepare requests. Simplify documentation.

See #1321
Original pull request: #1322
This commit is contained in:
Mark Paluch
2022-10-24 11:24:37 +02:00
parent a42bc734ea
commit 786cf3cc6d
27 changed files with 1189 additions and 798 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-data-cassandra-distribution/pom.xml
View File

@@ -42,48 +42,30 @@
<artifactId>exec-maven-plugin</artifactId>
<executions>
<execution>
<id>generate-metrics-metadata</id>
<id>generate-docs</id>
<phase>generate-resources</phase>
<goals>
<goal>java</goal>
</goals>
<configuration>
<mainClass>io.micrometer.docs.metrics.DocsFromSources</mainClass>
</configuration>
</execution>
<execution>
<id>generate-tracing-metadata</id>
<phase>generate-resources</phase>
<goals>
<goal>java</goal>
</goals>
<configuration>
<mainClass>io.micrometer.docs.spans.DocsFromSources</mainClass>
<mainClass>io.micrometer.docs.DocsGeneratorCommand</mainClass>
<includePluginDependencies>true</includePluginDependencies>
<arguments>
<argument>${micrometer-docs-generator.inputPath}</argument>
<argument>${micrometer-docs-generator.inclusionPattern}</argument>
<argument>${micrometer-docs-generator.outputPath}</argument>
</arguments>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-docs-generator-spans</artifactId>
<version>${micrometer-docs-generator}</version>
<type>jar</type>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-docs-generator-metrics</artifactId>
<artifactId>micrometer-docs-generator</artifactId>
<version>${micrometer-docs-generator}</version>
<type>jar</type>
</dependency>
</dependencies>
<configuration>
<includePluginDependencies>true</includePluginDependencies>
<arguments>
<argument>${micrometer-docs-generator.inputPath}</argument>
<argument>${micrometer-docs-generator.inclusionPattern}</argument>
<argument>${micrometer-docs-generator.outputPath}</argument>
</arguments>
</configuration>
</plugin>
<plugin>
<groupId>org.asciidoctor</groupId>

7
spring-data-cassandra/pom.xml
View File

@@ -77,6 +77,7 @@
<artifactId>micrometer-observation</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing</artifactId>
@@ -87,6 +88,12 @@
<groupId>io.micrometer</groupId>
<artifactId>micrometer-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>com.github.tomakehurst</groupId>
<artifactId>wiremock-jre8-standalone</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>

11
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java
View File

@@ -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<String> getStartupScripts() {

49
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservation.java
View File

@@ -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";
}
}
}
}

52
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConfiguration.java
View File

@@ -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> 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);
}
}

47
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionContext.java → spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationContext.java
View File

@@ -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<Object> {
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;
}
}

8
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationConvention.java → spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationConvention.java
View File

@@ -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<CqlSessionContext> {
public interface CassandraObservationConvention extends ObservationConvention<CassandraObservationContext> {
@Override
default boolean supportsContext(Observation.Context context) {
return context instanceof CqlSessionContext;
return context instanceof CassandraObservationContext;
}
}

26
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/EnableCassandraObservability.java → spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationSupplier.java
View File

@@ -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();
}

186
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.java Normal file
View File

@@ -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<Statement<?>, 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();
}
}

153
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingInterceptor.java
View File

@@ -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<Statement<?>, 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();
}
}

102
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingObservationHandler.java
View File

@@ -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<CqlSessionContext> {
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;
}
}

53
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraObservationConvention.java
View File

@@ -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;
}
}

43
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingFactory.java → spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactory.java
View File

@@ -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();

183
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSession.java Normal file
View File

@@ -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<CqlIdentifier> getKeyspace() {
return this.delegate.getKeyspace();
}
@Override
public Metadata getMetadata() {
return this.delegate.getMetadata();
}
@Override
public Mono<ReactiveResultSet> execute(String cql) {
return execute(SimpleStatement.newInstance(cql));
}
@Override
public Mono<ReactiveResultSet> execute(String cql, Object... objects) {
return execute(SimpleStatement.newInstance(cql, objects));
}
@Override
public Mono<ReactiveResultSet> execute(String cql, Map<String, Object> map) {
return execute(SimpleStatement.newInstance(cql, map));
}
@Override
public Mono<ReactiveResultSet> 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<PreparedStatement> prepare(String cql) {
return prepare(SimpleStatement.newInstance(cql));
}
@Override
public Mono<PreparedStatement> 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);
}
}

65
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactory.java Normal file
View File

@@ -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);
}
}

32
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessor.java → spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationBeanPostProcessor.java
View File

@@ -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);
}
}

145
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationRequestTracker.java Normal file
View File

@@ -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);
}
}
}

91
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationStatement.java Normal file
View File

@@ -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 <T> type of target
* @return proxied object with trace advice
*/
@SuppressWarnings("unchecked")
public static <T> 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;
}
}

5
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/package-info.java Normal file
View File

@@ -0,0 +1,5 @@
/**
* Infrastructure to provide driver observability using Micrometer.
*/
@org.springframework.lang.NonNullApi
package org.springframework.data.cassandra.observability;

105
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessorTests.java
View File

@@ -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<ObservationHandler<?>> handlers = (Collection<ObservationHandler<?>>) 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();
}
}
}

94
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingTests.java
View File

@@ -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( //

83
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ImperativeIntegrationTests.java Normal file
View File

@@ -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);
};
}
}

94
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ReactiveIntegrationTests.java Normal file
View File

@@ -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<ReactiveResultSet> drop = observableSession.execute("DROP KEYSPACE IF EXISTS ObservationTest");
Mono<ReactiveResultSet> create = observableSession.execute("CREATE KEYSPACE ObservationTest " + "WITH "
+ "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };");
Mono<ReactiveResultSet> use = observableSession.execute("USE ObservationTest");
Mono<ReactiveResultSet> 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);
};
}
}

69
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/TestConfig.java Normal file
View File

@@ -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);
}
}

130
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ZipkinIntegrationTests.java
View File

@@ -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<BuildingBlocks, Deque<ObservationHandler<? extends Observation.Context>>> 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);
}
}
}

2
src/main/asciidoc/index.adoc
View File

@@ -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]

116
src/main/asciidoc/reference/observability.adoc
View File

@@ -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.