diff --git a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/CodeOriginProbe.java b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/CodeOriginProbe.java index 5fef3ff9ffe..c8d6e4a49e2 100644 --- a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/CodeOriginProbe.java +++ b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/CodeOriginProbe.java @@ -126,6 +126,6 @@ public int hashCode() { public String toString() { return String.format( "CodeOriginProbe{probeId=%s, entrySpanProbe=%s, signature=%s, where=%s, location=%s}", - probeId, entrySpanProbe, signature, where, location); + getProbeId(), entrySpanProbe, signature, where, location); } } diff --git a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/LogProbe.java b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/LogProbe.java index b85f38401f2..52f1730416e 100644 --- a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/LogProbe.java +++ b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/LogProbe.java @@ -1,6 +1,8 @@ package com.datadog.debugger.probe; import static com.datadog.debugger.probe.LogProbe.Capture.toLimits; +import static com.datadog.debugger.probe.LogProbe.CoordinatedSamplingState.Status.DROP; +import static com.datadog.debugger.probe.LogProbe.CoordinatedSamplingState.Status.EMIT; import static datadog.trace.api.debugger.DebuggerMetricCollector.SkippedReason.RATE_LIMIT; import static java.lang.String.format; @@ -24,6 +26,8 @@ import com.squareup.moshi.JsonAdapter; import com.squareup.moshi.JsonReader; import com.squareup.moshi.JsonWriter; +import datadog.context.Context; +import datadog.context.ContextKey; import datadog.trace.api.Config; import datadog.trace.api.CorrelationIdentifier; import datadog.trace.api.DDTraceId; @@ -52,6 +56,8 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import org.slf4j.Logger; @@ -62,7 +68,8 @@ public class LogProbe extends ProbeDefinition implements Sampled, CapturedContex private static final Logger LOGGER = LoggerFactory.getLogger(LogProbe.class); private static final Limits LIMITS = new Limits(1, 3, 8192, 5); private static final int LOG_MSG_LIMIT = 8192; - + private static final ContextKey SAMPLING_KEY = + ContextKey.named("debugger_sampling"); public static final int CAPTURING_PROBE_BUDGET = 10; public static final int NON_CAPTURING_PROBE_BUDGET = 1000; @@ -511,7 +518,7 @@ public InstrumentationResult.Status instrument( public boolean isReadyToCapture() { if (!hasCondition()) { // we are sampling here to avoid creating CapturedContext when the sampling result is negative - boolean sampled = ProbeRateLimiter.tryProbe(sampler, isFullSnapshot()); + boolean sampled = trySample(sampler); if (!sampled) { DebuggerAgent.getSink().skipSnapshot(id, RATE_LIMIT); } @@ -585,9 +592,7 @@ private void sample(LogStatus logStatus, MethodLocation methodLocation) { // at 1/s rate instead of the log template one Sampler localSampler = logStatus.hasConditionErrors && !isFullSnapshot() ? errorSampler : sampler; - boolean sampled = - !logStatus.getDebugSessionStatus().isDisabled() - && ProbeRateLimiter.tryProbe(localSampler, isFullSnapshot()); + boolean sampled = !logStatus.getDebugSessionStatus().isDisabled() && trySample(localSampler); logStatus.setSampled(sampled); if (!sampled && !logStatus.getDebugSessionStatus().isDisabled()) { DebuggerAgent.getSink().skipSnapshot(id, RATE_LIMIT); @@ -642,6 +647,71 @@ public void commit( } } + /** + * Holds the once-per-trace coordinated sampling decision shared by all full-snapshot probes on + * a given local root span, so that either all of them emit or none of them do. + */ + static class CoordinatedSamplingState { + /** Outcome of the first probe's sampling decision for the trace, cached on the root span. */ + enum Status { + /** The trace was not sampled; every probe sharing this state must not emit. */ + DROP, + /** The trace was sampled; probes sharing this state may emit, once each. */ + EMIT + } + + private final Set emittedProbeIds; + private final Status status; + + CoordinatedSamplingState(Status status) { + this.status = status; + if (status == EMIT) { + emittedProbeIds = ConcurrentHashMap.newKeySet(); + } else { + emittedProbeIds = Collections.emptySet(); + } + } + + boolean tryEmit(String probeEncodedId) { + return status == EMIT && emittedProbeIds.add(probeEncodedId); + } + } + + private boolean trySample(Sampler sampler) { + if (!isFullSnapshot()) { + return ProbeRateLimiter.tryProbe(sampler, false); + } + + AgentSpan localRootSpan = getActiveLocalRootSpan(); + if (localRootSpan == null) { + return ProbeRateLimiter.tryProbe(sampler, true); + } + + CoordinatedSamplingState state = Context.from(localRootSpan).get(SAMPLING_KEY); + if (state == null) { + synchronized (localRootSpan) { + Context traceContext = Context.from(localRootSpan); + state = traceContext.get(SAMPLING_KEY); + if (state == null) { + boolean sampled = ProbeRateLimiter.tryProbe(sampler, true); + state = new CoordinatedSamplingState(sampled ? EMIT : DROP); + traceContext.with(SAMPLING_KEY, state).attachTo(localRootSpan); + } + } + } + return state.tryEmit(getProbeId().getEncodedId()); + } + + private AgentSpan getActiveLocalRootSpan() { + TracerAPI tracer = AgentTracer.get(); + AgentSpan activeSpan = tracer != null ? tracer.activeSpan() : null; + if (activeSpan == null || activeSpan == AgentTracer.noopSpan()) { + return null; + } + AgentSpan localRootSpan = activeSpan.getLocalRootSpan(); + return localRootSpan != AgentTracer.noopSpan() ? localRootSpan : null; + } + protected Snapshot createSnapshot() { int maxDepth = capture != null ? capture.maxReferenceDepth : -1; return new Snapshot(Thread.currentThread(), this, maxDepth); @@ -652,8 +722,9 @@ protected boolean fillSnapshot( CapturedContext exitContext, List caughtExceptions, Snapshot snapshot) { - LogStatus entryStatus = convertStatus(entryContext.getStatus(probeId.getEncodedId())); - LogStatus exitStatus = convertStatus(exitContext.getStatus(probeId.getEncodedId())); + String probeEncodedId = getProbeId().getEncodedId(); + LogStatus entryStatus = convertStatus(entryContext.getStatus(probeEncodedId)); + LogStatus exitStatus = convertStatus(exitContext.getStatus(probeEncodedId)); String message = null; switch (evaluateAt) { case ENTRY: @@ -829,7 +900,7 @@ protected void commitSnapshot(Snapshot snapshot, DebuggerSink sink) { @Override public void commit(CapturedContext lineContext, int line) { - LogStatus status = (LogStatus) lineContext.getStatus(probeId.getEncodedId()); + LogStatus status = (LogStatus) lineContext.getStatus(getProbeId().getEncodedId()); if (status == null) { return; } diff --git a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/ProbeDefinition.java b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/ProbeDefinition.java index 707ea9ee0de..39cde2b8938 100644 --- a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/ProbeDefinition.java +++ b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/ProbeDefinition.java @@ -28,7 +28,7 @@ public abstract class ProbeDefinition implements ProbeImplementation { protected final String language; protected final String id; protected final int version; - protected transient ProbeId probeId; + private transient ProbeId probeId; protected final Tag[] tags; protected final Map tagMap = new HashMap<>(); protected final Where where; diff --git a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/SpanDecorationProbe.java b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/SpanDecorationProbe.java index 942f1265a49..19abfb97235 100644 --- a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/SpanDecorationProbe.java +++ b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/SpanDecorationProbe.java @@ -273,10 +273,11 @@ public void commit( CapturedContext entryContext, CapturedContext exitContext, List caughtExceptions) { + String probeEncodedId = getProbeId().getEncodedId(); CapturedContext.Status status = evaluateAt == MethodLocation.EXIT - ? exitContext.getStatus(probeId.getEncodedId()) - : entryContext.getStatus(probeId.getEncodedId()); + ? exitContext.getStatus(probeEncodedId) + : entryContext.getStatus(probeEncodedId); if (status == null) { return; } @@ -287,7 +288,7 @@ public void commit( @Override public void commit(CapturedContext lineContext, int line) { - CapturedContext.Status status = lineContext.getStatus(probeId.getEncodedId()); + CapturedContext.Status status = lineContext.getStatus(getProbeId().getEncodedId()); if (status == null) { return; } @@ -319,7 +320,7 @@ private void decorateTags(SpanDecorationStatus status) { } if (!tagsToDecorate.isEmpty()) { // only send EMITTING status if we set at least one tag - DebuggerAgent.getSink().getProbeStatusSink().addEmitting(probeId); + DebuggerAgent.getSink().getProbeStatusSink().addEmitting(getProbeId()); } } diff --git a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/TriggerProbe.java b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/TriggerProbe.java index 945b2a052dc..e20f7c39d01 100644 --- a/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/TriggerProbe.java +++ b/dd-java-agent/agent-debugger/src/main/java/com/datadog/debugger/probe/TriggerProbe.java @@ -140,7 +140,7 @@ private boolean evaluateCondition(CapturedContext capture) { Duration timeout = Duration.ofMillis(Config.get().getDynamicInstrumentationEvalTimeout()); return probeCondition.execute(capture, TimeoutChecker.create(Config.get(), timeout)); } catch (Exception ex) { - DebuggerAgent.getSink().getProbeStatusSink().addError(probeId, ex); + DebuggerAgent.getSink().getProbeStatusSink().addError(getProbeId(), ex); return false; } finally { LOGGER.debug( @@ -199,7 +199,7 @@ public String toString() { language, location, probeCondition, - probeId, + getProbeId(), sampling, tagMap, Arrays.toString(tags), diff --git a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturedSnapshotTest.java b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturedSnapshotTest.java index 5144cc9f7c3..a800e1cb1e7 100644 --- a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturedSnapshotTest.java +++ b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturedSnapshotTest.java @@ -18,6 +18,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.params.provider.Arguments.arguments; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -444,15 +445,6 @@ public void multiProbeSameMethod() throws IOException, URISyntaxException { assertCaptureReturnValue(snapshot1.getCaptures().getReturn(), "int", "31"); } - private List assertSnapshots( - TestSnapshotListener listener, int expectedCount, ProbeId... probeIds) { - assertEquals(expectedCount, listener.snapshots.size()); - for (int i = 0; i < probeIds.length; i++) { - assertEquals(probeIds[i].getId(), listener.snapshots.get(i).getProbe().getId()); - } - return listener.snapshots; - } - @Test public void catchBlock() throws IOException, URISyntaxException { final String CLASS_NAME = "CapturedSnapshot02"; @@ -908,9 +900,14 @@ public void fieldExtractorDepth1() throws IOException, URISyntaxException { @Test public void fieldExtractorDuplicateUnionDepth() throws IOException, URISyntaxException { final String CLASS_NAME = "CapturedSnapshot04"; - LogProbe.Builder builder = createProbeBuilder(PROBE_ID, CLASS_NAME, "createSimpleData", "()"); - LogProbe probe1 = builder.capture(0, 100, 50, Limits.DEFAULT_FIELD_COUNT).build(); - LogProbe probe2 = builder.capture(3, 100, 50, Limits.DEFAULT_FIELD_COUNT).build(); + LogProbe probe1 = + createProbeBuilder(PROBE_ID1, CLASS_NAME, "createSimpleData", "()") + .capture(0, 100, 50, Limits.DEFAULT_FIELD_COUNT) + .build(); + LogProbe probe2 = + createProbeBuilder(PROBE_ID2, CLASS_NAME, "createSimpleData", "()") + .capture(3, 100, 50, Limits.DEFAULT_FIELD_COUNT) + .build(); TestSnapshotListener listener = installProbes(probe1, probe2); Class testClass = compileAndLoadClass(CLASS_NAME); int result = Reflect.onClass(testClass).call("main", "").get(); @@ -2330,7 +2327,8 @@ public void enumConstructorArgs() throws IOException, URISyntaxException { final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot23"; final String ENUM_CLASS = CLASS_NAME + "$MyEnum"; TestSnapshotListener listener = - installProbes(createMethodProbe(PROBE_ID, ENUM_CLASS, "", null)); + installProbes( + createProbeBuilder(PROBE_ID, ENUM_CLASS, "", null).sampling(10).build()); Class testClass = compileAndLoadClass(CLASS_NAME); int result = Reflect.onClass(testClass).call("main", "").get(); assertEquals(2, result); diff --git a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturingTestBase.java b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturingTestBase.java index 4fa605c2cc1..0236897cb96 100644 --- a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturingTestBase.java +++ b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CapturingTestBase.java @@ -14,6 +14,7 @@ import com.datadog.debugger.probe.Sampled; import com.datadog.debugger.sink.DebuggerSink; import com.datadog.debugger.sink.ProbeStatusSink; +import com.datadog.debugger.sink.Snapshot; import com.datadog.debugger.util.MoshiHelper; import com.datadog.debugger.util.MoshiSnapshotTestHelper; import com.datadog.debugger.util.SerializerWithLimits; @@ -160,6 +161,15 @@ private static Collection getCollection(CapturedContext.CapturedValue capture } } + protected List assertSnapshots( + TestSnapshotListener listener, int expectedCount, ProbeId... probeIds) { + assertEquals(expectedCount, listener.snapshots.size()); + for (int i = 0; i < probeIds.length; i++) { + assertEquals(probeIds[i].getId(), listener.snapshots.get(i).getProbe().getId()); + } + return listener.snapshots; + } + protected void assertCaptureFields( CapturedContext context, String name, String typeName, Map expectedMap) { CapturedContext.CapturedValue field = getFields(context.getArguments().get("this")).get(name); diff --git a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CoordinatedSamplingTest.java b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CoordinatedSamplingTest.java new file mode 100644 index 00000000000..181873a3f96 --- /dev/null +++ b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/CoordinatedSamplingTest.java @@ -0,0 +1,357 @@ +package com.datadog.debugger.agent; + +import static com.datadog.debugger.el.expressions.BooleanExpression.FALSE; +import static com.datadog.debugger.el.expressions.BooleanExpression.TRUE; +import static com.datadog.debugger.util.LogProbeTestHelper.parseTemplate; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.params.provider.Arguments.arguments; +import static utils.InstrumentationTestHelper.compileAndLoadClass; +import static utils.InstrumentationTestHelper.getLineForLineProbe; + +import com.datadog.debugger.el.DSL; +import com.datadog.debugger.el.ProbeCondition; +import com.datadog.debugger.el.expressions.BooleanExpression; +import com.datadog.debugger.probe.LogProbe; +import com.datadog.debugger.util.TestSnapshotListener; +import datadog.context.Context; +import datadog.trace.agent.tooling.TracerInstaller; +import datadog.trace.bootstrap.debugger.ProbeId; +import datadog.trace.bootstrap.debugger.ProbeRateLimiter; +import datadog.trace.core.CoreTracer; +import java.io.IOException; +import java.net.URISyntaxException; +import java.util.stream.Stream; +import org.joor.Reflect; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +public class CoordinatedSamplingTest extends CapturingTestBase { + private static final ProbeId PROBE_ID1 = new ProbeId("beae1807-f3b0-4ea8-a74f-826790c5e6f6", 0); + private static final ProbeId PROBE_ID2 = new ProbeId("beae1807-f3b0-4ea8-a74f-826790c5e6f7", 0); + private static final ProbeId PROBE_ID3 = new ProbeId("beae1807-f3b0-4ea8-a74f-826790c5e6f8", 0); + private static final ProbeId LINE_PROBE_ID1 = + new ProbeId("beae1817-f3b0-4ea8-a74f-000000000001", 0); + private static final ProbeId LINE_PROBE_ID2 = + new ProbeId("beae1817-f3b0-4ea8-a74f-000000000002", 0); + private static final ProbeId LINE_PROBE_ID3 = + new ProbeId("beae1817-f3b0-4ea8-a74f-000000000003", 0); + + interface TestListenerMethod { + TestSnapshotListener run() throws IOException, URISyntaxException; + } + + @Test + public void coordinatedSamplingFirstEmit() throws IOException, URISyntaxException { + TestSnapshotListener listener = doCoordinatedSamplingTest(this::coordinatedSampling, 1); + assertSnapshots(listener, 3, PROBE_ID3, PROBE_ID2, PROBE_ID1); + assertSame(Context.root(), Context.current()); + } + + @Test + public void coordinatedSamplingFirstDrop() throws IOException, URISyntaxException { + TestSnapshotListener listener = doCoordinatedSamplingTest(this::coordinatedSampling, 0); + assertSnapshots(listener, 0); + } + + @ParameterizedTest + @MethodSource("coordinatedSamplingConditionSource") + public void coordinatedSamplingCondition( + BooleanExpression cond1, + BooleanExpression cond2, + BooleanExpression cond3, + int numSamples, + int expectedSnapshots, + ProbeId... probeIds) + throws IOException, URISyntaxException { + TestSnapshotListener listener = + doCoordinatedSamplingTest( + () -> coordinatedSamplingWithCondition(cond1, cond2, cond3), numSamples); + assertSnapshots(listener, expectedSnapshots, probeIds); + } + + private static Stream coordinatedSamplingConditionSource() { + return Stream.of( + arguments(FALSE, FALSE, FALSE, 1, 0, new ProbeId[] {}), + arguments(TRUE, FALSE, FALSE, 1, 1, new ProbeId[] {PROBE_ID1}), + arguments(TRUE, TRUE, FALSE, 1, 2, new ProbeId[] {PROBE_ID2, PROBE_ID1}), + arguments(TRUE, TRUE, TRUE, 1, 3, new ProbeId[] {PROBE_ID3, PROBE_ID2, PROBE_ID1}), + arguments(FALSE, FALSE, TRUE, 1, 1, new ProbeId[] {PROBE_ID3}), + arguments(FALSE, TRUE, TRUE, 1, 2, new ProbeId[] {PROBE_ID3, PROBE_ID2}), + arguments(FALSE, TRUE, FALSE, 1, 1, new ProbeId[] {PROBE_ID2}), + arguments(TRUE, TRUE, TRUE, 0, 0, new ProbeId[] {})); + } + + @Test + public void coordinatedSamplingLoopFirstEmit() throws IOException, URISyntaxException { + TestSnapshotListener listener = doCoordinatedSamplingTest(this::coordinatedSamplingLoop, 10); + assertSnapshots(listener, 2, PROBE_ID3, PROBE_ID1); + } + + @Test + public void coordinatedSamplingSiblingFirstEmit() throws IOException, URISyntaxException { + TestSnapshotListener listener = doCoordinatedSamplingTest(this::coordinatedSamplingSibling, 10); + assertSnapshots(listener, 3, PROBE_ID1, PROBE_ID2, PROBE_ID3); + } + + @Test + public void coordinatedSamplingLineFirstEmit() throws IOException, URISyntaxException { + TestSnapshotListener listener = doCoordinatedSamplingTest(this::lineCoordinatedSampling, 10); + assertSnapshots(listener, 3, LINE_PROBE_ID1, LINE_PROBE_ID2, LINE_PROBE_ID3); + } + + @Test + public void coordinatedSamplingLineFirstDrop() throws IOException, URISyntaxException { + TestSnapshotListener listener = doCoordinatedSamplingTest(this::lineCoordinatedSampling, 0); + assertSnapshots(listener, 0); + } + + @ParameterizedTest + @MethodSource("coordinatedSamplingLineConditionSource") + public void coordinatedSamplingLineCondition( + BooleanExpression cond1, + BooleanExpression cond2, + BooleanExpression cond3, + int numSamples, + int expectedSnapshots, + ProbeId... probeIds) + throws IOException, URISyntaxException { + TestSnapshotListener listener = + doCoordinatedSamplingTest( + () -> coordinatedSamplingLineWithCondition(cond1, cond2, cond3), numSamples); + assertSnapshots(listener, expectedSnapshots, probeIds); + } + + private static Stream coordinatedSamplingLineConditionSource() { + return Stream.of( + arguments(FALSE, FALSE, FALSE, 1, 0, new ProbeId[] {}), + arguments(TRUE, FALSE, FALSE, 1, 1, new ProbeId[] {LINE_PROBE_ID1}), + arguments(TRUE, TRUE, FALSE, 1, 2, new ProbeId[] {LINE_PROBE_ID1, LINE_PROBE_ID2}), + arguments( + TRUE, TRUE, TRUE, 1, 3, new ProbeId[] {LINE_PROBE_ID1, LINE_PROBE_ID2, LINE_PROBE_ID3}), + arguments(FALSE, FALSE, TRUE, 1, 1, new ProbeId[] {LINE_PROBE_ID3}), + arguments(FALSE, TRUE, TRUE, 1, 2, new ProbeId[] {LINE_PROBE_ID2, LINE_PROBE_ID3}), + arguments(FALSE, TRUE, FALSE, 1, 1, new ProbeId[] {LINE_PROBE_ID2}), + arguments(TRUE, TRUE, TRUE, 0, 0, new ProbeId[] {})); + } + + @Test + public void noCoordinatedSamplingLogTemplate() throws IOException, URISyntaxException { + TestSnapshotListener listener = + doCoordinatedSamplingTest(this::coordinatedSamplingLogTemplate, 1); + // only one snapshot, the other probes do not benefit from coordinated sampling + assertSnapshots(listener, 1); + } + + @Test + public void coordinatedSamplingIsScopedToLocalRootSpan() throws IOException, URISyntaxException { + MockSampler probeSampler = new MockSampler(1); + ProbeRateLimiter.setSamplerSupplier(rate -> probeSampler); + try { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String className = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe probe1 = createMethodProbeAtExit(PROBE_ID1, className, "process1", null); + LogProbe probe2 = createMethodProbeAtExit(PROBE_ID2, className, "process2", null); + LogProbe probe3 = createMethodProbeAtExit(PROBE_ID3, className, "process3", null); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(className); + + Reflect.onClass(testClass).call("main", "1").get(); + Reflect.onClass(testClass).call("main", "1").get(); + + assertSnapshots(listener, 3, PROBE_ID3, PROBE_ID2, PROBE_ID1); + assertEquals(2, probeSampler.getCallCount()); + assertSame(Context.root(), Context.current()); + } finally { + ProbeRateLimiter.setSamplerSupplier(null); + } + } + + @Test + public void logProbesIgnoreCoordinatedSampling() throws IOException, URISyntaxException { + MockSampler probeSampler = new MockSampler(2); + ProbeRateLimiter.setSamplerSupplier(rate -> probeSampler); + try { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String className = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe logProbe2 = + createProbeBuilder(PROBE_ID2) + .where(className, "process2") + .template("arg={arg}", parseTemplate("arg={arg}")) + .captureSnapshot(false) + .build(); + LogProbe logProbe3 = + createProbeBuilder(PROBE_ID3) + .where(className, "process3") + .template("arg={arg}", parseTemplate("arg={arg}")) + .captureSnapshot(false) + .build(); + LogProbe snapshotProbe = createMethodProbeAtExit(PROBE_ID1, className, "process1", null); + TestSnapshotListener listener = installProbes(snapshotProbe, logProbe2, logProbe3); + Class testClass = compileAndLoadClass(className); + + Reflect.onClass(testClass).call("main", "1").get(); + + assertEquals(2, listener.snapshots.size()); + assertTrue( + listener.snapshots.stream() + .anyMatch(snapshot -> snapshot.getProbe().getId().equals(PROBE_ID1.getId()))); + assertTrue( + listener.snapshots.stream() + .anyMatch(snapshot -> !snapshot.getProbe().getId().equals(PROBE_ID1.getId()))); + assertEquals(3, probeSampler.getCallCount()); + } finally { + ProbeRateLimiter.setSamplerSupplier(null); + } + } + + private TestSnapshotListener coordinatedSampling() throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe probe1 = createMethodProbeAtExit(PROBE_ID1, CLASS_NAME, "process1", null); + LogProbe probe2 = createMethodProbeAtExit(PROBE_ID2, CLASS_NAME, "process2", null); + LogProbe probe3 = createMethodProbeAtExit(PROBE_ID3, CLASS_NAME, "process3", null); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "1").get(); + return listener; + } + + private TestSnapshotListener lineCoordinatedSampling() throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + int line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID1); + LogProbe probe1 = createLineProbe(LINE_PROBE_ID1, CLASS_NAME, line); + line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID2); + LogProbe probe2 = createLineProbe(LINE_PROBE_ID2, CLASS_NAME, line); + line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID3); + LogProbe probe3 = createLineProbe(LINE_PROBE_ID3, CLASS_NAME, line); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "1").get(); + return listener; + } + + private TestSnapshotListener coordinatedSamplingWithCondition( + BooleanExpression cond1, BooleanExpression cond2, BooleanExpression cond3) + throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe probe1 = + createProbeBuilder(PROBE_ID1, CLASS_NAME, "process1", null) + .when(new ProbeCondition(DSL.when(cond1), "")) + .build(); + LogProbe probe2 = + createProbeBuilder(PROBE_ID2, CLASS_NAME, "process2", null) + .when(new ProbeCondition(DSL.when(cond2), "")) + .build(); + LogProbe probe3 = + createProbeBuilder(PROBE_ID3, CLASS_NAME, "process3", null) + .when(new ProbeCondition(DSL.when(cond3), "")) + .build(); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "1").get(); + return listener; + } + + private TestSnapshotListener coordinatedSamplingLineWithCondition( + BooleanExpression cond1, BooleanExpression cond2, BooleanExpression cond3) + throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + int line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID1); + LogProbe probe1 = + createProbeBuilder(LINE_PROBE_ID1, CLASS_NAME, line) + .when(new ProbeCondition(DSL.when(cond1), "")) + .build(); + line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID2); + LogProbe probe2 = + createProbeBuilder(LINE_PROBE_ID2, CLASS_NAME, line) + .when(new ProbeCondition(DSL.when(cond2), "")) + .build(); + line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID3); + LogProbe probe3 = + createProbeBuilder(LINE_PROBE_ID3, CLASS_NAME, line) + .when(new ProbeCondition(DSL.when(cond3), "")) + .build(); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "1").get(); + return listener; + } + + private TestSnapshotListener coordinatedSamplingLoop() throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe probeRoot = createMethodProbeAtExit(PROBE_ID1, CLASS_NAME, "rootLoopProcess", null); + LogProbe probe3 = createMethodProbeAtExit(PROBE_ID3, CLASS_NAME, "process3", null); + TestSnapshotListener listener = installProbes(probeRoot, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "loop").get(); + return listener; + } + + private TestSnapshotListener coordinatedSamplingSibling() throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe probe1 = createMethodProbeAtExit(PROBE_ID1, CLASS_NAME, "siblingProcess1", null); + LogProbe probe2 = createMethodProbeAtExit(PROBE_ID2, CLASS_NAME, "siblingProcess2", null); + LogProbe probe3 = createMethodProbeAtExit(PROBE_ID3, CLASS_NAME, "siblingProcess3", null); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "sibling").get(); + return listener; + } + + private TestSnapshotListener coordinatedSamplingLogTemplate() + throws IOException, URISyntaxException { + CoreTracer tracer = CoreTracer.builder().build(); + TracerInstaller.forceInstallGlobalTracer(tracer); + final String CLASS_NAME = "com.datadog.debugger.CapturedSnapshot21"; + LogProbe probe1 = + createProbeBuilder(PROBE_ID1) + .where(CLASS_NAME, "process1") + .template("arg={arg}", parseTemplate("arg={arg}")) + .captureSnapshot(false) + .build(); + LogProbe probe2 = + createProbeBuilder(PROBE_ID2) + .where(CLASS_NAME, "process2") + .template("arg={arg}", parseTemplate("arg={arg}")) + .captureSnapshot(false) + .build(); + LogProbe probe3 = + createProbeBuilder(PROBE_ID3) + .where(CLASS_NAME, "process3") + .template("arg={arg}", parseTemplate("arg={arg}")) + .captureSnapshot(false) + .build(); + TestSnapshotListener listener = installProbes(probe1, probe2, probe3); + Class testClass = compileAndLoadClass(CLASS_NAME); + Reflect.onClass(testClass).call("main", "1").get(); + return listener; + } + + private TestSnapshotListener doCoordinatedSamplingTest(TestListenerMethod testRun, int numSamples) + throws IOException, URISyntaxException { + MockSampler probeSampler = new MockSampler(numSamples); + ProbeRateLimiter.setSamplerSupplier(rate -> probeSampler); + try { + return testRun.run(); + } finally { + ProbeRateLimiter.setSamplerSupplier(null); + } + } +} diff --git a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/MockSampler.java b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/MockSampler.java index 75e7190bdf7..0644d0ab2a0 100644 --- a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/MockSampler.java +++ b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/MockSampler.java @@ -3,13 +3,22 @@ import datadog.trace.api.sampling.Sampler; public class MockSampler implements Sampler { + private final int numSamples; private int callCount; + public MockSampler() { + this(Integer.MAX_VALUE); + } + + public MockSampler(int numSamples) { + this.numSamples = numSamples; + } + @Override public boolean sample() { callCount++; - return true; + return callCount <= numSamples; } @Override diff --git a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/SpanDecorationProbeInstrumentationTest.java b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/SpanDecorationProbeInstrumentationTest.java index 5bb18c36d7e..5ff94e443b4 100644 --- a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/SpanDecorationProbeInstrumentationTest.java +++ b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/agent/SpanDecorationProbeInstrumentationTest.java @@ -65,6 +65,8 @@ public class SpanDecorationProbeInstrumentationTest extends ProbeInstrumentation private static final ProbeId PROBE_ID4 = new ProbeId("beae1807-f3b0-4ea8-a74f-826790c5e6f9", 0); private static final ProbeId LINE_PROBE_ID1 = new ProbeId("beae1817-f3b0-4ea8-a74f-000000000001", 0); + private static final ProbeId LINE_PROBE_ID3 = + new ProbeId("beae1817-f3b0-4ea8-a74f-000000000003", 0); private TestTraceInterceptor traceInterceptor = new TestTraceInterceptor(); @@ -288,9 +290,9 @@ public void lineRootSpanTagList() throws IOException, URISyntaxException { SpanDecorationProbe.Decoration deco1 = createDecoration("tag1", "{arg}"); SpanDecorationProbe.Decoration deco2 = createDecoration("tag2", "{this.intField}"); SpanDecorationProbe.Decoration deco3 = createDecoration("tag3", "{strField}"); - int line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID1); + int line = getLineForLineProbe(CLASS_NAME, LINE_PROBE_ID3); installSingleSpanDecoration( - LINE_PROBE_ID1, + LINE_PROBE_ID3, CLASS_NAME, ROOT, asList(deco1, deco2, deco3), diff --git a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/probe/LogProbeTest.java b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/probe/LogProbeTest.java index 1928cc52829..225f3e2f91f 100644 --- a/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/probe/LogProbeTest.java +++ b/dd-java-agent/agent-debugger/src/test/java/com/datadog/debugger/probe/LogProbeTest.java @@ -44,6 +44,7 @@ import datadog.trace.bootstrap.instrumentation.api.AgentTracer.TracerAPI; import datadog.trace.bootstrap.instrumentation.api.Tags; import datadog.trace.core.CoreTracer; +import java.util.stream.IntStream; import java.util.stream.Stream; import javax.annotation.Nonnull; import org.junit.jupiter.api.Assertions; @@ -75,6 +76,17 @@ public void testSampling() { assertEquals(0.25, snapshotProbe.getSampling().getEventsPerSecond(), 0.01); } + @Test + public void coordinatedSamplingEmitsProbeOnceConcurrently() { + LogProbe.CoordinatedSamplingState state = + new LogProbe.CoordinatedSamplingState(LogProbe.CoordinatedSamplingState.Status.EMIT); + + long emitted = + IntStream.range(0, 1_000).parallel().filter(i -> state.tryEmit("probe-id")).count(); + + assertEquals(1, emitted); + } + @Test public void debugSessionActive() { assertTrue( @@ -160,8 +172,8 @@ private int runTrace(TracerAPI tracer, boolean captureSnapshot, Integer line, St LogProbe logProbe = builder.build(); logProbe.initSamplers(); - CapturedContext entryContext = capturedContext(span, logProbe); - CapturedContext exitContext = capturedContext(span, logProbe); + CapturedContext entryContext = capturedContext(span, logProbe, MethodLocation.ENTRY); + CapturedContext exitContext = capturedContext(span, logProbe, MethodLocation.EXIT); logProbe.evaluate(entryContext, new LogStatus(logProbe), MethodLocation.ENTRY, false); logProbe.evaluate(exitContext, new LogStatus(logProbe), MethodLocation.EXIT, false); @@ -206,8 +218,8 @@ private boolean fillSnapshot(DebugSessionStatus status) { LogProbe logProbe = builder.build(); - CapturedContext entryContext = capturedContext(span, logProbe); - CapturedContext exitContext = capturedContext(span, logProbe); + CapturedContext entryContext = capturedContext(span, logProbe, MethodLocation.ENTRY); + CapturedContext exitContext = capturedContext(span, logProbe, MethodLocation.EXIT); logProbe.evaluate(entryContext, new LogStatus(logProbe), MethodLocation.ENTRY, false); logProbe.evaluate(exitContext, new LogStatus(logProbe), MethodLocation.EXIT, false); @@ -216,14 +228,11 @@ private boolean fillSnapshot(DebugSessionStatus status) { } } - private static CapturedContext capturedContext(AgentSpan span, ProbeDefinition probeDefinition) { + private static CapturedContext capturedContext( + AgentSpan span, ProbeDefinition probeDefinition, MethodLocation methodLocation) { CapturedContext context = new CapturedContext(); context.evaluate( - probeDefinition, - "Log Probe test", - System.currentTimeMillis(), - MethodLocation.DEFAULT, - false); + probeDefinition, "Log Probe test", System.currentTimeMillis(), methodLocation, false); return context; } @@ -380,8 +389,8 @@ public void captureExpressionsInActiveDebugSession() { "greeting", new ValueScript(DSL.value("hello"), "'hello'"), null))) .build(); logProbe.initSamplers(); - CapturedContext entryContext = capturedContext(span, logProbe); - CapturedContext exitContext = capturedContext(span, logProbe); + CapturedContext entryContext = capturedContext(span, logProbe, MethodLocation.ENTRY); + CapturedContext exitContext = capturedContext(span, logProbe, MethodLocation.EXIT); logProbe.evaluate(entryContext, new LogStatus(logProbe), MethodLocation.ENTRY, false); logProbe.evaluate(exitContext, new LogStatus(logProbe), MethodLocation.EXIT, false); Snapshot snapshot = new Snapshot(currentThread(), logProbe, 3); diff --git a/dd-java-agent/agent-debugger/src/test/resources/com/datadog/debugger/CapturedSnapshot21.java b/dd-java-agent/agent-debugger/src/test/resources/com/datadog/debugger/CapturedSnapshot21.java index 3f79b2e81f5..18c7c162fc9 100644 --- a/dd-java-agent/agent-debugger/src/test/resources/com/datadog/debugger/CapturedSnapshot21.java +++ b/dd-java-agent/agent-debugger/src/test/resources/com/datadog/debugger/CapturedSnapshot21.java @@ -26,12 +26,74 @@ public static int main(String arg) { AgentTracer.TracerAPI tracerAPI = AgentTracer.get(); AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "rootProcess").start(); try (ContextScope scope = tracerAPI.activateManualSpan(span)) { + if (arg.equals("sibling")) { + return new CapturedSnapshot21().rootSiblingProcess(arg); + } + if (arg.equals("loop")) { + return new CapturedSnapshot21().rootLoopProcess(arg); + } return new CapturedSnapshot21().rootProcess(arg); } finally { span.finish(); } } + private int rootSiblingProcess(String arg) { + int result = 0; + AgentTracer.TracerAPI tracerAPI = AgentTracer.get(); + { + AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process1").start(); + try (ContextScope scope = tracerAPI.activateManualSpan(span)) { + result += siblingProcess1(arg); + } finally { + span.finish(); + } + } + { + AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process2").start(); + try (ContextScope scope = tracerAPI.activateManualSpan(span)) { + result += siblingProcess2(arg); + } finally { + span.finish(); + } + } + { + AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process3").start(); + try (ContextScope scope = tracerAPI.activateManualSpan(span)) { + result += siblingProcess3(arg); + } finally { + span.finish(); + } + } + return result; + } + + private int siblingProcess1(String arg) { + return intField; + } + + private int siblingProcess2(String arg) { + return intField; + } + + private int siblingProcess3(String arg) { + return intField; + } + + private int rootLoopProcess(String arg) { + int result = 0; + AgentTracer.TracerAPI tracerAPI = AgentTracer.get(); + for (int i = 0; i < 10; i++) { + AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process3").start(); + try (ContextScope scope = tracerAPI.activateManualSpan(span)) { + result += process3(arg); + } finally { + span.finish(); + } + } + return result; + } + private int rootProcess(String arg) { AgentTracer.TracerAPI tracerAPI = AgentTracer.get(); AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process1").start(); @@ -46,7 +108,7 @@ private int process1(String arg) { AgentTracer.TracerAPI tracerAPI = AgentTracer.get(); AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process2").start(); try (ContextScope scope = tracerAPI.activateManualSpan(span)) { - return process2(arg) + 1; + return process2(arg) + 1; // beae1817-f3b0-4ea8-a74f-000000000001 } finally { span.finish(); } @@ -56,13 +118,13 @@ private int process2(String arg) { AgentTracer.TracerAPI tracerAPI = AgentTracer.get(); AgentSpan span = tracerAPI.buildSpan("dynamic-instrumentation", "process3").start(); try (ContextScope scope = tracerAPI.activateManualSpan(span)) { - return process3(arg) + 1; + return process3(arg) + 1; // beae1817-f3b0-4ea8-a74f-000000000002 } finally { span.finish(); } } private int process3(String arg) { - return intField; // beae1817-f3b0-4ea8-a74f-000000000001 + return intField; // beae1817-f3b0-4ea8-a74f-000000000003 } }