apply rate limit to queue events

Author: richardstartin

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

@@ -1,13 +1,24 @@
 package datadog.trace.bootstrap.instrumentation.java.concurrent;
 
+import datadog.trace.api.config.ProfilingConfig;
 import datadog.trace.api.profiling.QueueTiming;
 import datadog.trace.api.profiling.Timer;
+import datadog.trace.api.sampling.PerRecordingRateLimiter;
 import datadog.trace.bootstrap.ContextStore;
+import datadog.trace.bootstrap.config.provider.ConfigProvider;
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
 import datadog.trace.bootstrap.instrumentation.jfr.InstrumentationBasedProfiling;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
 public class QueueTimerHelper {
 
+  private static final PerRecordingRateLimiter RATE_LIMITER = new PerRecordingRateLimiter(Duration.of(500, ChronoUnit.MILLIS),
+      10_000, // hard limit on queue events
+      ConfigProvider.getInstance().getInteger(ProfilingConfig.PROFILING_UPLOAD_PERIOD,
+          ProfilingConfig.PROFILING_UPLOAD_PERIOD_DEFAULT));
+
   public static <T> void startQueuingTimer(
       ContextStore<T, State> taskContextStore, Class<?> schedulerClass, T task) {
     State state = taskContextStore.get(task);
@@ -17,7 +28,7 @@ public static <T> void startQueuingTimer(
   public static void startQueuingTimer(State state, Class<?> schedulerClass, Object task) {
     // avoid calling this before JFR is initialised because it will lead to reading the wrong
     // TSC frequency before JFR has set it up properly
-    if (task != null && state != null && InstrumentationBasedProfiling.isJFRReady()) {
+    if (task != null && state != null && InstrumentationBasedProfiling.isJFRReady() && RATE_LIMITER.permit()) {
       QueueTiming timing =
           (QueueTiming) AgentTracer.get().getProfilingContext().start(Timer.TimerType.QUEUEING);
       timing.setTask(task);

dd-java-agent/agent-bootstrap/src/main/java11/datadog/trace/bootstrap/instrumentation/jfr/WindowSampler.java

@@ -2,7 +2,6 @@
 
 import datadog.trace.api.sampling.AdaptiveSampler;
 import java.time.Duration;
-import java.time.temporal.ChronoUnit;
 import jdk.jfr.Event;
 import jdk.jfr.EventType;
 
@@ -24,18 +23,4 @@ public void start() {
   public boolean sample() {
     return sampleType.isEnabled() && sampler.sample();
   }
-
-  protected static int samplingWindowsPerRecording(
-      long uploadPeriodSeconds, Duration samplingWindow) {
-    /*
-     * Java8 doesn't have dividedBy#Duration so we have to implement poor man's version.
-     * None of these durations should be big enough to warrant dealing with bigints.
-     * We also do not care about nanoseconds here.
-     */
-    return (int)
-        Math.min(
-            Duration.of(uploadPeriodSeconds, ChronoUnit.SECONDS).toMillis()
-                / samplingWindow.toMillis(),
-            Integer.MAX_VALUE);
-  }
 }

dd-java-agent/agent-bootstrap/src/main/java11/datadog/trace/bootstrap/instrumentation/jfr/backpressure/BackpressureSampler.java

@@ -5,6 +5,8 @@
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 
+import static datadog.trace.api.sampling.PerRecordingRateLimiter.samplingWindowsPerRecording;
+
 final class BackpressureSampler extends WindowSampler<BackpressureSampleEvent> {
   /*
    * Fixed 0.5 second sampling window.

dd-java-agent/agent-bootstrap/src/main/java11/datadog/trace/bootstrap/instrumentation/jfr/directallocation/DirectAllocationSampler.java

@@ -5,6 +5,8 @@
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 
+import static datadog.trace.api.sampling.PerRecordingRateLimiter.samplingWindowsPerRecording;
+
 public class DirectAllocationSampler extends WindowSampler<DirectAllocationSampleEvent> {
 
   /*

dd-java-agent/agent-bootstrap/src/main/java11/datadog/trace/bootstrap/instrumentation/jfr/exceptions/ExceptionSampler.java

@@ -5,6 +5,8 @@
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 
+import static datadog.trace.api.sampling.PerRecordingRateLimiter.samplingWindowsPerRecording;
+
 final class ExceptionSampler extends WindowSampler<ExceptionSampleEvent> {
   /*
    * Fixed 0.5 second sampling window.

internal-api/src/main/java/datadog/trace/api/sampling/PerRecordingRateLimiter.java

@@ -0,0 +1,38 @@
+package datadog.trace.api.sampling;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
+public class PerRecordingRateLimiter {
+
+  private final AdaptiveSampler sampler;
+
+  public PerRecordingRateLimiter(
+      Duration windowDuration, int limit, int recordingLength) {
+    int lookback = samplingWindowsPerRecording(recordingLength, windowDuration);
+    int samplesPerWindow = limit / samplingWindowsPerRecording(recordingLength, windowDuration);
+    sampler = new AdaptiveSampler(windowDuration, samplesPerWindow, lookback, 16, false);
+  }
+
+  public void start() {
+    sampler.start();
+  }
+
+  public boolean permit() {
+    return sampler.sample();
+  }
+
+  public static int samplingWindowsPerRecording(
+      long uploadPeriodSeconds, Duration samplingWindow) {
+    /*
+     * Java8 doesn't have dividedBy#Duration so we have to implement poor man's version.
+     * None of these durations should be big enough to warrant dealing with bigints.
+     * We also do not care about nanoseconds here.
+     */
+    return (int)
+        Math.min(
+            Duration.of(uploadPeriodSeconds, ChronoUnit.SECONDS).toMillis()
+                / samplingWindow.toMillis(),
+            Integer.MAX_VALUE);
+  }
+}