diff --git a/pom.xml b/pom.xml index e6ff770f2..ec9a8fbb9 100644 --- a/pom.xml +++ b/pom.xml @@ -354,7 +354,7 @@ external-cassandra external - 9042 + 9042 9160 7001 7000 @@ -369,7 +369,7 @@ testcontainers - 0 + 0 0 0 0 diff --git a/spring-data-cassandra/pom.xml b/spring-data-cassandra/pom.xml index 138d200ac..75e938030 100644 --- a/spring-data-cassandra/pom.xml +++ b/spring-data-cassandra/pom.xml @@ -71,6 +71,36 @@ java-driver-query-builder + + + io.micrometer + micrometer-observation + true + + + io.micrometer + micrometer-tracing-api + true + + + + io.micrometer + micrometer-test + test + + + + io.micrometer + micrometer-tracing-test + test + + + + io.micrometer + micrometer-tracing-integration-test + test + + io.projectreactor 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 new file mode 100644 index 000000000..43e31b0a4 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservation.java @@ -0,0 +1,127 @@ +/* + * 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.docs.TagKey; +import io.micrometer.observation.docs.DocumentedObservation; + +/** + * Cassandra-based implementation of {@link DocumentedObservation}. + * + * @author Mark Paluch + * @author Marcin Grzejszczak + * @author Greg Turnquist + * @since 4.0.0 + */ +enum CassandraObservation implements DocumentedObservation { + + /** + * Create an {@link io.micrometer.observation.Observation} for Cassandra-based queries. + */ + CASSANDRA_QUERY_OBSERVATION { + + @Override + public String getName() { + return "spring.data.cassandra.query"; + } + + @Override + public String getContextualName() { + return "query"; + } + + @Override + public TagKey[] getLowCardinalityTagKeys() { + return LowCardinalityTags.values(); + } + + @Override + public TagKey[] getHighCardinalityTagKeys() { + return HighCardinalityTags.values(); + } + + @Override + public String getPrefix() { + return "spring.data.cassandra."; + } + }; + + enum LowCardinalityTags implements TagKey { + + /** + * Name of the Cassandra keyspace. + */ + KEYSPACE_NAME { + @Override + public String getKey() { + return "spring.data.cassandra.keyspace"; + } + }, + + /** + * Cassandra session + */ + SESSION_NAME { + @Override + public String getKey() { + return "spring.data.cassandra.sessionName"; + } + }, + + /** + * The method name + */ + METHOD_NAME { + @Override + public String getKey() { + return "spring.data.cassandra.methodName"; + } + }, + + /** + * Cassandra URL + */ + URL { + @Override + public String getKey() { + return "spring.data.cassandra.url"; + } + }, + + /** + * A tag containing error that occurred for the given node. + */ + NODE_ERROR_TAG { + @Override + public String getKey() { + return "spring.data.cassandra.node[%s].error"; + } + } + } + + enum HighCardinalityTags implements TagKey { + + /** + * A tag containing Cassandra CQL. + */ + CQL_TAG { + @Override + public String getKey() { + return "spring.data.cassandra.cql"; + } + } + } +} 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/CqlSessionContext.java new file mode 100644 index 000000000..f4852ec28 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionContext.java @@ -0,0 +1,57 @@ +/* + * 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 org.springframework.lang.Nullable; + +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.cql.Statement; + +/** + * A {@link Observation.Context} for {@link CqlSession}. + * + * @author Greg Turnquist + * @since 4.0.0 + */ +public class CqlSessionContext extends Observation.Context { + + private final @Nullable Statement statement; + private final String methodName; + private final @Nullable CqlSession delegateSession; + + public CqlSessionContext(@Nullable Statement statement, String methodName, @Nullable CqlSession delegateSession) { + + this.statement = statement; + this.methodName = methodName; + this.delegateSession = delegateSession; + } + + @Nullable + public Statement getStatement() { + return statement; + } + + public String getMethodName() { + return methodName; + } + + @Nullable + public CqlSession getDelegateSession() { + return delegateSession; + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTagsProvider.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTagsProvider.java new file mode 100644 index 000000000..48bd0f4ad --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTagsProvider.java @@ -0,0 +1,32 @@ +/* + * 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; + +/** + * {@link Observation.TagsProvider} for Cassandra. + * + * @author Greg Turnquist + * @since 4.0.0 + */ +public interface CqlSessionTagsProvider extends Observation.TagsProvider { + + @Override + default boolean supportsContext(Observation.Context context) { + return context instanceof CqlSessionContext; + } +} 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/CqlSessionTracingBeanPostProcessor.java new file mode 100644 index 000000000..12d692d26 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessor.java @@ -0,0 +1,55 @@ +/* + * 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 org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanPostProcessor; + +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 { + + private final ObservationRegistry observationRegistry; + + private final CqlSessionTagsProvider tagsProvider; + + public CqlSessionTracingBeanPostProcessor(ObservationRegistry observationRegistry, + CqlSessionTagsProvider tagsProvider) { + + this.observationRegistry = observationRegistry; + this.tagsProvider = tagsProvider; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + + if (bean instanceof CqlSession) { + return CqlSessionTracingFactory.wrap((CqlSession) bean, this.observationRegistry, this.tagsProvider); + } + + return bean; + } +} 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/CqlSessionTracingFactory.java new file mode 100644 index 000000000..e18130a01 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingFactory.java @@ -0,0 +1,56 @@ +/* + * 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 org.springframework.aop.framework.ProxyFactory; + +import com.datastax.oss.driver.api.core.CqlSession; + +/** + * Factory to wrap a {@link CqlSession} with a {@link CqlSessionTracingInterceptor}. + * + * @author Mark Paluch + * @author Greg Turnquist + * @since 4.0.0 + */ +public final class CqlSessionTracingFactory { + + private CqlSessionTracingFactory() { + throw new IllegalStateException("Can't instantiate a utility class."); + } + + /** + * Wrap the {@link CqlSession} with a {@link CqlSessionTracingInterceptor}. + * + * @param session + * @param observationRegistry + * @param tagsProvider + * @return + */ + public static CqlSession wrap(CqlSession session, ObservationRegistry observationRegistry, + CqlSessionTagsProvider tagsProvider) { + + ProxyFactory proxyFactory = new ProxyFactory(); + + proxyFactory.setTarget(session); + proxyFactory.addAdvice(new CqlSessionTracingInterceptor(session, observationRegistry, tagsProvider)); + proxyFactory.addInterface(CqlSession.class); + + return (CqlSession) proxyFactory.getProxy(); + } +} 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 new file mode 100644 index 000000000..c467e25b2 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingInterceptor.java @@ -0,0 +1,165 @@ +/* + * 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, Observation.TagsProviderAware { + + private static final Log log = LogFactory.getLog(CqlSessionTracingInterceptor.class); + + private final CqlSession delegateSession; + + private final ObservationRegistry observationRegistry; + + private CqlSessionTagsProvider tagsProvider; + + CqlSessionTracingInterceptor(CqlSession delegateSession, ObservationRegistry observationRegistry, + CqlSessionTagsProvider tagsProvider) { + + this.delegateSession = delegateSession; + this.observationRegistry = observationRegistry; + this.tagsProvider = tagsProvider; + } + + @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 null; + } + + Observation observation = childObservation(statement, methodName, this.delegateSession); + + if (log.isDebugEnabled()) { + log.debug("Created a new child observation before query [" + observation + "]"); + } + + try (Observation.Scope scope = observation.openScope()) { + return statementExecutor.apply(statement); + } catch (Exception e) { + observation.error(e); + throw e; + } finally { + observation.stop(); + } + } + + /** + * 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()) // + .tagsProvider(this.tagsProvider) // + .start(); + } + + @Override + public void setTagsProvider(CqlSessionTagsProvider tagsProvider) { + this.tagsProvider = tagsProvider; + } +} 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 new file mode 100644 index 000000000..63a53194f --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionTracingObservationHandler.java @@ -0,0 +1,102 @@ +/* + * 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.Tag; +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 (Tag tag : context.getLowCardinalityTags()) { + + if (tag.getKey().equals(CassandraObservation.LowCardinalityTags.SESSION_NAME.getKey())) { + sessionName = tag.getValue(); + } + + if (tag.getKey().equals(CassandraObservation.LowCardinalityTags.URL.getKey())) { + url = tag.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/DefaultCassandraTagsProvider.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraTagsProvider.java new file mode 100644 index 000000000..89e98960b --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraTagsProvider.java @@ -0,0 +1,107 @@ +/* + * 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.Tags; + +import java.util.Optional; +import java.util.StringJoiner; + +import com.datastax.oss.driver.api.core.CqlIdentifier; +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; + +/** + * Default {@link CqlSessionTagsProvider} implementation. + * + * @author Greg Turnquist + * @since 4.0.0 + */ +public class DefaultCassandraTagsProvider implements CqlSessionTagsProvider { + + @Override + public Tags getLowCardinalityTags(CqlSessionContext context) { + + Tags tags = Tags.of( // + CassandraObservation.LowCardinalityTags.SESSION_NAME + .of(Optional.ofNullable(context.getDelegateSession().getName()).orElse("unknown")), + CassandraObservation.LowCardinalityTags.KEYSPACE_NAME.of( + Optional.ofNullable(context.getStatement().getKeyspace()).map(CqlIdentifier::asInternal).orElse("unknown")), + CassandraObservation.LowCardinalityTags.METHOD_NAME.of(context.getMethodName())); + + if (context.getStatement().getNode() != null) { + tags = tags.and(CassandraObservation.LowCardinalityTags.URL + .of(context.getStatement().getNode().getEndPoint().resolve().toString())); + } + + return tags; + } + + @Override + public Tags getHighCardinalityTags(CqlSessionContext context) { + return Tags.of(CassandraObservation.HighCardinalityTags.CQL_TAG.of(getCql(context.getStatement()))); + } + + /** + * Extract the CQL query from the delegate {@link Statement}. + * + * @return string-based CQL of the delegate + * @param statement + */ + private static String getCql(Statement statement) { + + String query = ""; + + if (statement instanceof SimpleStatement) { + query = getQuery(statement); + } + + if (statement instanceof BatchStatement) { + + StringJoiner joiner = new StringJoiner(";"); + + for (BatchableStatement bs : (BatchStatement) statement) { + joiner.add(getQuery(bs)); + } + + query = joiner.toString(); + } + + return query; + } + + /** + * Extract the query from a {@link Statement}. + * + * @param statement + * @return query + */ + private static String getQuery(Statement statement) { + + if (statement instanceof SimpleStatement) { + return ((SimpleStatement) statement).getQuery(); + } + + if (statement instanceof BoundStatement) { + return ((BoundStatement) statement).getPreparedStatement().getQuery(); + } + + return ""; + } +} 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 new file mode 100644 index 000000000..8eeefa80a --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingBeanPostProcessorTests.java @@ -0,0 +1,87 @@ +/* + * 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 static org.mockito.Mockito.*; + +import io.micrometer.observation.ObservationRegistry; + +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 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; + + @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); + } + + @Configuration + static class TestConfig { + + static CqlSession originalSession = mock(CqlSession.class); + + @Bean + CqlSession originalSession() { + return originalSession; + } + + @Bean + ObservationRegistry meterRegistry() { + return ObservationRegistry.create(); + } + + @Bean + CqlSessionTagsProvider tagsProvider() { + return new DefaultCassandraTagsProvider(); + } + + @Bean + CqlSessionTracingBeanPostProcessor traceCqlSessionBeanPostProcessor(ObservationRegistry observationRegistry, + CqlSessionTagsProvider tagsProvider) { + return new CqlSessionTracingBeanPostProcessor(observationRegistry, tagsProvider); + } + } +} 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 new file mode 100644 index 000000000..1226e7d69 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/CqlSessionTracingTests.java @@ -0,0 +1,182 @@ +/* + * 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 static org.springframework.data.cassandra.observability.CassandraObservation.*; + +import io.micrometer.common.Tag; +import io.micrometer.common.Tags; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.observation.TimerObservationHandler; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.core.tck.MeterRegistryAssert; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.tracing.Span; +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. + * + * @author Greg Turnquist + * @since 4.0.0 + */ +public class CqlSessionTracingTests extends IntegrationTestsSupport { + + private static final String CREATE_KEYSPACE = "CREATE KEYSPACE ConfigTest " + "WITH " + + "REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };"; + + private ConfigurableApplicationContext context; + + private CqlSession session; + + @BeforeEach + void setUp() { + + this.context = new AnnotationConfigApplicationContext(Config.class); + this.session = context.getBean(CqlSession.class); + } + + @AfterEach + void tearDown() { + this.context.close(); + } + + @Test + void shouldNotCreateAnyMetricsWhenThereIsNoObservation() { + + MeterRegistry meterRegistry = new SimpleMeterRegistry(); + ObservationRegistry observationRegistry = ObservationRegistry.create(); + observationRegistry.observationConfig().observationHandler(new TimerObservationHandler(meterRegistry)); + + MeterRegistryAssert.then(meterRegistry).hasNoMetrics(); + } + + @Test + void tracingNoStatementsShouldProduceNoMetrics() { + + MeterRegistry meterRegistry = new SimpleMeterRegistry(); + ObservationRegistry observationRegistry = ObservationRegistry.create(); + observationRegistry.observationConfig().observationHandler(new TimerObservationHandler(meterRegistry)); + + CqlSessionTagsProvider tagsProvider = new DefaultCassandraTagsProvider(); + + SimpleTracer tracer = new SimpleTracer(); + observationRegistry.observationConfig().observationHandler(new CqlSessionTracingObservationHandler(tracer)); + + CqlSessionTracingFactory.wrap(session, observationRegistry, tagsProvider); + + MeterRegistryAssert.then(meterRegistry).hasNoMetrics(); + } + + @Test + void shouldCreateObservationForCqlSessionOperations() { + + MeterRegistry meterRegistry = new SimpleMeterRegistry(); + ObservationRegistry observationRegistry = ObservationRegistry.create(); + observationRegistry.observationConfig().observationHandler(new TimerObservationHandler(meterRegistry)); + + CqlSessionTagsProvider tagsProvider = new DefaultCassandraTagsProvider(); + SimpleTracer tracer = new SimpleTracer(); + observationRegistry.observationConfig().observationHandler(new CqlSessionTracingObservationHandler(tracer)); + + Observation.start("test", observationRegistry).scoped(() -> { + + CqlSession traceSession = CqlSessionTracingFactory.wrap(session, observationRegistry, tagsProvider); + + traceSession.execute(CREATE_KEYSPACE); + traceSession.executeAsync(CREATE_KEYSPACE); + traceSession.prepare(CREATE_KEYSPACE); + traceSession.prepareAsync(CREATE_KEYSPACE); + }); + + MeterRegistryAssert.then(meterRegistry).hasTimerWithNameAndTags(CASSANDRA_QUERY_OBSERVATION.getName(), Tags.of( // + LowCardinalityTags.SESSION_NAME.of("s5"), // + LowCardinalityTags.KEYSPACE_NAME.of("unknown"), // + Tag.of("error", "none") // + )); + + assertThat(tracer.getSpans()).hasSize(4); + + assertThat(findSpan(tracer.getSpans(), LowCardinalityTags.METHOD_NAME.getKey(), "execute")).isNotNull(); + assertThat(findSpan(tracer.getSpans(), LowCardinalityTags.METHOD_NAME.getKey(), "executeAsync")).isNotNull(); + assertThat(findSpan(tracer.getSpans(), LowCardinalityTags.METHOD_NAME.getKey(), "prepare")).isNotNull(); + assertThat(findSpan(tracer.getSpans(), LowCardinalityTags.METHOD_NAME.getKey(), "prepareAsync")).isNotNull(); + + tracer.getSpans().forEach(simpleSpan -> SpanAssert.then(simpleSpan) // + .hasRemoteServiceNameEqualTo("cassandra-s5") // + .hasNameEqualTo(CASSANDRA_QUERY_OBSERVATION.getContextualName()) // + .hasTag(LowCardinalityTags.SESSION_NAME.getKey(), "s5") // + .hasTag(LowCardinalityTags.KEYSPACE_NAME.getKey(), "unknown") // + .hasTag(HighCardinalityTags.CQL_TAG.getKey(), CREATE_KEYSPACE) // + .hasIpThatIsBlank() // + .hasPortEqualTo(0) // + .hasKindEqualTo(Span.Kind.CLIENT)); + } + + /** + * Find a {@link Span} with a specific key and value. + * + * @param spans + * @param key + * @param value + * @return + */ + private SimpleSpan findSpan(Deque spans, String key, String value) { + + return spans.stream() // + .filter(simpleSpan -> { + Map tags = simpleSpan.getTags(); + return tags.containsKey(key) && tags.get(key).equals(value); + }) // + .findAny() // + .orElse(null); + + } + + @Configuration + static class Config extends AbstractSessionConfiguration { + + @Override + protected String getKeyspaceName() { + return "system"; + } + + @Override + protected int getPort() { + return CassandraExtension.getResources().getPort(); + } + } +} 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 new file mode 100644 index 000000000..1ba605bbf --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ZipkinIntegrationTests.java @@ -0,0 +1,119 @@ +/* + * 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.TimerObservationHandler; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +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.Disabled; +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 + */ +@Disabled("Run this manually to visually test spans in Zipkin") +@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 TimerObservationHandler(METER_REGISTRY)); + } + + @Autowired CqlSession session; + + ZipkinIntegrationTests() { + super(SampleRunnerConfig.builder().build(), OBSERVATION_REGISTRY, METER_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) -> { + + 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 + CqlSessionTagsProvider tagsProvider() { + return new DefaultCassandraTagsProvider(); + } + + @Bean + CqlSessionTracingBeanPostProcessor traceCqlSessionBeanPostProcessor(ObservationRegistry observationRegistry, + CqlSessionTagsProvider tagsProvider) { + return new CqlSessionTracingBeanPostProcessor(observationRegistry, tagsProvider); + } + } +}