Migrate Memcache instrumentation to Decorator
MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/spymemcached-2.12.gradle
dd-java-agent/instrumentation/spymemcached-2.12/spymemcached-2.12.gradle
-// building and testing against 2.12 because setListenerExecutorService exists to facilitate easier
-// testing. Instrumentation should work since 2.10
muzzle {
pass {
group = "net.spy"
module = 'spymemcached'
- versions = "[2.10.0,)"
+ versions = "[2.12.0,)"
assertInverse = true
}
}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/BulkGetCompletionListener.java
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/BulkGetCompletionListener.java
package datadog.trace.instrumentation.spymemcached;
import io.opentracing.Span;
-import io.opentracing.Tracer;
import java.util.concurrent.ExecutionException;
+import net.spy.memcached.MemcachedConnection;
import net.spy.memcached.internal.BulkGetFuture;
public class BulkGetCompletionListener extends CompletionListener<BulkGetFuture<?>>
implements net.spy.memcached.internal.BulkGetCompletionListener {
- public BulkGetCompletionListener(final Tracer tracer, final String methodName) {
- super(tracer, methodName, true);
+ public BulkGetCompletionListener(final MemcachedConnection connection, final String methodName) {
+ super(connection, methodName);
}
@Override
@@ -21,7 +21,7 @@ public class BulkGetCompletionListener extends CompletionListener<BulkGetFuture<
throws ExecutionException, InterruptedException {
/*
Note: for now we do not have an affective way of representing results of bulk operations,
- i.e. we cannot day that we got 4 hits out of 10. So we will just ignore results for now.
+ i.e. we cannot say that we got 4 hits out of 10. So we will just ignore results for now.
*/
future.get();
}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/CompletionListener.java
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/CompletionListener.java
package datadog.trace.instrumentation.spymemcached;
-import static io.opentracing.log.Fields.ERROR_OBJECT;
+import static datadog.trace.instrumentation.spymemcached.MemcacheClientDecorator.DECORATE;
-import datadog.trace.api.DDSpanTypes;
-import datadog.trace.api.DDTags;
-import io.opentracing.Scope;
import io.opentracing.Span;
-import io.opentracing.Tracer;
-import io.opentracing.tag.Tags;
-import java.util.Collections;
+import io.opentracing.util.GlobalTracer;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
+import net.spy.memcached.MemcachedConnection;
@Slf4j
public abstract class CompletionListener<T> {
@@ -27,34 +23,18 @@ public abstract class CompletionListener<T> {
static final String HIT = "hit";
static final String MISS = "miss";
- private final Tracer tracer;
- private final Scope scope;
+ private final MemcachedConnection connection;
+ private final Span span;
- public CompletionListener(final Tracer tracer, final String methodName, final boolean async) {
- this.tracer = tracer;
- scope = buildSpan(getOperationName(methodName), async);
- }
-
- private Scope buildSpan(final String operation, final boolean async) {
- final Tracer.SpanBuilder spanBuilder =
- tracer
- .buildSpan(OPERATION_NAME)
- .withTag(DDTags.SERVICE_NAME, SERVICE_NAME)
- .withTag(DDTags.RESOURCE_NAME, operation)
- .withTag(DDTags.SPAN_TYPE, DDSpanTypes.MEMCACHED)
- .withTag(Tags.COMPONENT.getKey(), COMPONENT_NAME)
- .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT)
- .withTag(Tags.DB_TYPE.getKey(), DB_TYPE);
-
- final Scope scope = spanBuilder.startActive(false);
- if (async) {
- scope.close();
- }
- return scope;
+ public CompletionListener(final MemcachedConnection connection, final String methodName) {
+ this.connection = connection;
+ span = GlobalTracer.get().buildSpan(OPERATION_NAME).start();
+ DECORATE.afterStart(span);
+ DECORATE.onConnection(span, connection);
+ DECORATE.onOperation(span, methodName);
}
protected void closeAsyncSpan(final T future) {
- final Span span = scope.span();
try {
processResult(span, future);
} catch (final CancellationException e) {
@@ -64,33 +44,25 @@ public abstract class CompletionListener<T> {
// Looks like underlying OperationFuture wraps CancellationException into ExecutionException
span.setTag(DB_COMMAND_CANCELLED, true);
} else {
- Tags.ERROR.set(span, Boolean.TRUE);
- span.log(Collections.singletonMap(ERROR_OBJECT, e.getCause()));
+ DECORATE.onError(span, e.getCause());
}
} catch (final InterruptedException e) {
// Avoid swallowing InterruptedException
- Tags.ERROR.set(span, Boolean.TRUE);
- span.log(Collections.singletonMap(ERROR_OBJECT, e));
+ DECORATE.onError(span, e);
Thread.currentThread().interrupt();
} catch (final Exception e) {
// This should never happen, just in case to make sure we cover all unexpected exceptions
- Tags.ERROR.set(span, Boolean.TRUE);
- span.log(Collections.singletonMap(ERROR_OBJECT, e));
+ DECORATE.onError(span, e);
} finally {
+ DECORATE.beforeFinish(span);
span.finish();
}
}
protected void closeSyncSpan(final Throwable thrown) {
- final Span span = scope.span();
-
- if (thrown != null) {
- Tags.ERROR.set(span, Boolean.TRUE);
- span.log(Collections.singletonMap(ERROR_OBJECT, thrown));
- }
-
+ DECORATE.onError(span, thrown);
+ DECORATE.beforeFinish(span);
span.finish();
- scope.close();
}
protected abstract void processResult(Span span, T future)
@@ -99,17 +71,4 @@ public abstract class CompletionListener<T> {
protected void setResultTag(final Span span, final boolean hit) {
span.setTag(MEMCACHED_RESULT, hit ? HIT : MISS);
}
-
- private static String getOperationName(final String methodName) {
- final char[] chars =
- methodName
- .replaceFirst("^async", "")
- // 'CAS' name is special, we have to lowercase whole name
- .replaceFirst("^CAS", "cas")
- .toCharArray();
-
- // Lowercase first letter
- chars[0] = Character.toLowerCase(chars[0]);
- return new String(chars);
- }
}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/GetCompletionListener.java
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/GetCompletionListener.java
package datadog.trace.instrumentation.spymemcached;
import io.opentracing.Span;
-import io.opentracing.Tracer;
import java.util.concurrent.ExecutionException;
+import net.spy.memcached.MemcachedConnection;
import net.spy.memcached.internal.GetFuture;
public class GetCompletionListener extends CompletionListener<GetFuture<?>>
implements net.spy.memcached.internal.GetCompletionListener {
- public GetCompletionListener(Tracer tracer, String methodName) {
- super(tracer, methodName, true);
+ public GetCompletionListener(final MemcachedConnection connection, final String methodName) {
+ super(connection, methodName);
}
@Override
- public void onComplete(GetFuture<?> future) {
+ public void onComplete(final GetFuture<?> future) {
closeAsyncSpan(future);
}
@Override
- protected void processResult(Span span, GetFuture<?> future)
+ protected void processResult(final Span span, final GetFuture<?> future)
throws ExecutionException, InterruptedException {
- Object result = future.get();
+ final Object result = future.get();
setResultTag(span, result != null);
}
}

ADD
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/MemcacheClientDecorator.java
+package datadog.trace.instrumentation.spymemcached;
+
+import datadog.trace.agent.decorator.DatabaseClientDecorator;
+import datadog.trace.api.DDSpanTypes;
+import datadog.trace.api.DDTags;
+import io.opentracing.Span;
+import net.spy.memcached.MemcachedConnection;
+
+public class MemcacheClientDecorator extends DatabaseClientDecorator<MemcachedConnection> {
+ public static final MemcacheClientDecorator DECORATE = new MemcacheClientDecorator();
+
+ @Override
+ protected String[] instrumentationNames() {
+ return new String[] {"spymemcached"};
+ }
+
+ @Override
+ protected String service() {
+ return "memcached";
+ }
+
+ @Override
+ protected String component() {
+ return "java-spymemcached";
+ }
+
+ @Override
+ protected String spanType() {
+ return DDSpanTypes.MEMCACHED;
+ }
+
+ @Override
+ protected String dbType() {
+ return "memcached";
+ }
+
+ @Override
+ protected String dbUser(final MemcachedConnection session) {
+ return null;
+ }
+
+ @Override
+ protected String dbInstance(final MemcachedConnection connection) {
+ return connection.connectionsStatus();
+ }
+
+ public Span onOperation(final Span span, final String methodName) {
+
+ final char[] chars =
+ methodName
+ .replaceFirst("^async", "")
+ // 'CAS' name is special, we have to lowercase whole name
+ .replaceFirst("^CAS", "cas")
+ .toCharArray();
+
+ // Lowercase first letter
+ chars[0] = Character.toLowerCase(chars[0]);
+
+ span.setTag(DDTags.RESOURCE_NAME, new String(chars));
+ return span;
+ }
+}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/MemcachedClientInstrumentation.java
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/MemcachedClientInstrumentation.java
@@ -9,8 +9,6 @@ import static net.bytebuddy.matcher.ElementMatchers.returns;
import com.google.auto.service.AutoService;
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.bootstrap.CallDepthThreadLocalMap;
-import io.opentracing.util.GlobalTracer;
-import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import net.bytebuddy.asm.Advice;
@@ -26,8 +24,6 @@ import net.spy.memcached.internal.OperationFuture;
public final class MemcachedClientInstrumentation extends Instrumenter.Default {
private static final String MEMCACHED_PACKAGE = "net.spy.memcached";
- private static final String HELPERS_PACKAGE =
- MemcachedClientInstrumentation.class.getPackage().getName();
public MemcachedClientInstrumentation() {
super("spymemcached");
@@ -41,11 +37,15 @@ public final class MemcachedClientInstrumentation extends Instrumenter.Default {
@Override
public String[] helperClassNames() {
return new String[] {
- HELPERS_PACKAGE + ".CompletionListener",
- HELPERS_PACKAGE + ".SyncCompletionListener",
- HELPERS_PACKAGE + ".GetCompletionListener",
- HELPERS_PACKAGE + ".OperationCompletionListener",
- HELPERS_PACKAGE + ".BulkGetCompletionListener"
+ "datadog.trace.agent.decorator.BaseDecorator",
+ "datadog.trace.agent.decorator.ClientDecorator",
+ "datadog.trace.agent.decorator.DatabaseClientDecorator",
+ packageName + ".MemcacheClientDecorator",
+ packageName + ".CompletionListener",
+ packageName + ".SyncCompletionListener",
+ packageName + ".GetCompletionListener",
+ packageName + ".OperationCompletionListener",
+ packageName + ".BulkGetCompletionListener"
};
}
@@ -84,13 +84,14 @@ public final class MemcachedClientInstrumentation extends Instrumenter.Default {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void methodExit(
@Advice.Enter final boolean shouldInjectListener,
- @Advice.Origin final Method method,
+ @Advice.This final MemcachedClient client,
+ @Advice.Origin("#m") final String methodName,
@Advice.Return final OperationFuture future) {
if (shouldInjectListener && future != null) {
+ CallDepthThreadLocalMap.reset(MemcachedClient.class);
final OperationCompletionListener listener =
- new OperationCompletionListener(GlobalTracer.get(), method.getName());
+ new OperationCompletionListener(client.getConnection(), methodName);
future.addListener(listener);
- CallDepthThreadLocalMap.reset(MemcachedClient.class);
}
}
}
@@ -105,13 +106,14 @@ public final class MemcachedClientInstrumentation extends Instrumenter.Default {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void methodExit(
@Advice.Enter final boolean shouldInjectListener,
- @Advice.Origin final Method method,
+ @Advice.This final MemcachedClient client,
+ @Advice.Origin("#m") final String methodName,
@Advice.Return final GetFuture future) {
if (shouldInjectListener && future != null) {
+ CallDepthThreadLocalMap.reset(MemcachedClient.class);
final GetCompletionListener listener =
- new GetCompletionListener(GlobalTracer.get(), method.getName());
+ new GetCompletionListener(client.getConnection(), methodName);
future.addListener(listener);
- CallDepthThreadLocalMap.reset(MemcachedClient.class);
}
}
}
@@ -126,13 +128,14 @@ public final class MemcachedClientInstrumentation extends Instrumenter.Default {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void methodExit(
@Advice.Enter final boolean shouldInjectListener,
- @Advice.Origin final Method method,
+ @Advice.This final MemcachedClient client,
+ @Advice.Origin("#m") final String methodName,
@Advice.Return final BulkFuture future) {
if (shouldInjectListener && future != null) {
+ CallDepthThreadLocalMap.reset(MemcachedClient.class);
final BulkGetCompletionListener listener =
- new BulkGetCompletionListener(GlobalTracer.get(), method.getName());
+ new BulkGetCompletionListener(client.getConnection(), methodName);
future.addListener(listener);
- CallDepthThreadLocalMap.reset(MemcachedClient.class);
}
}
}
@@ -140,9 +143,10 @@ public final class MemcachedClientInstrumentation extends Instrumenter.Default {
public static class SyncOperationAdvice {
@Advice.OnMethodEnter(suppress = Throwable.class)
- public static SyncCompletionListener methodEnter(@Advice.Origin final Method method) {
+ public static SyncCompletionListener methodEnter(
+ @Advice.This final MemcachedClient client, @Advice.Origin("#m") final String methodName) {
if (CallDepthThreadLocalMap.incrementCallDepth(MemcachedClient.class) <= 0) {
- return new SyncCompletionListener(GlobalTracer.get(), method.getName());
+ return new SyncCompletionListener(client.getConnection(), methodName);
} else {
return null;
}
@@ -153,8 +157,8 @@ public final class MemcachedClientInstrumentation extends Instrumenter.Default {
@Advice.Enter final SyncCompletionListener listener,
@Advice.Thrown final Throwable thrown) {
if (listener != null) {
- listener.done(thrown);
CallDepthThreadLocalMap.reset(MemcachedClient.class);
+ listener.done(thrown);
}
}
}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/OperationCompletionListener.java
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/OperationCompletionListener.java
package datadog.trace.instrumentation.spymemcached;
import io.opentracing.Span;
-import io.opentracing.Tracer;
import java.util.concurrent.ExecutionException;
+import net.spy.memcached.MemcachedConnection;
import net.spy.memcached.internal.OperationFuture;
public class OperationCompletionListener
extends CompletionListener<OperationFuture<? extends Object>>
implements net.spy.memcached.internal.OperationCompletionListener {
- public OperationCompletionListener(Tracer tracer, String methodName) {
- super(tracer, methodName, true);
+ public OperationCompletionListener(
+ final MemcachedConnection connection, final String methodName) {
+ super(connection, methodName);
}
@Override
- public void onComplete(OperationFuture<? extends Object> future) {
+ public void onComplete(final OperationFuture<? extends Object> future) {
closeAsyncSpan(future);
}
@Override
- protected void processResult(Span span, OperationFuture<? extends Object> future)
+ protected void processResult(final Span span, final OperationFuture<? extends Object> future)
throws ExecutionException, InterruptedException {
future.get();
}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/SyncCompletionListener.java
dd-java-agent/instrumentation/spymemcached-2.12/src/main/java/datadog/trace/instrumentation/spymemcached/SyncCompletionListener.java
package datadog.trace.instrumentation.spymemcached;
import io.opentracing.Span;
-import io.opentracing.Tracer;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
+import net.spy.memcached.MemcachedConnection;
@Slf4j
public class SyncCompletionListener extends CompletionListener<Void> {
- public SyncCompletionListener(Tracer tracer, String methodName) {
- super(tracer, methodName, false);
+ public SyncCompletionListener(final MemcachedConnection connection, final String methodName) {
+ super(connection, methodName);
}
@Override
- protected void processResult(Span span, Void future)
+ protected void processResult(final Span span, final Void future)
throws ExecutionException, InterruptedException {
log.error("processResult was called on SyncCompletionListener. This should never happen. ");
}
- public void done(Throwable thrown) {
+ public void done(final Throwable thrown) {
closeSyncSpan(thrown);
}
}

MODIFY
dd-java-agent/instrumentation/spymemcached-2.12/src/test/groovy/datadog/trace/instrumentation/spymemcached/SpymemcachedTest.groovy
dd-java-agent/instrumentation/spymemcached-2.12/src/test/groovy/datadog/trace/instrumentation/spymemcached/SpymemcachedTest.groovy
@@ -635,6 +635,7 @@ class SpymemcachedTest extends AgentTestRunner {
"${Tags.COMPONENT.key}" COMPONENT_NAME
"${Tags.SPAN_KIND.key}" Tags.SPAN_KIND_CLIENT
"${Tags.DB_TYPE.key}" CompletionListener.DB_TYPE
+ "$Tags.DB_INSTANCE.key" ~/Connection Status \{ \w*\/127.0.0.1:\d+ active: true, authed: true, last read: \d+ ms ago }/
if (error == "canceled") {
"${CompletionListener.DB_COMMAND_CANCELLED}" true

Migrate Memcache instrumentation to Decorator