Failing tests WIP
ADD
dd-java-agent/instrumentation/couchbase-2.0/src/test/groovy/CouchbaseAsyncClientTest.groovy
+import com.couchbase.client.java.document.JsonDocument
+import com.couchbase.client.java.document.json.JsonObject
+import com.couchbase.client.java.query.N1qlQuery
+import spock.util.concurrent.BlockingVariable
+import util.AbstractCouchbaseTest
+
+import static datadog.trace.agent.test.utils.TraceUtils.basicSpan
+import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+
+class CouchbaseAsyncClientTest extends AbstractCouchbaseTest {
+ def "test hasBucket #type"() {
+ setup:
+ def hasBucket = new BlockingVariable<Boolean>()
+
+ when:
+ manager.hasBucket(bucketSettings.name()).subscribe({ result -> hasBucket.set(result) })
+
+ then:
+ assert hasBucket.get()
+ assertTraces(1) {
+ trace(0, 1) {
+ assertCouchbaseCall(it, 0, "ClusterManager.hasBucket")
+ }
+ }
+
+ where:
+ manager | cluster | bucketSettings
+ couchbaseAsyncManager | couchbaseAsyncCluster | bucketCouchbase
+ memcacheAsyncManager | memcacheAsyncCluster | bucketMemcache
+
+ type = bucketSettings.type().name()
+ }
+
+ def "test upsert #type"() {
+ setup:
+ JsonObject content = JsonObject.create().put("hello", "world")
+ def inserted = new BlockingVariable<JsonDocument>()
+
+ when:
+ runUnderTrace("someTrace") {
+ // Connect to the bucket and open it
+ cluster.openBucket(bucketSettings.name(), bucketSettings.password()).subscribe({ bkt ->
+ bkt.upsert(JsonDocument.create("helloworld", content)).subscribe({ result -> inserted.set(result) })
+ })
+ }
+
+ then:
+ inserted.get().content().getString("hello") == "world"
+
+ assertTraces(1) {
+ trace(0, 3) {
+ basicSpan(it, 0, "someTrace")
+
+ assertCouchbaseCall(it, 0, "Cluster.openBucket", null, span(0))
+ assertCouchbaseCall(it, 0, "Bucket.upsert", bucketSettings.name(), span(0))
+ }
+ }
+
+ where:
+ manager | cluster | bucketSettings
+ couchbaseAsyncManager | couchbaseAsyncCluster | bucketCouchbase
+ memcacheAsyncManager | memcacheAsyncCluster | bucketMemcache
+
+ type = bucketSettings.type().name()
+ }
+
+ def "test upsert and get #type"() {
+ setup:
+ JsonObject content = JsonObject.create().put("hello", "world")
+ def inserted = new BlockingVariable<JsonDocument>()
+ def found = new BlockingVariable<JsonDocument>()
+
+ when:
+ runUnderTrace("someTrace") {
+ cluster.openBucket(bucketSettings.name(), bucketSettings.password()).subscribe({ bkt ->
+ bkt.upsert(JsonDocument.create("helloworld", content))
+ .subscribe({ result ->
+ inserted.set(result)
+ bkt.get("helloworld")
+ .subscribe({ searchResult -> found.set(searchResult)
+ })
+ })
+ })
+ }
+
+ // Create a JSON document and store it with the ID "helloworld"
+ then:
+ found.get() == inserted.get()
+ found.get().content().getString("hello") == "world"
+
+ assertTraces(1) {
+ trace(0, 4) {
+ basicSpan(it, 0, "someTrace")
+
+ assertCouchbaseCall(it, 0, "Cluster.openBucket", null, span(0))
+ assertCouchbaseCall(it, 0, "Bucket.upsert", bucketSettings.name(), span(0))
+ assertCouchbaseCall(it, 0, "Bucket.get", bucketSettings.name(), span(0))
+ }
+ }
+
+ where:
+ manager | cluster | bucketSettings
+ couchbaseAsyncManager | couchbaseAsyncCluster | bucketCouchbase
+ memcacheAsyncManager | memcacheAsyncCluster | bucketMemcache
+
+ type = bucketSettings.type().name()
+ }
+
+ def "test query"() {
+ setup:
+ def queryResult = new BlockingVariable<JsonObject>()
+
+ when:
+ // Mock expects this specific query.
+ // See com.couchbase.mock.http.query.QueryServer.handleString.
+ runUnderTrace("someTrace") {
+ cluster.openBucket(bucketSettings.name(), bucketSettings.password()).subscribe({
+ bkt ->
+ bkt.query(N1qlQuery.simple("SELECT mockrow"))
+ .flatMap({ query -> query.rows() })
+ .single()
+ .subscribe({ row -> queryResult.set(row.value()) })
+ })
+ }
+
+ then:
+ queryResult.get().get("row") == "value"
+
+ assertTraces(1) {
+ trace(0, 3) {
+ basicSpan(it, 0, "someTrace")
+
+ assertCouchbaseCall(it, 0, "Cluster.openBucket", null, span(0))
+ assertCouchbaseCall(it, 0, "Bucket.query", bucketSettings.name(), span(0))
+ }
+ }
+
+ where:
+ manager | cluster | bucketSettings
+ couchbaseAsyncManager | couchbaseAsyncCluster | bucketCouchbase
+ // Only couchbase buckets support queries.
+
+ type = bucketSettings.type().name()
+ }
+}

MODIFY
dd-java-agent/instrumentation/couchbase-2.0/src/test/groovy/util/AbstractCouchbaseTest.groovy
dd-java-agent/instrumentation/couchbase-2.0/src/test/groovy/util/AbstractCouchbaseTest.groovy
package util
-
import com.couchbase.client.core.metrics.DefaultLatencyMetricsCollectorConfig
import com.couchbase.client.core.metrics.DefaultMetricsCollectorConfig
+import com.couchbase.client.java.CouchbaseAsyncCluster
import com.couchbase.client.java.CouchbaseCluster
import com.couchbase.client.java.bucket.BucketType
+import com.couchbase.client.java.cluster.AsyncClusterManager
import com.couchbase.client.java.cluster.BucketSettings
import com.couchbase.client.java.cluster.ClusterManager
import com.couchbase.client.java.cluster.DefaultBucketSettings
@@ -14,6 +15,7 @@ import com.couchbase.mock.Bucket
import com.couchbase.mock.BucketConfiguration
import com.couchbase.mock.CouchbaseMock
import com.couchbase.mock.http.query.QueryServer
+import datadog.opentracing.DDSpan
import datadog.trace.agent.test.AgentTestRunner
import datadog.trace.agent.test.asserts.TraceAssert
import datadog.trace.agent.test.utils.PortUtils
@@ -59,15 +61,23 @@ abstract class AbstractCouchbaseTest extends AgentTestRunner {
@Shared
protected CouchbaseCluster couchbaseCluster
@Shared
+ protected CouchbaseAsyncCluster couchbaseAsyncCluster
+ @Shared
protected CouchbaseCluster memcacheCluster
@Shared
+ protected CouchbaseAsyncCluster memcacheAsyncCluster
+ @Shared
protected CouchbaseEnvironment couchbaseEnvironment
@Shared
protected CouchbaseEnvironment memcacheEnvironment
@Shared
protected ClusterManager couchbaseManager
@Shared
+ protected AsyncClusterManager couchbaseAsyncManager
+ @Shared
protected ClusterManager memcacheManager
+ @Shared
+ protected AsyncClusterManager memcacheAsyncManager
def setupSpec() {
@@ -80,13 +90,17 @@ abstract class AbstractCouchbaseTest extends AgentTestRunner {
couchbaseEnvironment = envBuilder(bucketCouchbase).build()
couchbaseCluster = CouchbaseCluster.create(couchbaseEnvironment, Arrays.asList("127.0.0.1"))
+ couchbaseAsyncCluster = CouchbaseAsyncCluster.create(couchbaseEnvironment, Arrays.asList("127.0.0.1"))
couchbaseManager = couchbaseCluster.clusterManager(USERNAME, PASSWORD)
+ couchbaseAsyncManager = couchbaseAsyncCluster.clusterManager(USERNAME, PASSWORD).toBlocking().single()
mock.createBucket(convert(bucketMemcache))
memcacheEnvironment = envBuilder(bucketMemcache).build()
memcacheCluster = CouchbaseCluster.create(memcacheEnvironment, Arrays.asList("127.0.0.1"))
+ memcacheAsyncCluster = CouchbaseAsyncCluster.create(memcacheEnvironment, Arrays.asList("127.0.0.1"))
memcacheManager = memcacheCluster.clusterManager(USERNAME, PASSWORD)
+ memcacheAsyncManager = memcacheAsyncCluster.clusterManager(USERNAME, PASSWORD).toBlocking().single()
// Cache buckets:
couchbaseCluster.openBucket(bucketCouchbase.name(), bucketCouchbase.password())
@@ -145,19 +159,23 @@ abstract class AbstractCouchbaseTest extends AgentTestRunner {
.socketConnectTimeout(timeout.intValue())
}
- void assertCouchbaseCall(TraceAssert trace, int index, String name, String bucketName = null) {
+ void assertCouchbaseCall(TraceAssert trace, int index, String name, String bucketName = null, Object parentSpan = null) {
trace.span(index) {
serviceName "couchbase"
resourceName name
operationName "couchbase.call"
spanType DDSpanTypes.COUCHBASE
errored false
+ if (parentSpan == null) {
parent()
+ } else {
+ childOf((DDSpan) parentSpan)
+ }
tags {
"$Tags.COMPONENT.key" "couchbase-client"
"$Tags.DB_TYPE.key" "couchbase"
"$Tags.SPAN_KIND.key" Tags.SPAN_KIND_CLIENT
- if (bucketName) {
+ if (bucketName != null) {
"bucket" bucketName
}
defaultTags()

Failing tests WIP