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

Align tracing tag names with OpenTelemetry spec.

Browse Source 
See #1321
Original pull request: #1322
This commit is contained in:
Mark Paluch
2022-10-24 13:59:00 +02:00
parent d75da45038
commit 1d3e754e3e
12 changed files with 210 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,11 +39,6 @@ enum CassandraObservation implements ObservationDocumentation {
return "spring.data.cassandra.query";
}
@Override
public String getContextualName() {
return "query";
}
@Override
public KeyName[] getLowCardinalityKeyNames() {
return LowCardinalityKeyNames.values();
@@ -54,21 +49,77 @@ enum CassandraObservation implements ObservationDocumentation {
return HighCardinalityKeyNames.values();
}
@Override
public String getPrefix() {
return "spring.data.cassandra.";
}
};
enum LowCardinalityKeyNames implements KeyName {
/**
* Database system.
*/
DATABASE_SYSTEM {
@Override
public String asString() {
return "db.system";
}
},
/**
* Network transport.
*/
NET_TRANSPORT {
@Override
public String asString() {
return "net.transport";
}
},
/**
* Name of the database host.
*/
NET_PEER_NAME {
@Override
public String asString() {
return "net.peer.name";
}
},
/**
* Logical remote port number.
*/
NET_PEER_PORT {
@Override
public String asString() {
return "net.peer.port";
}
},
/**
* Cassandra peer address.
*/
NET_SOCK_PEER_ADDR {
@Override
public String asString() {
return "net.sock.peer.addr";
}
},
/**
* Cassandra peer port.
*/
NET_SOCK_PEER_PORT {
@Override
public String asString() {
return "net.sock.peer.port";
}
},
/**
* Name of the Cassandra keyspace.
*/
KEYSPACE_NAME {
@Override
public String asString() {
return "spring.data.cassandra.keyspace";
return "db.name";
}
},
@@ -93,15 +144,27 @@ enum CassandraObservation implements ObservationDocumentation {
},
/**
* Cassandra URL
* The database operation.
*/
URL {
DB_OPERATION {
@Override
public String asString() {
return "spring.data.cassandra.url";
return "db.operation";
}
},
COORDINATOR {
@Override
public String asString() {
return "db.cassandra.coordinator.id";
}
},
COORDINATOR_DC {
@Override
public String asString() {
return "db.cassandra.coordinator.dc";
}
}
}
enum HighCardinalityKeyNames implements KeyName {
@@ -109,10 +172,29 @@ enum CassandraObservation implements ObservationDocumentation {
/**
* A key-value containing Cassandra CQL.
*/
CQL_TAG {
DB_STATEMENT {
@Override
public String asString() {
return "spring.data.cassandra.cql";
return "db.statement";
}
},
PAGE_SIZE {
@Override
public String asString() {
return "db.cassandra.page_size";
}
},
CONSISTENCY_LEVEL {
@Override
public String asString() {
return "db.cassandra.consistency_level";
}
},
IDEMPOTENCE {
@Override
public String asString() {
return "db.cassandra.idempotence";
}
},

14
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CassandraObservationContext.java
View File

@@ -15,8 +15,11 @@
*/
package org.springframework.data.cassandra.observability;
import org.springframework.lang.Nullable;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.Statement;
import com.datastax.oss.driver.api.core.metadata.Node;
import io.micrometer.observation.Observation;
import io.micrometer.observation.transport.Kind;
@@ -38,6 +41,8 @@ public class CassandraObservationContext extends SenderContext<Object> {
private final String sessionName;
private final String keyspaceName;
private volatile @Nullable Node node;
public CassandraObservationContext(Statement<?> statement, String remoteServiceName, boolean prepare,
String methodName, String sessionName, String keyspaceName) {
@@ -71,4 +76,13 @@ public class CassandraObservationContext extends SenderContext<Object> {
public String getKeyspaceName() {
return keyspaceName;
}
public void setNode(Node node) {
this.node = node;
}
@Nullable
public Node getNode() {
return node;
}
}

4
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/CqlSessionObservationInterceptor.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.
@@ -39,7 +39,7 @@ import io.micrometer.observation.ObservationRegistry;
* @author Mark Paluch
* @author Marcin Grzejszczak
* @author Greg Turnquist
* @since 4.0.0
* @since 4.0
*/
final class CqlSessionObservationInterceptor implements MethodInterceptor {

78
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/DefaultCassandraObservationConvention.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,17 +15,22 @@
*/
package org.springframework.data.cassandra.observability;
import java.net.InetSocketAddress;
import java.util.StringJoiner;
import org.springframework.data.cassandra.observability.CassandraObservation.HighCardinalityKeyNames;
import org.springframework.data.cassandra.observability.CassandraObservation.LowCardinalityKeyNames;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;
import com.datastax.oss.driver.api.core.ConsistencyLevel;
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 com.datastax.oss.driver.api.core.metadata.EndPoint;
import com.datastax.oss.driver.api.core.metadata.Node;
import io.micrometer.common.KeyValues;
@@ -41,13 +46,36 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
@Override
public KeyValues getLowCardinalityKeyValues(CassandraObservationContext context) {
KeyValues keyValues = KeyValues.of(LowCardinalityKeyNames.SESSION_NAME.withValue(context.getSessionName()),
LowCardinalityKeyNames.KEYSPACE_NAME.withValue(context.getKeyspaceName()),
LowCardinalityKeyNames.METHOD_NAME.withValue(context.getMethodName()));
String dbOperation = context.isPrepare() ? "PREPARE" : getOperationName(getCql(context.getStatement()), "");
if (context.getStatement().getNode() != null) {
keyValues = keyValues.and(
LowCardinalityKeyNames.URL.withValue(context.getStatement().getNode().getEndPoint().resolve().toString()));
KeyValues keyValues = KeyValues.of(LowCardinalityKeyNames.DATABASE_SYSTEM.withValue("cassandra"),
LowCardinalityKeyNames.KEYSPACE_NAME.withValue(context.getKeyspaceName()),
LowCardinalityKeyNames.SESSION_NAME.withValue(context.getSessionName()),
LowCardinalityKeyNames.METHOD_NAME.withValue(context.getMethodName()),
LowCardinalityKeyNames.DB_OPERATION.withValue(dbOperation));
Node node = context.getNode();
if (node == null) {
node = context.getStatement().getNode();
}
if (node != null) {
EndPoint endPoint = node.getEndPoint();
keyValues = keyValues.and(LowCardinalityKeyNames.COORDINATOR.withValue("" + node.getHostId()),
LowCardinalityKeyNames.COORDINATOR_DC.withValue("" + node.getDatacenter()));
keyValues.and(LowCardinalityKeyNames.NET_PEER_NAME.withValue(endPoint.toString()));
InetSocketAddress socketAddress = tryGetSocketAddress(endPoint);
if (socketAddress != null) {
keyValues = keyValues.and(LowCardinalityKeyNames.NET_TRANSPORT.withValue("IP.TCP"),
LowCardinalityKeyNames.NET_SOCK_PEER_ADDR.withValue(socketAddress.getHostString()),
LowCardinalityKeyNames.NET_SOCK_PEER_PORT.withValue("" + socketAddress.getPort()));
}
}
return keyValues;
@@ -55,12 +83,42 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
@Override
public KeyValues getHighCardinalityKeyValues(CassandraObservationContext context) {
return KeyValues.of(HighCardinalityKeyNames.CQL_TAG.withValue(getCql(context.getStatement())));
Statement<?> statement = context.getStatement();
KeyValues keyValues = KeyValues.of(HighCardinalityKeyNames.DB_STATEMENT.withValue(getCql(statement)),
HighCardinalityKeyNames.PAGE_SIZE.withValue("" + statement.getPageSize()));
Boolean idempotent = statement.isIdempotent();
if (idempotent != null) {
keyValues = keyValues
.and(HighCardinalityKeyNames.IDEMPOTENCE.withValue(idempotent ? "idempotent" : "non-idempotent"));
}
ConsistencyLevel consistencyLevel = statement.getConsistencyLevel();
if (consistencyLevel != null) {
keyValues = keyValues.and(HighCardinalityKeyNames.CONSISTENCY_LEVEL.withValue("" + consistencyLevel.name()));
}
return keyValues;
}
@Nullable
private InetSocketAddress tryGetSocketAddress(EndPoint endPoint) {
try {
if (endPoint.resolve()instanceof InetSocketAddress inet) {
return inet;
}
} catch (RuntimeException e) {}
return null;
}
@Override
public String getContextualName(CassandraObservationContext context) {
return (context.isPrepare() ? "PREPARE: " : "") + getSpanName(getCql(context.getStatement()), "");
return (context.isPrepare() ? "PREPARE: " : "") + getOperationName(getCql(context.getStatement()), "");
}
/**
@@ -116,7 +174,7 @@ class DefaultCassandraObservationConvention implements CassandraObservationConve
* @param defaultName if there's no query
* @return span name
*/
public String getSpanName(String cql, String defaultName) {
public String getOperationName(String cql, String defaultName) {
if (StringUtils.hasText(cql) && cql.indexOf(' ') > -1) {
return cql.substring(0, cql.indexOf(' '));

2
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableCqlSessionFactory.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.

2
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservableReactiveSessionFactory.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.

31
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/observability/ObservationRequestTracker.java
View File

@@ -15,9 +15,6 @@
*/
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;
@@ -89,11 +86,12 @@ public enum ObservationRequestTracker implements RequestTracker {
Observation observation = ((CassandraObservationSupplier) request).getObservation();
observation.event(Event.of(Events.NODE_SUCCESS.getValue()));
((CassandraObservationContext) observation.getContext()).setNode(node);
observation.highCardinalityKeyValue(
String.format(HighCardinalityKeyNames.NODE_ERROR_TAG.asString(), node.getEndPoint()), error.toString());
observation.event(Event.of(Events.NODE_ERROR.getValue()));
tryAddingRemoteIpAndPort(node, observation);
if (log.isDebugEnabled()) {
log.debug("Marking node error for [" + observation + "]");
}
@@ -108,8 +106,9 @@ public enum ObservationRequestTracker implements RequestTracker {
Observation observation = ((CassandraObservationSupplier) request).getObservation();
((CassandraObservationContext) observation.getContext()).setNode(node);
observation.event(Event.of(Events.NODE_SUCCESS.getValue()));
tryAddingRemoteIpAndPort(node, observation);
if (log.isDebugEnabled()) {
log.debug("Marking node success for [" + observation + "]");
@@ -122,24 +121,4 @@ public enum ObservationRequestTracker implements RequestTracker {
}
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);
}
}
}

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

@@ -1,157 +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.*;
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;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
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;
/**
* 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;
@BeforeEach
void setUp() {
this.context = new AnnotationConfigApplicationContext(Config.class);
this.session = ObservableCqlSessionFactory.wrap(context.getBean(CqlSession.class), "my-cassandra",
observationRegistry);
this.observationRegistry.observationConfig().observationHandler(new DefaultMeterObservationHandler(meterRegistry));
}
@AfterEach
void tearDown() {
this.context.close();
}
@Test
void shouldCreateObservationForCqlSessionOperations() {
Observation.start("test", observationRegistry).scoped(() -> {
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( //
LowCardinalityKeyNames.SESSION_NAME.withValue("s5"), //
LowCardinalityKeyNames.KEYSPACE_NAME.withValue("system"), //
KeyValue.of("error", "none") //
));
assertThat(tracer.getSpans()).hasSize(4);
assertThat(findSpan(tracer.getSpans(), LowCardinalityKeyNames.METHOD_NAME, "execute")).isNotNull();
assertThat(findSpan(tracer.getSpans(), LowCardinalityKeyNames.METHOD_NAME, "executeAsync")).isNotNull();
assertThat(findSpan(tracer.getSpans(), LowCardinalityKeyNames.METHOD_NAME, "prepare")).isNotNull();
assertThat(findSpan(tracer.getSpans(), LowCardinalityKeyNames.METHOD_NAME, "prepareAsync")).isNotNull();
tracer.getSpans().forEach(simpleSpan -> SpanAssert.then(simpleSpan) //
.hasRemoteServiceNameEqualTo("cassandra-s5") //
.hasNameEqualTo(CASSANDRA_QUERY_OBSERVATION.getContextualName()) //
.hasTag(LowCardinalityKeyNames.SESSION_NAME, "s5") //
.hasTag(LowCardinalityKeyNames.KEYSPACE_NAME, "system") //
.hasTag(HighCardinalityKeyNames.CQL_TAG, 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<SimpleSpan> spans, KeyName key, String value) {
return spans.stream() //
.filter(simpleSpan -> {
Map<String, String> tags = simpleSpan.getTags();
return tags.containsKey(key.asString()) && tags.get(key.asString()).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();
}
}
}

8
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ImperativeIntegrationTests.java
View File

@@ -29,6 +29,7 @@ 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.exporter.FinishedSpan;
import io.micrometer.tracing.test.SampleTestRunner;
/**
@@ -78,6 +79,13 @@ public class ImperativeIntegrationTests extends SampleTestRunner {
System.out.println(((SimpleMeterRegistry) meterRegistry).getMetersAsString());
assertThat(tracer.getFinishedSpans()).hasSize(6);
for (FinishedSpan finishedSpan : tracer.getFinishedSpans()) {
assertThat(finishedSpan.getTags()).containsEntry("db.system", "cassandra");
assertThat(finishedSpan.getTags()).containsKeys("db.operation", "db.statement",
"spring.data.cassandra.methodName", "spring.data.cassandra.sessionName");
}
};
}
}

8
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/ReactiveIntegrationTests.java
View File

@@ -82,13 +82,15 @@ public class ReactiveIntegrationTests extends SampleTestRunner {
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();
.contextWrite(Context.of(ObservationThreadLocalAccessor.KEY, intermediate)) //
.as(StepVerifier::create) //
.expectNextCount(1) //
.verifyComplete();
});
System.out.println(((SimpleMeterRegistry) meterRegistry).getMetersAsString());
assertThat(tracer.getFinishedSpans()).hasSize(7);
assertThat(tracer.getFinishedSpans().size()).isGreaterThanOrEqualTo(7);
};
}
}

8
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/observability/TestConfig.java
View File

@@ -16,6 +16,8 @@
package org.springframework.data.cassandra.observability;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.data.cassandra.config.SessionBuilderConfigurer;
import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession;
@@ -62,6 +64,12 @@ class TestConfig extends AbstractTestJavaConfig {
return sessionBuilder -> sessionBuilder.addRequestTracker(ObservationRequestTracker.INSTANCE);
}
@Nullable
@Override
protected Resource getDriverConfigurationResource() {
return new ClassPathResource("application.conf");
}
@Bean
ReactiveSession reactiveSession(CqlSession session) {
return new DefaultBridgedReactiveSession(session);

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

@@ -57,4 +57,4 @@ include::../../../../target/_metrics.adoc[]
include::../../../../target/_spans.adoc[]
See also https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/database/#Cassandra[OpenTelemetry Semantic Conventions] for further reference.
See also https://opentelemetry.io/docs/reference/specification/trace/semantic_conventions/database/#cassandra[OpenTelemetry Semantic Conventions] for further reference.