Skip to content

Commit 1aa7d8e

Browse files
richardstartinrichardstartin
committed
add queue type and length to queue events
1 parent ae1aa30 commit 1aa7d8e

File tree

14 files changed

+386
-27
lines changed

14 files changed

+386
-27
lines changed

‎dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/java/concurrent/QueueTimerHelper.java‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,23 +30,31 @@ private static final class RateLimiterHolder {
3030
}
3131

3232
public static <T> void startQueuingTimer(
33-
ContextStore<T, State> taskContextStore, Class<?> schedulerClass, T task) {
33+
ContextStore<T, State> taskContextStore,
34+
Class<?> schedulerClass,
35+
Class<?> queueClass,
36+
int queueLength,
37+
T task) {
3438
State state = taskContextStore.get(task);
35-
startQueuingTimer(state, schedulerClass, task);
39+
startQueuingTimer(state, schedulerClass, queueClass, queueLength, task);
3640
}
3741

38-
public static void startQueuingTimer(State state, Class<?> schedulerClass, Object task) {
42+
public static void startQueuingTimer(
43+
State state, Class<?> schedulerClass, Class<?> queueClass, int queueLength, Object task) {
3944
if (Platform.isNativeImage()) {
4045
// explicitly not supported for Graal native image
4146
return;
4247
}
48+
// TODO consider queue length based sampling here to reduce overhead
4349
// avoid calling this before JFR is initialised because it will lead to reading the wrong
4450
// TSC frequency before JFR has set it up properly
4551
if (task != null && state != null && InstrumentationBasedProfiling.isJFRReady()) {
4652
QueueTiming timing =
4753
(QueueTiming) AgentTracer.get().getProfilingContext().start(Timer.TimerType.QUEUEING);
4854
timing.setTask(task);
4955
timing.setScheduler(schedulerClass);
56+
timing.setQueue(queueClass);
57+
timing.setQueueLength(queueLength);
5058
state.setTiming(timing);
5159
}
5260
}

‎dd-java-agent/agent-profiling/profiling-controller-openjdk/src/main/java/com/datadog/profiling/controller/openjdk/events/QueueTimeEvent.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,12 @@ public class QueueTimeEvent extends Event implements QueueTiming {
3333
@Label("Scheduler")
3434
private Class<?> scheduler;
3535

36+
@Label("Queue")
37+
private Class<?> queueType;
38+
39+
@Label("Queue Length on Entry")
40+
private int queueLength;
41+
3642
public QueueTimeEvent() {
3743
this.origin = Thread.currentThread();
3844
AgentSpan activeSpan = AgentTracer.activeSpan();
@@ -55,6 +61,16 @@ public void setScheduler(Class<?> scheduler) {
5561
this.scheduler = scheduler;
5662
}
5763

64+
@Override
65+
public void setQueue(Class<?> queueType) {
66+
this.queueType = queueType;
67+
}
68+
69+
@Override
70+
public void setQueueLength(int queueLength) {
71+
this.queueLength = queueLength;
72+
}
73+
5874
@Override
5975
public void report() {
6076
commit();

‎dd-java-agent/agent-profiling/profiling-ddprof/src/main/java/com/datadog/profiling/ddprof/QueueTimeTracker.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@ public class QueueTimeTracker implements QueueTiming {
1313
// FIXME this can be eliminated by altering the instrumentation
1414
// since it is known when the item is polled from the queue
1515
private Class<?> scheduler;
16+
private Class<?> queue;
17+
private int queueLength;
1618

1719
public QueueTimeTracker(DatadogProfiler profiler, long startTicks) {
1820
this.profiler = profiler;
@@ -31,6 +33,16 @@ public void setScheduler(Class<?> scheduler) {
3133
this.scheduler = scheduler;
3234
}
3335

36+
@Override
37+
public void setQueue(Class<?> queue) {
38+
this.queue = queue;
39+
}
40+
41+
@Override
42+
public void setQueueLength(int queueLength) {
43+
this.queueLength = queueLength;
44+
}
45+
3446
@Override
3547
public void report() {
3648
assert weakTask != null && scheduler != null;

‎dd-java-agent/instrumentation/grpc-1.5/src/main/java/datadog/trace/instrumentation/grpc/QueuedCommandInstrumentation.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import java.nio.channels.Channel;
2121
import java.util.Collections;
2222
import java.util.Map;
23+
import java.util.concurrent.ConcurrentLinkedQueue;
2324
import net.bytebuddy.asm.Advice;
2425

2526
@AutoService(InstrumenterModule.class)
@@ -66,7 +67,14 @@ public static final class Construct {
6667
public static void after(@Advice.This Object command) {
6768
ContextStore<Object, State> contextStore = InstrumentationContext.get(QUEUED_COMMAND, STATE);
6869
capture(contextStore, command);
69-
QueueTimerHelper.startQueuingTimer(contextStore, Channel.class, command);
70+
// FIXME hard to handle both the lifecyle and get access to the queue instance in the same
71+
// frame within the WriteQueue class.
72+
// This means we can't get the queue length. A (bad) alternative would be to instrument
73+
// ConcurrentLinkedQueue broadly,
74+
// or we could write more brittle instrumentation targeting code patterns in different gRPC
75+
// versions.
76+
QueueTimerHelper.startQueuingTimer(
77+
contextStore, Channel.class, ConcurrentLinkedQueue.class, 0, command);
7078
}
7179
}
7280

‎dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/executor/ThreadPoolExecutorInstrumentation.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import java.util.Collections;
3232
import java.util.HashMap;
3333
import java.util.Map;
34+
import java.util.Queue;
3435
import java.util.concurrent.RunnableFuture;
3536
import java.util.concurrent.ThreadPoolExecutor;
3637
import net.bytebuddy.asm.Advice;
@@ -162,12 +163,20 @@ public static void capture(
162163
// excluded as
163164
// Runnables but it is not until now that they will be put on the executor's queue
164165
if (!exclude(RUNNABLE, task)) {
166+
Queue<?> queue = tpe.getQueue();
165167
QueueTimerHelper.startQueuingTimer(
166-
InstrumentationContext.get(Runnable.class, State.class), tpe.getClass(), task);
168+
InstrumentationContext.get(Runnable.class, State.class),
169+
tpe.getClass(),
170+
queue.getClass(),
171+
queue.size(),
172+
task);
167173
} else if (!exclude(RUNNABLE_FUTURE, task) && task instanceof RunnableFuture) {
174+
Queue<?> queue = tpe.getQueue();
168175
QueueTimerHelper.startQueuingTimer(
169176
InstrumentationContext.get(RunnableFuture.class, State.class),
170177
tpe.getClass(),
178+
queue.getClass(),
179+
queue.size(),
171180
(RunnableFuture<?>) task);
172181
}
173182
}

‎dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/forkjoin/JavaForkJoinPoolInstrumentation.java‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,8 @@
1616
import datadog.trace.agent.tooling.InstrumenterModule;
1717
import datadog.trace.bootstrap.ContextStore;
1818
import datadog.trace.bootstrap.InstrumentationContext;
19-
import datadog.trace.bootstrap.instrumentation.java.concurrent.QueueTimerHelper;
2019
import datadog.trace.bootstrap.instrumentation.java.concurrent.State;
2120
import java.util.Map;
22-
import java.util.concurrent.ForkJoinPool;
2321
import java.util.concurrent.ForkJoinTask;
2422
import net.bytebuddy.asm.Advice;
2523

@@ -53,13 +51,11 @@ public void methodAdvice(MethodTransformer transformer) {
5351
public static final class ExternalPush {
5452
@SuppressWarnings("rawtypes")
5553
@Advice.OnMethodEnter
56-
public static <T> void externalPush(
57-
@Advice.This ForkJoinPool pool, @Advice.Argument(0) ForkJoinTask<T> task) {
54+
public static <T> void externalPush(@Advice.Argument(0) ForkJoinTask<T> task) {
5855
if (!exclude(FORK_JOIN_TASK, task)) {
5956
ContextStore<ForkJoinTask, State> contextStore =
6057
InstrumentationContext.get(ForkJoinTask.class, State.class);
6158
capture(contextStore, task);
62-
QueueTimerHelper.startQueuingTimer(contextStore, pool.getClass(), task);
6359
}
6460
}
6561

@@ -74,13 +70,11 @@ public static <T> void cleanup(
7470

7571
public static final class PoolSubmit {
7672
@Advice.OnMethodEnter
77-
public static <T> void poolSubmit(
78-
@Advice.This ForkJoinPool pool, @Advice.Argument(1) ForkJoinTask<T> task) {
73+
public static <T> void poolSubmit(@Advice.Argument(1) ForkJoinTask<T> task) {
7974
if (!exclude(FORK_JOIN_TASK, task)) {
8075
ContextStore<ForkJoinTask, State> contextStore =
8176
InstrumentationContext.get(ForkJoinTask.class, State.class);
8277
capture(contextStore, task);
83-
QueueTimerHelper.startQueuingTimer(contextStore, pool.getClass(), task);
8478
}
8579
}
8680

‎dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/forkjoin/JavaForkJoinWorkQueueInstrumentation.java‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
package datadog.trace.instrumentation.java.concurrent.forkjoin;
2+
3+
import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
4+
import static datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter.ExcludeType.FORK_JOIN_TASK;
5+
import static datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter.exclude;
6+
import static datadog.trace.instrumentation.java.concurrent.ConcurrentInstrumentationNames.EXECUTOR_INSTRUMENTATION_NAME;
7+
import static datadog.trace.instrumentation.java.concurrent.ConcurrentInstrumentationNames.FORK_JOIN_POOL_INSTRUMENTATION_NAME;
8+
import static java.util.Collections.singletonMap;
9+
import static net.bytebuddy.matcher.ElementMatchers.declaresField;
10+
import static net.bytebuddy.matcher.ElementMatchers.fieldType;
11+
import static net.bytebuddy.matcher.ElementMatchers.isDeclaredBy;
12+
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
13+
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
14+
15+
import com.google.auto.service.AutoService;
16+
import datadog.trace.agent.tooling.Instrumenter;
17+
import datadog.trace.agent.tooling.InstrumenterModule;
18+
import datadog.trace.api.config.ProfilingConfig;
19+
import datadog.trace.bootstrap.ContextStore;
20+
import datadog.trace.bootstrap.InstrumentationContext;
21+
import datadog.trace.bootstrap.config.provider.ConfigProvider;
22+
import datadog.trace.bootstrap.instrumentation.java.concurrent.QueueTimerHelper;
23+
import datadog.trace.bootstrap.instrumentation.java.concurrent.State;
24+
import java.util.Map;
25+
import java.util.concurrent.ForkJoinPool;
26+
import java.util.concurrent.ForkJoinTask;
27+
import net.bytebuddy.asm.Advice;
28+
29+
@AutoService(InstrumenterModule.class)
30+
public class JavaForkJoinWorkQueueInstrumentation extends InstrumenterModule.Profiling
31+
implements Instrumenter.ForBootstrap, Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
32+
33+
public JavaForkJoinWorkQueueInstrumentation() {
34+
super(
35+
EXECUTOR_INSTRUMENTATION_NAME,
36+
FORK_JOIN_POOL_INSTRUMENTATION_NAME,
37+
FORK_JOIN_POOL_INSTRUMENTATION_NAME + "-workqueue");
38+
}
39+
40+
@Override
41+
public String instrumentedType() {
42+
return "java.util.concurrent.ForkJoinPool$WorkQueue";
43+
}
44+
45+
@Override
46+
public boolean isEnabled() {
47+
return super.isEnabled()
48+
&& ConfigProvider.getInstance()
49+
.getBoolean(
50+
ProfilingConfig.PROFILING_QUEUEING_TIME_ENABLED,
51+
ProfilingConfig.PROFILING_QUEUEING_TIME_ENABLED_DEFAULT);
52+
}
53+
54+
@Override
55+
public Map<String, String> contextStore() {
56+
return singletonMap("java.util.concurrent.ForkJoinTask", State.class.getName());
57+
}
58+
59+
@Override
60+
public void methodAdvice(MethodTransformer transformer) {
61+
String name = getClass().getName();
62+
transformer.applyAdvice(
63+
isMethod()
64+
.and(named("push"))
65+
.and(takesArgument(0, named("java.util.concurrent.ForkJoinTask")))
66+
.and(
67+
isDeclaredBy(
68+
declaresField(fieldType(int.class).and(named("top")))
69+
.and(declaresField(fieldType(int.class).and(named("base")))))),
70+
name + "$PushTask");
71+
}
72+
73+
public static final class PushTask {
74+
@SuppressWarnings("rawtypes")
75+
@Advice.OnMethodEnter
76+
public static <T> void push(
77+
@Advice.This Object workQueue,
78+
@Advice.FieldValue("top") int top,
79+
@Advice.FieldValue("base") int base,
80+
@Advice.Argument(0) ForkJoinTask<T> task) {
81+
if (!exclude(FORK_JOIN_TASK, task)) {
82+
ContextStore<ForkJoinTask, State> contextStore =
83+
InstrumentationContext.get(ForkJoinTask.class, State.class);
84+
QueueTimerHelper.startQueuingTimer(
85+
contextStore, ForkJoinPool.class, workQueue.getClass(), top - base, task);
86+
}
87+
}
88+
}
89+
}

‎dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/timer/JavaTimerInstrumentation.java‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
import static datadog.trace.bootstrap.instrumentation.java.concurrent.AdviceUtils.capture;
66
import static datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter.ExcludeType.RUNNABLE;
77
import static datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter.exclude;
8-
import static datadog.trace.bootstrap.instrumentation.java.concurrent.QueueTimerHelper.startQueuingTimer;
98
import static datadog.trace.instrumentation.java.concurrent.ConcurrentInstrumentationNames.EXECUTOR_INSTRUMENTATION_NAME;
109
import static datadog.trace.instrumentation.java.concurrent.ConcurrentInstrumentationNames.RUNNABLE_INSTRUMENTATION_NAME;
1110
import static java.util.Collections.singletonMap;
@@ -21,7 +20,6 @@
2120
import datadog.trace.bootstrap.InstrumentationContext;
2221
import datadog.trace.bootstrap.instrumentation.java.concurrent.State;
2322
import java.util.Map;
24-
import java.util.Timer;
2523
import java.util.TimerTask;
2624
import net.bytebuddy.asm.Advice;
2725

@@ -67,7 +65,6 @@ public static void before(@Advice.Argument(0) TimerTask task, @Advice.Argument(2
6765
ContextStore<Runnable, State> contextStore =
6866
InstrumentationContext.get(Runnable.class, State.class);
6967
capture(contextStore, task);
70-
startQueuingTimer(contextStore, Timer.class, task);
7168
}
7269
}
7370

‎dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/timer/TaskQueueInstrumentation.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
package datadog.trace.instrumentation.java.concurrent.timer;
2+
3+
import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
4+
import static datadog.trace.instrumentation.java.concurrent.ConcurrentInstrumentationNames.EXECUTOR_INSTRUMENTATION_NAME;
5+
import static datadog.trace.instrumentation.java.concurrent.ConcurrentInstrumentationNames.RUNNABLE_INSTRUMENTATION_NAME;
6+
import static java.util.Collections.singletonMap;
7+
import static net.bytebuddy.matcher.ElementMatchers.declaresField;
8+
import static net.bytebuddy.matcher.ElementMatchers.fieldType;
9+
import static net.bytebuddy.matcher.ElementMatchers.isDeclaredBy;
10+
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
11+
import static net.bytebuddy.matcher.ElementMatchers.isPrivate;
12+
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
13+
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
14+
15+
import com.google.auto.service.AutoService;
16+
import datadog.trace.agent.tooling.Instrumenter;
17+
import datadog.trace.agent.tooling.InstrumenterModule;
18+
import datadog.trace.api.config.ProfilingConfig;
19+
import datadog.trace.bootstrap.ContextStore;
20+
import datadog.trace.bootstrap.InstrumentationContext;
21+
import datadog.trace.bootstrap.config.provider.ConfigProvider;
22+
import datadog.trace.bootstrap.instrumentation.java.concurrent.QueueTimerHelper;
23+
import datadog.trace.bootstrap.instrumentation.java.concurrent.State;
24+
import java.util.Map;
25+
import java.util.Timer;
26+
import java.util.TimerTask;
27+
import net.bytebuddy.asm.Advice;
28+
29+
@AutoService(InstrumenterModule.class)
30+
public class TaskQueueInstrumentation extends InstrumenterModule.Profiling
31+
implements Instrumenter.ForBootstrap, Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
32+
public TaskQueueInstrumentation() {
33+
super("java_timer", EXECUTOR_INSTRUMENTATION_NAME, RUNNABLE_INSTRUMENTATION_NAME, "task-queue");
34+
}
35+
36+
@Override
37+
public boolean isEnabled() {
38+
return super.isEnabled()
39+
&& ConfigProvider.getInstance()
40+
.getBoolean(
41+
ProfilingConfig.PROFILING_QUEUEING_TIME_ENABLED,
42+
ProfilingConfig.PROFILING_QUEUEING_TIME_ENABLED_DEFAULT);
43+
}
44+
45+
@Override
46+
public String instrumentedType() {
47+
return "java.util.TaskQueue";
48+
}
49+
50+
@Override
51+
public Map<String, String> contextStore() {
52+
return singletonMap("java.util.TimerTask", State.class.getName());
53+
}
54+
55+
@Override
56+
public void methodAdvice(MethodTransformer transformer) {
57+
transformer.applyAdvice(
58+
isMethod()
59+
.and(isPrivate())
60+
.and(
61+
named("add")
62+
.and(takesArguments(1))
63+
.and(takesArgument(0, named("java.util.TimerTask")))
64+
.and(isDeclaredBy(declaresField(fieldType(int.class).and(named("size")))))),
65+
getClass().getName() + "$AddTask");
66+
}
67+
68+
public static final class AddTask {
69+
@Advice.OnMethodEnter(suppress = Throwable.class)
70+
public static void before(
71+
@Advice.This Object taskQueue,
72+
@Advice.FieldValue("size") int size,
73+
@Advice.Argument(0) TimerTask task) {
74+
ContextStore<TimerTask, State> contextStore =
75+
InstrumentationContext.get(TimerTask.class, State.class);
76+
QueueTimerHelper.startQueuingTimer(
77+
contextStore, Timer.class, taskQueue.getClass(), size, task);
78+
}
79+
}
80+
}

0 commit comments

Comments
 (0)