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

Allow configuration of CassandraObservationConvention.

Browse Source 
Closes #1490
This commit is contained in:
Mark Paluch
2024-08-21 10:54:49 +02:00
parent 8010a78f34
commit cdc15e1fd0
11 changed files with 175 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -27,7 +27,7 @@ import io.micrometer.tracing.docs.EventValue;
* @author Greg Turnquist
* @since 4.0
*/
enum CassandraObservation implements ObservationDocumentation {
public enum CassandraObservation implements ObservationDocumentation {
/**
* Create an {@link io.micrometer.observation.Observation} for Cassandra-based queries.

9
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.java
View File

@@ -50,13 +50,14 @@ final class CqlSessionObservationInterceptor implements MethodInterceptor {
private final ObservationRegistry observationRegistry;
private final CassandraObservationConvention observationConvention = new DefaultCassandraObservationConvention();
private final CassandraObservationConvention convention;
CqlSessionObservationInterceptor(CqlSession delegate, String remoteServiceName,
ObservationRegistry observationRegistry) {
CassandraObservationConvention convention, ObservationRegistry observationRegistry) {
this.delegate = delegate;
this.remoteServiceName = remoteServiceName;
this.convention = convention;
this.observationRegistry = observationRegistry;
}
@@ -140,7 +141,7 @@ final class CqlSessionObservationInterceptor implements MethodInterceptor {
return SimpleStatement.newInstance((String) args[0]);
}
if (args[0]instanceof String query && args.length == 2) {
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]);
@@ -179,7 +180,7 @@ final class CqlSessionObservationInterceptor implements MethodInterceptor {
delegate.getContext().getSessionName(),
delegate.getKeyspace().map(CqlIdentifier::asInternal).orElse("system")),
observationRegistry)
.observationConvention(observationConvention);
.observationConvention(convention);
if (currentObservation != null) {
observation.parentObservation(currentObservation);

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

@@ -41,7 +41,9 @@ import com.datastax.oss.driver.api.core.metadata.Node;
* @author Mark Paluch
* @since 4.0
*/
class DefaultCassandraObservationConvention implements CassandraObservationConvention {
public class DefaultCassandraObservationConvention implements CassandraObservationConvention {
public static final CassandraObservationConvention INSTANCE = new DefaultCassandraObservationConvention();
@Override
public KeyValues getLowCardinalityKeyValues(CassandraObservationContext context) {
@@ -104,10 +106,10 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
}
@Nullable
private InetSocketAddress tryGetSocketAddress(EndPoint endPoint) {
protected InetSocketAddress tryGetSocketAddress(EndPoint endPoint) {
try {
if (endPoint.resolve()instanceof InetSocketAddress inet) {
if (endPoint.resolve() instanceof InetSocketAddress inet) {
return inet;
}
@@ -121,13 +123,28 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
return (context.isPrepare() ? "PREPARE: " : "") + getOperationName(getCql(context.getStatement()), "");
}
/**
* Tries to parse the CQL query or provides the default name.
*
* @param defaultName if there's no query
* @return span name
*/
public String getOperationName(String cql, String defaultName) {
if (StringUtils.hasText(cql) && cql.indexOf(' ') > -1) {
return cql.substring(0, cql.indexOf(' '));
}
return defaultName;
}
/**
* Extract the CQL query from the delegate {@link Statement}.
*
* @return string-based CQL of the delegate
* @param statement
*/
private static String getCql(Statement<?> statement) {
protected static String getCql(Statement<?> statement) {
String query = "";
@@ -155,7 +172,7 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
* @param statement
* @return query
*/
private static String getQuery(Statement<?> statement) {
protected static String getQuery(Statement<?> statement) {
if (statement instanceof SimpleStatement) {
return ((SimpleStatement) statement).getQuery();
@@ -167,19 +184,4 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
return "";
}
/**
* Tries to parse the CQL query or provides the default name.
*
* @param defaultName if there's no query
* @return span name
*/
public String getOperationName(String cql, String defaultName) {
if (StringUtils.hasText(cql) && cql.indexOf(' ') > -1) {
return cql.substring(0, cql.indexOf(' '));
}
return defaultName;
}
}

24
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactory.java
View File

@@ -17,8 +17,6 @@ package org.springframework.data.cassandra.observability;
import io.micrometer.observation.ObservationRegistry;
import org.springframework.aop.RawTargetAccess;
import org.springframework.aop.TargetSource;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.data.cassandra.observability.CqlSessionObservationInterceptor.ObservationDecoratedProxy;
import org.springframework.util.Assert;
@@ -58,20 +56,36 @@ public final class ObservableCqlSessionFactory {
* @return
*/
public static CqlSession wrap(CqlSession session, String remoteServiceName, ObservationRegistry observationRegistry) {
return wrap(session, remoteServiceName, DefaultCassandraObservationConvention.INSTANCE, observationRegistry);
}
/**
* Wrap the {@link CqlSession} with a {@link CqlSessionObservationInterceptor}.
*
* @param session must not be {@literal null}.
* @param remoteServiceName must not be {@literal null}.
* @param convention the observation convention.
* @param observationRegistry must not be {@literal null}.
* @return
* @since 4.3.4
*/
public static CqlSession wrap(CqlSession session, String remoteServiceName, CassandraObservationConvention convention,
ObservationRegistry observationRegistry) {
Assert.notNull(session, "CqlSession must not be null");
Assert.notNull(remoteServiceName, "CqlSessionObservationConvention must not be null");
Assert.notNull(remoteServiceName, "Remote service name must not be null");
Assert.notNull(convention, "CassandraObservationConvention must not be null");
Assert.notNull(observationRegistry, "ObservationRegistry must not be null");
ProxyFactory proxyFactory = new ProxyFactory();
proxyFactory.setTarget(session);
proxyFactory.addAdvice(new CqlSessionObservationInterceptor(session, remoteServiceName, observationRegistry));
proxyFactory
.addAdvice(new CqlSessionObservationInterceptor(session, remoteServiceName, convention, observationRegistry));
proxyFactory.addInterface(CqlSession.class);
proxyFactory.addInterface(ObservationDecoratedProxy.class);
return (CqlSession) proxyFactory.getProxy();
}
}

46
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactoryBean.java
View File

@@ -43,6 +43,8 @@ public class ObservableCqlSessionFactoryBean extends AbstractFactoryBean<CqlSess
private @Nullable String remoteServiceName;
private CassandraObservationConvention convention = DefaultCassandraObservationConvention.INSTANCE;
/**
* Construct a new {@link ObservableCqlSessionFactoryBean}.
*
@@ -58,16 +60,39 @@ public class ObservableCqlSessionFactoryBean extends AbstractFactoryBean<CqlSess
this.observationRegistry = observationRegistry;
}
@Nullable
public String getRemoteServiceName() {
return remoteServiceName;
}
/**
* Set the remote service name.
*
* @param remoteServiceName
*/
public void setRemoteServiceName(@Nullable String remoteServiceName) {
this.remoteServiceName = remoteServiceName;
}
/**
* Set the observation convention.
*
* @param convention
* @since 4.3.4
*/
public void setConvention(CassandraObservationConvention convention) {
this.convention = convention;
}
@Override
protected CqlSession createInstance() {
cqlSessionBuilder.addRequestTracker(ObservationRequestTracker.INSTANCE);
if (ObjectUtils.isEmpty(getRemoteServiceName())) {
return ObservableCqlSessionFactory.wrap(cqlSessionBuilder.build(), observationRegistry);
}
String remoteServiceName = ObjectUtils.isEmpty(getRemoteServiceName()) ? "Cassandra" : getRemoteServiceName();
return ObservableCqlSessionFactory.wrap(cqlSessionBuilder.build(), getRemoteServiceName(), observationRegistry);
return ObservableCqlSessionFactory.wrap(cqlSessionBuilder.build(), remoteServiceName, convention,
observationRegistry);
}
@Override
@@ -82,17 +107,4 @@ public class ObservableCqlSessionFactoryBean extends AbstractFactoryBean<CqlSess
return CqlSession.class;
}
@Nullable
public String getRemoteServiceName() {
return remoteServiceName;
}
/**
* Set the remote service name.
*
* @param remoteServiceName
*/
public void setRemoteServiceName(@Nullable String remoteServiceName) {
this.remoteServiceName = remoteServiceName;
}
}

27
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSession.java
View File

@@ -50,12 +50,13 @@ public class ObservableReactiveSession implements ReactiveSession {
private final ObservationRegistry observationRegistry;
private final CassandraObservationConvention convention = new DefaultCassandraObservationConvention();
private final CassandraObservationConvention convention;
ObservableReactiveSession(ReactiveSession delegate, String remoteServiceName,
ObservationRegistry observationRegistry) {
CassandraObservationConvention convention, ObservationRegistry observationRegistry) {
this.delegate = delegate;
this.remoteServiceName = remoteServiceName;
this.convention = convention;
this.observationRegistry = observationRegistry;
}
@@ -67,19 +68,37 @@ public class ObservableReactiveSession implements ReactiveSession {
* @return traced representation of a {@link ReactiveSession}.
*/
public static ReactiveSession create(ReactiveSession session, ObservationRegistry observationRegistry) {
return new ObservableReactiveSession(session, "Cassandra", observationRegistry);
return new ObservableReactiveSession(session, "Cassandra", DefaultCassandraObservationConvention.INSTANCE,
observationRegistry);
}
/**
* Factory method for creation of a {@link ObservableReactiveSession}.
*
* @param session reactive session.
* @param remoteServiceName the remote service name.
* @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);
return new ObservableReactiveSession(session, remoteServiceName, DefaultCassandraObservationConvention.INSTANCE,
observationRegistry);
}
/**
* Factory method for creation of a {@link ObservableReactiveSession}.
*
* @param session reactive session.
* @param remoteServiceName the remote service name.
* @param convention the observation convention.
* @param observationRegistry observation registry.
* @return traced representation of a {@link ReactiveSession}.
* @since 4.3.4
*/
public static ReactiveSession create(ReactiveSession session, String remoteServiceName,
CassandraObservationConvention convention, ObservationRegistry observationRegistry) {
return new ObservableReactiveSession(session, remoteServiceName, convention, observationRegistry);
}
@Override

20
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactory.java
View File

@@ -55,11 +55,27 @@ public final class ObservableReactiveSessionFactory {
*/
public static ReactiveSession wrap(ReactiveSession session, String remoteServiceName,
ObservationRegistry observationRegistry) {
return wrap(session, remoteServiceName, DefaultCassandraObservationConvention.INSTANCE, observationRegistry);
}
/**
* Wrap the {@link CqlSession} with a {@link CqlSessionObservationInterceptor}.
*
* @param session must not be {@literal null}.
* @param remoteServiceName must not be {@literal null}.
* @param convention the observation convention.
* @param observationRegistry must not be {@literal null}.
* @return
* @since 4.3.4
*/
public static ReactiveSession wrap(ReactiveSession session, String remoteServiceName,
CassandraObservationConvention convention, ObservationRegistry observationRegistry) {
Assert.notNull(session, "CqlSession must not be null");
Assert.notNull(remoteServiceName, "CqlSessionObservationConvention must not be null");
Assert.notNull(remoteServiceName, "Remote service name must not be null");
Assert.notNull(convention, "CassandraObservationConvention must not be null");
Assert.notNull(observationRegistry, "ObservationRegistry must not be null");
return ObservableReactiveSession.create(session, remoteServiceName, observationRegistry);
return ObservableReactiveSession.create(session, remoteServiceName, convention, observationRegistry);
}
}

59
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactoryBean.java
View File

@@ -50,6 +50,8 @@ public class ObservableReactiveSessionFactoryBean extends AbstractFactoryBean<Re
private @Nullable String remoteServiceName;
private CassandraObservationConvention convention = DefaultCassandraObservationConvention.INSTANCE;
/**
* Construct a new {@link ObservableReactiveSessionFactoryBean}.
*
@@ -85,35 +87,11 @@ public class ObservableReactiveSessionFactoryBean extends AbstractFactoryBean<Re
this.observationRegistry = observationRegistry;
}
@Override
protected ReactiveSession createInstance() {
if (ObjectUtils.isEmpty(getRemoteServiceName())) {
return ObservableReactiveSessionFactory.wrap(new DefaultBridgedReactiveSession(cqlSession), observationRegistry);
}
return ObservableReactiveSessionFactory.wrap(new DefaultBridgedReactiveSession(cqlSession), getRemoteServiceName(),
observationRegistry);
}
@Override
public Class<?> getObjectType() {
return ReactiveSession.class;
}
@Nullable
public String getRemoteServiceName() {
return remoteServiceName;
}
@Override
public void destroy() {
if (requiresDestroy) {
cqlSession.close();
}
}
/**
* Set the remote service name.
*
@@ -122,4 +100,37 @@ public class ObservableReactiveSessionFactoryBean extends AbstractFactoryBean<Re
public void setRemoteServiceName(@Nullable String remoteServiceName) {
this.remoteServiceName = remoteServiceName;
}
/**
* Set the observation convention.
*
* @param convention
* @since 4.3.4
*/
public void setConvention(CassandraObservationConvention convention) {
this.convention = convention;
}
@Override
protected ReactiveSession createInstance() {
String remoteServiceName = ObjectUtils.isEmpty(getRemoteServiceName()) ? "Cassandra" : getRemoteServiceName();
return ObservableReactiveSessionFactory.wrap(new DefaultBridgedReactiveSession(cqlSession), remoteServiceName,
convention, observationRegistry);
}
@Override
public Class<?> getObjectType() {
return ReactiveSession.class;
}
@Override
public void destroy() {
if (requiresDestroy) {
cqlSession.close();
}
}
}

3
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactoryUnitTests.java
View File

@@ -21,6 +21,7 @@ import static org.mockito.Mockito.*;
import io.micrometer.observation.ObservationRegistry;
import org.junit.jupiter.api.Test;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.support.AopUtils;
@@ -34,7 +35,7 @@ import com.datastax.oss.driver.api.core.CqlSession;
class ObservableCqlSessionFactoryUnitTests {
@Test // GH-1426
void sessionFactoryBeanUnwrapsObservationProxy() throws Exception {
void sessionFactoryBeanUnwrapsObservationProxy() {
CqlSession session = mock(CqlSession.class);
ObservationRegistry registry = ObservationRegistry.NOOP;

19
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactoryBeanUnitTests.java
View File

@@ -96,4 +96,23 @@ class ObservableReactiveSessionFactoryBeanUnitTests {
verifyNoInteractions(sessionMock);
}
@Test // GH-1490
void considersConvention() throws Exception {
CqlSession sessionMock = mock(CqlSession.class);
CassandraObservationConvention conventionMock = mock(CassandraObservationConvention.class);
ObservationRegistry registry = ObservationRegistry.NOOP;
CqlSession wrapped = ObservableCqlSessionFactory.wrap(sessionMock, registry);
ObservableReactiveSessionFactoryBean bean = new ObservableReactiveSessionFactoryBean(wrapped, registry);
bean.setConvention(conventionMock);
bean.afterPropertiesSet();
ReactiveSession object = bean.getObject();
Object usedConvention = ReflectionTestUtils.getField(object, "convention");
assertThat(usedConvention).isSameAs(conventionMock);
}
}

2
src/main/antora/modules/ROOT/pages/observability.adoc
View File

@@ -32,6 +32,8 @@ Also, registers `ObservationRequestTracker.INSTANCE` with the `CqlSessionBuilder
<2> Wraps a CQL session object to observe reactive Cassandra statement execution.
====
Both, javadoc:org.springframework.data.cassandra.observability.ObservableCqlSessionFactoryBean[] and javadoc:org.springframework.data.cassandra.observability.ObservableReactiveSessionFactoryBean[] support configuration of javadoc:org.springframework.data.cassandra.observability.CassandraObservationConvention[].
See also https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/database/#cassandra[OpenTelemetry Semantic Conventions] for further reference.
include::observability/conventions.adoc[leveloffset=+1]