Cassandra java tests
DELETE
instrumentation/cassandra/cassandra-3.0/javaagent/src/test/groovy/CassandraClientTest.groovy
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-import com.datastax.driver.core.Cluster
-import com.datastax.driver.core.Session
-import io.opentelemetry.api.trace.SpanKind
-import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-import io.opentelemetry.instrumentation.test.asserts.TraceAssert
-import io.opentelemetry.sdk.trace.data.SpanData
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testcontainers.containers.GenericContainer
-import org.testcontainers.containers.output.Slf4jLogConsumer
-import spock.lang.Shared
-
-import java.time.Duration
-import java.util.concurrent.Executors
-import java.util.concurrent.atomic.AtomicBoolean
-
-import static io.opentelemetry.api.trace.SpanKind.CLIENT
-
-class CassandraClientTest extends AgentInstrumentationSpecification {
- private static final Logger logger = LoggerFactory.getLogger(CassandraClientTest)
-
- @Shared
- def executor = Executors.newCachedThreadPool()
-
- @Shared
- GenericContainer cassandra
- @Shared
- int cassandraPort
- @Shared
- Cluster cluster
-
- def setupSpec() {
- cassandra = new GenericContainer("cassandra:3")
- // limit memory usage
- .withEnv("JVM_OPTS", "-Xmx128m -Xms128m")
- .withExposedPorts(9042)
- .withLogConsumer(new Slf4jLogConsumer(logger))
- .withStartupTimeout(Duration.ofSeconds(120))
- cassandra.start()
-
- cassandraPort = cassandra.getMappedPort(9042)
-
- cluster = Cluster.builder()
- .addContactPointsWithPorts(new InetSocketAddress("localhost", cassandraPort))
- .build()
- }
-
- def cleanupSpec() {
- cluster?.close()
- cassandra.stop()
- }
-
- def "test sync"() {
- setup:
- Session session = cluster.connect(keyspace)
-
- session.execute(statement)
-
- expect:
- assertTraces(keyspace ? 2 : 1) {
- if (keyspace) {
- trace(0, 1) {
- cassandraSpan(it, 0, "DB Query", "USE $keyspace")
- }
- }
- trace(keyspace ? 1 : 0, 1) {
- cassandraSpan(it, 0, spanName, expectedStatement, operation, keyspace, table)
- }
- }
-
- cleanup:
- session.close()
-
- where:
- keyspace | statement | expectedStatement | spanName | operation | table
- null | "DROP KEYSPACE IF EXISTS sync_test" | "DROP KEYSPACE IF EXISTS sync_test" | "DB Query" | null | null
- null | "CREATE KEYSPACE sync_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}" | "CREATE KEYSPACE sync_test WITH REPLICATION = {?:?, ?:?}" | "DB Query" | null | null
- "sync_test" | "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )" | "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )" | "sync_test" | null | null
- "sync_test" | "INSERT INTO sync_test.users (id, name) values (uuid(), 'alice')" | "INSERT INTO sync_test.users (id, name) values (uuid(), ?)" | "INSERT sync_test.users" | "INSERT" | "sync_test.users"
- "sync_test" | "SELECT * FROM users where name = 'alice' ALLOW FILTERING" | "SELECT * FROM users where name = ? ALLOW FILTERING" | "SELECT sync_test.users" | "SELECT" | "users"
- }
-
- def "test async"() {
- setup:
- def callbackExecuted = new AtomicBoolean()
- Session session = cluster.connect(keyspace)
- runWithSpan("parent") {
- def future = session.executeAsync(statement)
- future.addListener({ ->
- runWithSpan("callbackListener") {
- callbackExecuted.set(true)
- }
- }, executor)
- }
-
- expect:
- assertTraces(keyspace ? 2 : 1) {
- if (keyspace) {
- trace(0, 1) {
- cassandraSpan(it, 0, "DB Query", "USE $keyspace")
- }
- }
- trace(keyspace ? 1 : 0, 3) {
- span(0) {
- name "parent"
- kind SpanKind.INTERNAL
- hasNoParent()
- }
- cassandraSpan(it, 1, spanName, expectedStatement, operation, keyspace, table, span(0))
- span(2) {
- name "callbackListener"
- kind SpanKind.INTERNAL
- childOf span(0)
- }
- }
- }
-
- cleanup:
- session.close()
-
- where:
- keyspace | statement | expectedStatement | spanName | operation | table
- null | "DROP KEYSPACE IF EXISTS async_test" | "DROP KEYSPACE IF EXISTS async_test" | "DB Query" | null | null
- null | "CREATE KEYSPACE async_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}" | "CREATE KEYSPACE async_test WITH REPLICATION = {?:?, ?:?}" | "DB Query" | null | null
- "async_test" | "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )" | "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )" | "async_test" | null | null
- "async_test" | "INSERT INTO async_test.users (id, name) values (uuid(), 'alice')" | "INSERT INTO async_test.users (id, name) values (uuid(), ?)" | "INSERT async_test.users" | "INSERT" | "async_test.users"
- "async_test" | "SELECT * FROM users where name = 'alice' ALLOW FILTERING" | "SELECT * FROM users where name = ? ALLOW FILTERING" | "SELECT async_test.users" | "SELECT" | "users"
- }
-
- def cassandraSpan(TraceAssert trace, int index, String spanName, String statement,
- String operation = null,
- String keyspace = null,
- String table = null,
- Object parentSpan = null) {
- trace.span(index) {
- name spanName
- kind CLIENT
- if (parentSpan == null) {
- hasNoParent()
- } else {
- childOf((SpanData) parentSpan)
- }
- attributes {
- "$SemanticAttributes.NET_SOCK_PEER_ADDR" "127.0.0.1"
- "$SemanticAttributes.NET_SOCK_PEER_NAME" "localhost"
- "$SemanticAttributes.NET_SOCK_PEER_PORT" cassandraPort
- "$SemanticAttributes.DB_SYSTEM" "cassandra"
- "$SemanticAttributes.DB_NAME" keyspace
- "$SemanticAttributes.DB_STATEMENT" statement
- "$SemanticAttributes.DB_OPERATION" operation
- "$SemanticAttributes.DB_CASSANDRA_TABLE" table
- }
- }
- }
-}

ADD
instrumentation/cassandra/cassandra-3.0/javaagent/src/test/java/CassandraClientTest.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_TABLE;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_NAME;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_OPERATION;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_STATEMENT;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_SYSTEM;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NET_SOCK_PEER_ADDR;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NET_SOCK_PEER_NAME;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NET_SOCK_PEER_PORT;
+import static org.junit.jupiter.api.Named.named;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.Session;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+
+public class CassandraClientTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(CassandraClientTest.class);
+
+ private static final Executor executor = Executors.newCachedThreadPool();
+
+ @RegisterExtension
+ static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+ @SuppressWarnings("rawtypes")
+ private static GenericContainer cassandra;
+
+ private static int cassandraPort;
+ private static Cluster cluster;
+
+ @BeforeAll
+ static void beforeAll() {
+ cassandra =
+ new GenericContainer<>("cassandra:3")
+ .withEnv("JVM_OPTS", "-Xmx128m -Xms128m")
+ .withExposedPorts(9042)
+ .withLogConsumer(new Slf4jLogConsumer(logger))
+ .withStartupTimeout(Duration.ofMinutes(2));
+ cassandra.start();
+
+ cassandraPort = cassandra.getMappedPort(9042);
+ cluster =
+ Cluster.builder()
+ .addContactPointsWithPorts(new InetSocketAddress("localhost", cassandraPort))
+ .build();
+ }
+
+ @AfterAll
+ static void afterAll() {
+ cluster.close();
+ cassandra.stop();
+ }
+
+ @ParameterizedTest(name = "{index}: {0}")
+ @MethodSource("provideSyncParameters")
+ void syncTest(Parameter parameter) {
+ Session session = cluster.connect(parameter.keyspace);
+
+ session.execute(parameter.statement);
+
+ if (parameter.keyspace != null) {
+ testing.waitAndAssertTraces(
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span ->
+ span.hasName("DB Query")
+ .hasKind(SpanKind.CLIENT)
+ .hasNoParent()
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_STATEMENT, "USE " + parameter.keyspace))),
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span ->
+ span.hasName(parameter.spanName)
+ .hasKind(SpanKind.CLIENT)
+ .hasNoParent()
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_NAME, parameter.keyspace),
+ equalTo(DB_STATEMENT, parameter.expectedStatement),
+ equalTo(DB_OPERATION, parameter.operation),
+ equalTo(DB_CASSANDRA_TABLE, parameter.table))));
+ } else {
+ testing.waitAndAssertTraces(
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span ->
+ span.hasName(parameter.spanName)
+ .hasKind(SpanKind.CLIENT)
+ .hasNoParent()
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_STATEMENT, parameter.expectedStatement),
+ equalTo(DB_OPERATION, parameter.operation),
+ equalTo(DB_CASSANDRA_TABLE, parameter.table))));
+ }
+
+ session.close();
+ }
+
+ @ParameterizedTest(name = "{index}: {0}")
+ @MethodSource("provideAsyncParameters")
+ void asyncTest(Parameter parameter) {
+ @SuppressWarnings("WriteOnlyObject")
+ AtomicBoolean callbackExecuted = new AtomicBoolean();
+ Session session = cluster.connect(parameter.keyspace);
+
+ testing.runWithSpan(
+ "parent",
+ () -> {
+ ResultSetFuture future = session.executeAsync(parameter.statement);
+ future.addListener(
+ () -> testing.runWithSpan("callbackListener", () -> callbackExecuted.set(true)),
+ executor);
+ });
+
+ if (parameter.keyspace != null) {
+ testing.waitAndAssertTraces(
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span ->
+ span.hasName("DB Query")
+ .hasKind(SpanKind.CLIENT)
+ .hasNoParent()
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_STATEMENT, "USE " + parameter.keyspace))),
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+ span ->
+ span.hasName(parameter.spanName)
+ .hasKind(SpanKind.CLIENT)
+ .hasParent(trace.getSpan(0))
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_NAME, parameter.keyspace),
+ equalTo(DB_STATEMENT, parameter.expectedStatement),
+ equalTo(DB_OPERATION, parameter.operation),
+ equalTo(DB_CASSANDRA_TABLE, parameter.table)),
+ span ->
+ span.hasName("callbackListener")
+ .hasKind(SpanKind.INTERNAL)
+ .hasParent(trace.getSpan(0))));
+ } else {
+ testing.waitAndAssertTraces(
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+ span ->
+ span.hasName(parameter.spanName)
+ .hasKind(SpanKind.CLIENT)
+ .hasParent(trace.getSpan(0))
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_STATEMENT, parameter.expectedStatement),
+ equalTo(DB_OPERATION, parameter.operation),
+ equalTo(DB_CASSANDRA_TABLE, parameter.table)),
+ span ->
+ span.hasName("callbackListener")
+ .hasKind(SpanKind.INTERNAL)
+ .hasParent(trace.getSpan(0))));
+ }
+
+ session.close();
+ }
+
+ private static Stream<Arguments> provideSyncParameters() {
+ return Stream.of(
+ Arguments.of(
+ named(
+ "Drop keyspace if exists",
+ new Parameter(
+ null,
+ "DROP KEYSPACE IF EXISTS sync_test",
+ "DROP KEYSPACE IF EXISTS sync_test",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create keyspace with replication",
+ new Parameter(
+ null,
+ "CREATE KEYSPACE sync_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}",
+ "CREATE KEYSPACE sync_test WITH REPLICATION = {?:?, ?:?}",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create table",
+ new Parameter(
+ "sync_test",
+ "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )",
+ "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )",
+ "sync_test",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Insert data",
+ new Parameter(
+ "sync_test",
+ "INSERT INTO sync_test.users (id, name) values (uuid(), 'alice')",
+ "INSERT INTO sync_test.users (id, name) values (uuid(), ?)",
+ "INSERT sync_test.users",
+ "INSERT",
+ "sync_test.users"))),
+ Arguments.of(
+ named(
+ "Select data",
+ new Parameter(
+ "sync_test",
+ "SELECT * FROM users where name = 'alice' ALLOW FILTERING",
+ "SELECT * FROM users where name = ? ALLOW FILTERING",
+ "SELECT sync_test.users",
+ "SELECT",
+ "users"))));
+ }
+
+ private static Stream<Arguments> provideAsyncParameters() {
+ return Stream.of(
+ Arguments.of(
+ named(
+ "Drop keyspace if exists",
+ new Parameter(
+ null,
+ "DROP KEYSPACE IF EXISTS async_test",
+ "DROP KEYSPACE IF EXISTS async_test",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create keyspace with replication",
+ new Parameter(
+ null,
+ "CREATE KEYSPACE async_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}",
+ "CREATE KEYSPACE async_test WITH REPLICATION = {?:?, ?:?}",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create table",
+ new Parameter(
+ "async_test",
+ "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )",
+ "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )",
+ "async_test",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Insert data",
+ new Parameter(
+ "async_test",
+ "INSERT INTO async_test.users (id, name) values (uuid(), 'alice')",
+ "INSERT INTO async_test.users (id, name) values (uuid(), ?)",
+ "INSERT async_test.users",
+ "INSERT",
+ "async_test.users"))),
+ Arguments.of(
+ named(
+ "Select data",
+ new Parameter(
+ "async_test",
+ "SELECT * FROM users where name = 'alice' ALLOW FILTERING",
+ "SELECT * FROM users where name = ? ALLOW FILTERING",
+ "SELECT async_test.users",
+ "SELECT",
+ "users"))));
+ }
+
+ private static class Parameter {
+ public final String keyspace;
+ public final String statement;
+ public final String expectedStatement;
+ public final String spanName;
+ public final String operation;
+ public final String table;
+
+ public Parameter(
+ String keyspace,
+ String statement,
+ String expectedStatement,
+ String spanName,
+ String operation,
+ String table) {
+ this.keyspace = keyspace;
+ this.statement = statement;
+ this.expectedStatement = expectedStatement;
+ this.spanName = spanName;
+ this.operation = operation;
+ this.table = table;
+ }
+ }
+}

DELETE
instrumentation/cassandra/cassandra-4.0/javaagent/src/test/groovy/CassandraClientTest.groovy
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-import com.datastax.oss.driver.api.core.CqlSession
-import com.datastax.oss.driver.api.core.config.DefaultDriverOption
-import com.datastax.oss.driver.api.core.config.DriverConfigLoader
-import com.datastax.oss.driver.internal.core.config.typesafe.DefaultDriverConfigLoader
-import io.opentelemetry.api.trace.SpanKind
-import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification
-import io.opentelemetry.instrumentation.test.asserts.TraceAssert
-import io.opentelemetry.sdk.trace.data.SpanData
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-import org.testcontainers.containers.GenericContainer
-import org.testcontainers.containers.output.Slf4jLogConsumer
-import spock.lang.Shared
-
-import java.time.Duration
-
-import static io.opentelemetry.api.trace.SpanKind.CLIENT
-
-class CassandraClientTest extends AgentInstrumentationSpecification {
- private static final Logger logger = LoggerFactory.getLogger(CassandraClientTest)
-
- @Shared
- GenericContainer cassandra
- @Shared
- int cassandraPort
-
- def setupSpec() {
- cassandra = new GenericContainer("cassandra:4.0")
- // limit memory usage
- .withEnv("JVM_OPTS", "-Xmx128m -Xms128m")
- .withExposedPorts(9042)
- .withLogConsumer(new Slf4jLogConsumer(logger))
- .withStartupTimeout(Duration.ofSeconds(120))
- cassandra.start()
-
- cassandraPort = cassandra.getMappedPort(9042)
- }
-
- def cleanupSpec() {
- cassandra.stop()
- }
-
- def "test sync"() {
- setup:
- CqlSession session = getSession(keyspace)
-
- session.execute(statement)
-
- expect:
- assertTraces(1) {
- trace(0, 1) {
- cassandraSpan(it, 0, spanName, expectedStatement, operation, keyspace, table)
- }
- }
-
- cleanup:
- session.close()
-
- where:
- keyspace | statement | expectedStatement | spanName | operation | table
- null | "DROP KEYSPACE IF EXISTS sync_test" | "DROP KEYSPACE IF EXISTS sync_test" | "DB Query" | null | null
- null | "CREATE KEYSPACE sync_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}" | "CREATE KEYSPACE sync_test WITH REPLICATION = {?:?, ?:?}" | "DB Query" | null | null
- "sync_test" | "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )" | "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )" | "sync_test" | null | null
- "sync_test" | "INSERT INTO sync_test.users (id, name) values (uuid(), 'alice')" | "INSERT INTO sync_test.users (id, name) values (uuid(), ?)" | "INSERT sync_test.users" | "INSERT" | "sync_test.users"
- "sync_test" | "SELECT * FROM users where name = 'alice' ALLOW FILTERING" | "SELECT * FROM users where name = ? ALLOW FILTERING" | "SELECT sync_test.users" | "SELECT" | "users"
- }
-
- def "test async"() {
- setup:
- CqlSession session = getSession(keyspace)
-
- runWithSpan("parent") {
- session.executeAsync(statement).toCompletableFuture().whenComplete({ result, throwable ->
- runWithSpan("child") {}
- }).get()
- }
-
- expect:
- assertTraces(1) {
- trace(0, 3) {
- span(0) {
- name "parent"
- kind SpanKind.INTERNAL
- hasNoParent()
- }
- cassandraSpan(it, 1, spanName, expectedStatement, operation, keyspace, table, span(0))
- span(2) {
- name "child"
- kind SpanKind.INTERNAL
- childOf span(0)
- }
- }
- }
-
- cleanup:
- session.close()
-
- where:
- keyspace | statement | expectedStatement | spanName | operation | table
- null | "DROP KEYSPACE IF EXISTS async_test" | "DROP KEYSPACE IF EXISTS async_test" | "DB Query" | null | null
- null | "CREATE KEYSPACE async_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}" | "CREATE KEYSPACE async_test WITH REPLICATION = {?:?, ?:?}" | "DB Query" | null | null
- "async_test" | "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )" | "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )" | "async_test" | null | null
- "async_test" | "INSERT INTO async_test.users (id, name) values (uuid(), 'alice')" | "INSERT INTO async_test.users (id, name) values (uuid(), ?)" | "INSERT async_test.users" | "INSERT" | "async_test.users"
- "async_test" | "SELECT * FROM users where name = 'alice' ALLOW FILTERING" | "SELECT * FROM users where name = ? ALLOW FILTERING" | "SELECT async_test.users" | "SELECT" | "users"
- }
-
- def cassandraSpan(TraceAssert trace, int index, String spanName, String statement, String operation, String keyspace, String table, Object parentSpan = null) {
- trace.span(index) {
- name spanName
- kind CLIENT
- if (parentSpan == null) {
- hasNoParent()
- } else {
- childOf((SpanData) parentSpan)
- }
- attributes {
- "$SemanticAttributes.NET_SOCK_PEER_ADDR" "127.0.0.1"
- "$SemanticAttributes.NET_SOCK_PEER_NAME" "localhost"
- "$SemanticAttributes.NET_SOCK_PEER_PORT" cassandraPort
- "$SemanticAttributes.DB_SYSTEM" "cassandra"
- "$SemanticAttributes.DB_NAME" keyspace
- "$SemanticAttributes.DB_STATEMENT" statement
- "$SemanticAttributes.DB_OPERATION" operation
- "$SemanticAttributes.DB_CASSANDRA_CONSISTENCY_LEVEL" "LOCAL_ONE"
- "$SemanticAttributes.DB_CASSANDRA_COORDINATOR_DC" "datacenter1"
- "$SemanticAttributes.DB_CASSANDRA_COORDINATOR_ID" String
- "$SemanticAttributes.DB_CASSANDRA_IDEMPOTENCE" Boolean
- "$SemanticAttributes.DB_CASSANDRA_PAGE_SIZE" 5000
- "$SemanticAttributes.DB_CASSANDRA_SPECULATIVE_EXECUTION_COUNT" 0
- // the SqlStatementSanitizer can't handle CREATE statements yet
- "$SemanticAttributes.DB_CASSANDRA_TABLE" table
- }
- }
- }
-
- def getSession(String keyspace) {
- DriverConfigLoader configLoader = DefaultDriverConfigLoader.builder()
- .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(0))
- .build()
- return CqlSession.builder()
- .addContactPoint(new InetSocketAddress("localhost", cassandraPort))
- .withConfigLoader(configLoader)
- .withLocalDatacenter("datacenter1")
- .withKeyspace((String) keyspace)
- .build()
- }
-}

ADD
instrumentation/cassandra/cassandra-4.0/javaagent/src/test/java/CassandraClientTest.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
+import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_CONSISTENCY_LEVEL;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_COORDINATOR_DC;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_COORDINATOR_ID;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_IDEMPOTENCE;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_PAGE_SIZE;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_SPECULATIVE_EXECUTION_COUNT;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_CASSANDRA_TABLE;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_NAME;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_OPERATION;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_STATEMENT;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.DB_SYSTEM;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NET_SOCK_PEER_ADDR;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NET_SOCK_PEER_NAME;
+import static io.opentelemetry.semconv.trace.attributes.SemanticAttributes.NET_SOCK_PEER_PORT;
+import static org.junit.jupiter.api.Named.named;
+
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
+import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
+import com.datastax.oss.driver.internal.core.config.typesafe.DefaultDriverConfigLoader;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+
+public class CassandraClientTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(CassandraClientTest.class);
+
+ @RegisterExtension
+ static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+ @SuppressWarnings("rawtypes")
+ private static GenericContainer cassandra;
+
+ private static int cassandraPort;
+
+ @BeforeAll
+ static void beforeAll() {
+ cassandra =
+ new GenericContainer<>("cassandra:4.0")
+ .withEnv("JVM_OPTS", "-Xmx128m -Xms128m")
+ .withExposedPorts(9042)
+ .withLogConsumer(new Slf4jLogConsumer(logger))
+ .withStartupTimeout(Duration.ofMinutes(2));
+ cassandra.start();
+
+ cassandraPort = cassandra.getMappedPort(9042);
+ }
+
+ @AfterAll
+ static void afterAll() {
+ cassandra.stop();
+ }
+
+ @ParameterizedTest(name = "{index}: {0}")
+ @MethodSource("provideSyncParameters")
+ void syncTest(Parameter parameter) {
+ CqlSession session = getSession(parameter.keyspace);
+
+ session.execute(parameter.statement);
+
+ testing.waitAndAssertTraces(
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span ->
+ span.hasName(parameter.spanName)
+ .hasKind(SpanKind.CLIENT)
+ .hasNoParent()
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_NAME, parameter.keyspace),
+ equalTo(DB_STATEMENT, parameter.expectedStatement),
+ equalTo(DB_OPERATION, parameter.operation),
+ equalTo(DB_CASSANDRA_CONSISTENCY_LEVEL, "LOCAL_ONE"),
+ equalTo(DB_CASSANDRA_COORDINATOR_DC, "datacenter1"),
+ satisfies(
+ DB_CASSANDRA_COORDINATOR_ID, val -> val.isInstanceOf(String.class)),
+ satisfies(
+ DB_CASSANDRA_IDEMPOTENCE, val -> val.isInstanceOf(Boolean.class)),
+ equalTo(DB_CASSANDRA_PAGE_SIZE, 5000),
+ equalTo(DB_CASSANDRA_SPECULATIVE_EXECUTION_COUNT, 0),
+ equalTo(DB_CASSANDRA_TABLE, parameter.table))));
+
+ session.close();
+ }
+
+ @ParameterizedTest(name = "{index}: {0}")
+ @MethodSource("provideAsyncParameters")
+ void asyncTest(Parameter parameter) throws Exception {
+ CqlSession session = getSession(parameter.keyspace);
+
+ testing.runWithSpan(
+ "parent",
+ () ->
+ session
+ .executeAsync(parameter.statement)
+ .toCompletableFuture()
+ .whenComplete((result, throwable) -> testing.runWithSpan("child", () -> {}))
+ .get());
+
+ testing.waitAndAssertTraces(
+ trace ->
+ trace.hasSpansSatisfyingExactly(
+ span -> span.hasName("parent").hasKind(SpanKind.INTERNAL).hasNoParent(),
+ span ->
+ span.hasName(parameter.spanName)
+ .hasKind(SpanKind.CLIENT)
+ .hasParent(trace.getSpan(0))
+ .hasAttributesSatisfyingExactly(
+ equalTo(NET_SOCK_PEER_ADDR, "127.0.0.1"),
+ equalTo(NET_SOCK_PEER_NAME, "localhost"),
+ equalTo(NET_SOCK_PEER_PORT, cassandraPort),
+ equalTo(DB_SYSTEM, "cassandra"),
+ equalTo(DB_NAME, parameter.keyspace),
+ equalTo(DB_STATEMENT, parameter.expectedStatement),
+ equalTo(DB_OPERATION, parameter.operation),
+ equalTo(DB_CASSANDRA_CONSISTENCY_LEVEL, "LOCAL_ONE"),
+ equalTo(DB_CASSANDRA_COORDINATOR_DC, "datacenter1"),
+ satisfies(
+ DB_CASSANDRA_COORDINATOR_ID, val -> val.isInstanceOf(String.class)),
+ satisfies(
+ DB_CASSANDRA_IDEMPOTENCE, val -> val.isInstanceOf(Boolean.class)),
+ equalTo(DB_CASSANDRA_PAGE_SIZE, 5000),
+ equalTo(DB_CASSANDRA_SPECULATIVE_EXECUTION_COUNT, 0),
+ equalTo(DB_CASSANDRA_TABLE, parameter.table)),
+ span ->
+ span.hasName("child").hasKind(SpanKind.INTERNAL).hasParent(trace.getSpan(0))));
+
+ session.close();
+ }
+
+ private static Stream<Arguments> provideSyncParameters() {
+ return Stream.of(
+ Arguments.of(
+ named(
+ "Drop keyspace if exists",
+ new Parameter(
+ null,
+ "DROP KEYSPACE IF EXISTS sync_test",
+ "DROP KEYSPACE IF EXISTS sync_test",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create keyspace with replication",
+ new Parameter(
+ null,
+ "CREATE KEYSPACE sync_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}",
+ "CREATE KEYSPACE sync_test WITH REPLICATION = {?:?, ?:?}",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create table",
+ new Parameter(
+ "sync_test",
+ "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )",
+ "CREATE TABLE sync_test.users ( id UUID PRIMARY KEY, name text )",
+ "sync_test",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Insert data",
+ new Parameter(
+ "sync_test",
+ "INSERT INTO sync_test.users (id, name) values (uuid(), 'alice')",
+ "INSERT INTO sync_test.users (id, name) values (uuid(), ?)",
+ "INSERT sync_test.users",
+ "INSERT",
+ "sync_test.users"))),
+ Arguments.of(
+ named(
+ "Select data",
+ new Parameter(
+ "sync_test",
+ "SELECT * FROM users where name = 'alice' ALLOW FILTERING",
+ "SELECT * FROM users where name = ? ALLOW FILTERING",
+ "SELECT sync_test.users",
+ "SELECT",
+ "users"))));
+ }
+
+ private static Stream<Arguments> provideAsyncParameters() {
+ return Stream.of(
+ Arguments.of(
+ named(
+ "Drop keyspace if exists",
+ new Parameter(
+ null,
+ "DROP KEYSPACE IF EXISTS async_test",
+ "DROP KEYSPACE IF EXISTS async_test",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create keyspace with replication",
+ new Parameter(
+ null,
+ "CREATE KEYSPACE async_test WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':3}",
+ "CREATE KEYSPACE async_test WITH REPLICATION = {?:?, ?:?}",
+ "DB Query",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Create table",
+ new Parameter(
+ "async_test",
+ "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )",
+ "CREATE TABLE async_test.users ( id UUID PRIMARY KEY, name text )",
+ "async_test",
+ null,
+ null))),
+ Arguments.of(
+ named(
+ "Insert data",
+ new Parameter(
+ "async_test",
+ "INSERT INTO async_test.users (id, name) values (uuid(), 'alice')",
+ "INSERT INTO async_test.users (id, name) values (uuid(), ?)",
+ "INSERT async_test.users",
+ "INSERT",
+ "async_test.users"))),
+ Arguments.of(
+ named(
+ "Select data",
+ new Parameter(
+ "async_test",
+ "SELECT * FROM users where name = 'alice' ALLOW FILTERING",
+ "SELECT * FROM users where name = ? ALLOW FILTERING",
+ "SELECT async_test.users",
+ "SELECT",
+ "users"))));
+ }
+
+ private static class Parameter {
+ public final String keyspace;
+ public final String statement;
+ public final String expectedStatement;
+ public final String spanName;
+ public final String operation;
+ public final String table;
+
+ public Parameter(
+ String keyspace,
+ String statement,
+ String expectedStatement,
+ String spanName,
+ String operation,
+ String table) {
+ this.keyspace = keyspace;
+ this.statement = statement;
+ this.expectedStatement = expectedStatement;
+ this.spanName = spanName;
+ this.operation = operation;
+ this.table = table;
+ }
+ }
+
+ CqlSession getSession(String keyspace) {
+ DriverConfigLoader configLoader =
+ DefaultDriverConfigLoader.builder()
+ .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(0))
+ .build();
+ return CqlSession.builder()
+ .addContactPoint(new InetSocketAddress("localhost", cassandraPort))
+ .withConfigLoader(configLoader)
+ .withLocalDatacenter("datacenter1")
+ .withKeyspace(keyspace)
+ .build();
+ }
+}

Cassandra java tests (#7390)