From e28cb054f903b39c36c883efb317bb07f1d35783 Mon Sep 17 00:00:00 2001
From: Oleg Pudeyev <156273877+p-datadog@users.noreply.github.com>
Date: Tue, 12 Nov 2024 10:36:42 -0500
Subject: [PATCH] DEBUG-2334 DI internal settings reorg + telemetry + error
 handling (#4089)

- Moves internal DI settings to "internal" subkey
- Minimum send interval for the network submission worker moved to a new internal configuration setting from the constant
- Injects telemetry object into DI components
- Reports unhandled exceptions to telemetry in DI code already in master
- Line probes now require a single match of the provided path suffix, if multiple paths match an exception is raised (this will be reported via UI to the user, on the assumption that user wanted a specific file instrumented and we don't know which of the multiple matching files is the correct one)
- When probe installation fails for reasons other than the referenced target being missing (presumably not yet loaded), the failure is noted and that probe is not attempted again in the future (this PR contains part of the code for this change)
- Improves probe notifier worker flush logic to be event-driven instead of sleeping
---
 benchmarks/di_instrument.rb                   |   6 +-
 lib/datadog/di/code_tracker.rb                |  30 ++++-
 lib/datadog/di/configuration/settings.rb      | 113 +++++++++++-------
 lib/datadog/di/error.rb                       |  17 +++
 lib/datadog/di/instrumenter.rb                |  38 ++++--
 lib/datadog/di/probe.rb                       |  12 ++
 lib/datadog/di/probe_notifier_worker.rb       | 109 ++++++++++-------
 sig/datadog/di/code_tracker.rbs               |   2 +-
 sig/datadog/di/error.rbs                      |   4 +
 sig/datadog/di/instrumenter.rbs               |   6 +-
 sig/datadog/di/probe.rbs                      |   1 +
 sig/datadog/di/probe_notifier_worker.rbs      |   8 +-
 spec/datadog/di/code_tracker_spec.rb          |  14 ++-
 .../datadog/di/configuration/settings_spec.rb |  46 ++++---
 spec/datadog/di/instrumenter_spec.rb          |  30 ++---
 spec/datadog/di/probe_notifier_worker_spec.rb |  32 ++---
 spec/datadog/di/spec_helper.rb                |  25 ++++
 17 files changed, 325 insertions(+), 168 deletions(-)

diff --git a/benchmarks/di_instrument.rb b/benchmarks/di_instrument.rb
index d5a027e9131..935a90b0e20 100644
--- a/benchmarks/di_instrument.rb
+++ b/benchmarks/di_instrument.rb
@@ -73,7 +73,11 @@ def run_benchmark
     settings = Datadog.configuration
     # We benchmark untargeted and targeted trace points; untargeted ones
     # are prohibited by default, permit them.
-    settings.dynamic_instrumentation.untargeted_trace_points = true
+    begin
+      settings.dynamic_instrumentation.internal.untargeted_trace_points = true
+    rescue NoMethodError
+      settings.dynamic_instrumentation.untargeted_trace_points = true
+    end
     redactor = Datadog::DI::Redactor.new(settings)
     serializer = Datadog::DI::Serializer.new(settings, redactor)
     logger = Logger.new(STDERR)
diff --git a/lib/datadog/di/code_tracker.rb b/lib/datadog/di/code_tracker.rb
index 33aba799852..dbf3d394c45 100644
--- a/lib/datadog/di/code_tracker.rb
+++ b/lib/datadog/di/code_tracker.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+# rubocop:disable Lint/AssignmentInCondition
+
 module Datadog
   module DI
     # Tracks loaded Ruby code by source file and maintains a map from
@@ -77,6 +79,23 @@ def start
                 registry[path] = tp.instruction_sequence
               end
             end
+          # Since this method normally is called from customer applications,
+          # rescue any exceptions that might not be handled to not break said
+          # customer applications.
+          rescue => exc
+            # TODO we do not have DI.component defined yet, remove steep:ignore
+            # before release.
+            if component = DI.component # steep:ignore
+              raise if component.settings.dynamic_instrumentation.internal.propagate_all_exceptions
+              component.logger.warn("Unhandled exception in script_compiled trace point: #{exc.class}: #{exc}")
+              component.telemetry&.report(exc, description: "Unhandled exception in script_compiled trace point")
+              # TODO test this path
+            else
+              # If we don't have a component, we cannot log anything properly.
+              # Do not just print a warning to avoid spamming customer logs.
+              # Don't reraise the exception either.
+              # TODO test this path
+            end
           end
         end
       end
@@ -112,15 +131,18 @@ def active?
       def iseqs_for_path_suffix(suffix)
         registry_lock.synchronize do
           exact = registry[suffix]
-          return [exact] if exact
+          return [suffix, exact] if exact
 
           inexact = []
           registry.each do |path, iseq|
             if Utils.path_matches_suffix?(path, suffix)
-              inexact << iseq
+              inexact << [path, iseq]
             end
           end
-          inexact
+          if inexact.length > 1
+            raise Error::MultiplePathsMatch, "Multiple paths matched requested suffix"
+          end
+          inexact.first
         end
       end
 
@@ -164,3 +186,5 @@ def clear
     end
   end
 end
+
+# rubocop:enable Lint/AssignmentInCondition
diff --git a/lib/datadog/di/configuration/settings.rb b/lib/datadog/di/configuration/settings.rb
index 5bf6f83d441..55413613494 100644
--- a/lib/datadog/di/configuration/settings.rb
+++ b/lib/datadog/di/configuration/settings.rb
@@ -12,8 +12,6 @@ def self.extended(base)
 
         def self.add_settings!(base)
           base.class_eval do
-            # The setting has "internal" prefix to prevent it from being
-            # prematurely turned on by customers.
             settings :dynamic_instrumentation do
               option :enabled do |o|
                 o.type :bool
@@ -26,48 +24,6 @@ def self.add_settings!(base)
                 o.default false
               end
 
-              # This option instructs dynamic instrumentation to use
-              # untargeted trace points when installing line probes and
-              # code tracking is not active.
-              # WARNING: untargeted trace points carry a massive performance
-              # penalty for the entire file in which a line probe is placed.
-              #
-              # If this option is set to false, which is the default,
-              # dynamic instrumentation will add probes that reference
-              # unknown files to the list of pending probes, and when
-              # the respective files are loaded, the line probes will be
-              # installed using targeted trace points. If the file in
-              # question is already loaded when the probe is received
-              # (for example, it is in a third-party library loaded during
-              # application boot), and code tracking was not active when
-              # the file was loaded, such files will not be instrumentable
-              # via line probes.
-              #
-              # If this option is set to true
-              #
-              # activated, DI will in
-              # activated or because the files being targeted have beenIf true and code tracking is not enabled, dynamic instrumentation
-              # will use untargeted trace points.
-              # If false and code tracking is not enabled, dynamic
-              # instrumentation will not instrument any files loaded
-              # WARNING: these trace points will greatly degrade performance
-              # of all code in the instrumented files.
-              option :untargeted_trace_points do |o|
-                o.type :bool
-                o.default false
-              end
-
-              # If true, all of the catch-all rescue blocks in DI
-              # will propagate the exceptions onward.
-              # WARNING: for internal Datadog use only - this will break
-              # the DI product and potentially the library in general in
-              # a multitude of ways, cause resource leakage, permanent
-              # performance decreases, etc.
-              option :propagate_all_exceptions do |o|
-                o.type :bool
-                o.default false
-              end
-
               # An array of variable and key names to redact in addition to
               # the built-in list of identifiers.
               #
@@ -154,6 +110,75 @@ def self.add_settings!(base)
                 o.type :int
                 o.default 20
               end
+
+              # Settings in the 'internal' group are for internal Datadog
+              # use only, and are needed to test dynamic instrumentation or
+              # experiment with features not released to customers.
+              settings :internal do
+                # This option instructs dynamic instrumentation to use
+                # untargeted trace points when installing line probes and
+                # code tracking is not active.
+                # WARNING: untargeted trace points carry a massive performance
+                # penalty for the entire file in which a line probe is placed.
+                #
+                # If this option is set to false, which is the default,
+                # dynamic instrumentation will add probes that reference
+                # unknown files to the list of pending probes, and when
+                # the respective files are loaded, the line probes will be
+                # installed using targeted trace points. If the file in
+                # question is already loaded when the probe is received
+                # (for example, it is in a third-party library loaded during
+                # application boot), and code tracking was not active when
+                # the file was loaded, such files will not be instrumentable
+                # via line probes.
+                #
+                # If this option is set to true, dynamic instrumentation will
+                # install untargeted trace points for all line probes,
+                # regardless of whether the referenced file is loaded.
+                # This permits instrumenting code which was loaded prior to
+                # code tracking being activated and instrumenting lines when
+                # code tracking is not activated at all. However, untargeted
+                # trace points are extremely slow and will greatly degrade
+                # performance of *all* code executed while they are installed,
+                # not just the instrumentation target.
+                option :untargeted_trace_points do |o|
+                  o.type :bool
+                  o.default false
+                end
+
+                # If true, all of the catch-all rescue blocks in DI
+                # will propagate the exceptions onward.
+                # WARNING: for internal Datadog use only - this will break
+                # the DI product and potentially the library in general in
+                # a multitude of ways, cause resource leakage, permanent
+                # performance decreases, etc.
+                option :propagate_all_exceptions do |o|
+                  o.type :bool
+                  o.default false
+                end
+
+                # Minimum interval, in seconds, between probe status and
+                # snapshot submissions to the agent. Probe notifier worker will
+                # batch together payloads submitted during each interval.
+                # A longer interval reduces the overhead imposed by dynamic
+                # instrumentation on the application, but also increases the
+                # time when application code cannot run (when the batches are
+                # being sent out by the probe notifier worker) and creates a
+                # possibility of dropping payloads if the queue gets too long.
+                option :min_send_interval do |o|
+                  o.type :int
+                  o.default 3
+                end
+
+                # Enable dynamic instrumentation in development environments.
+                # Currently DI does not fully implement support for code
+                # unloading and reloading, and is not supported in
+                # non-production environments.
+                option :development do |o|
+                  o.type :bool
+                  o.default false
+                end
+              end
             end
           end
         end
diff --git a/lib/datadog/di/error.rb b/lib/datadog/di/error.rb
index 5a90d259959..d5f305617cd 100644
--- a/lib/datadog/di/error.rb
+++ b/lib/datadog/di/error.rb
@@ -26,6 +26,23 @@ class AgentCommunicationError < Error
       # that does not in fact exist anywhere (e.g. due to a misspelling).
       class DITargetNotDefined < Error
       end
+
+      # Raised when trying to install a probe whose installation failed
+      # earlier in the same process. This exception should contain the
+      # original exception report from initial installation attempt.
+      class ProbePreviouslyFailed < Error
+      end
+
+      # Raised when installing a line probe and multiple files match the
+      # specified path suffix.
+      # A probe must be installed into one file only, since UI only
+      # supports one instrumented location for a probe.
+      # If multiple files match, UI cannot properly render the data from
+      # all of them, and arbitrarily choosing one file may be not what the
+      # user intended. Instrumentation will fail when multiple files match
+      # and the user will need to make their suffix more precise.
+      class MultiplePathsMatch < Error
+      end
     end
   end
 end
diff --git a/lib/datadog/di/instrumenter.rb b/lib/datadog/di/instrumenter.rb
index 540a6c0db7e..068a5205e9c 100644
--- a/lib/datadog/di/instrumenter.rb
+++ b/lib/datadog/di/instrumenter.rb
@@ -54,10 +54,11 @@ module DI
     #
     # @api private
     class Instrumenter
-      def initialize(settings, serializer, logger, code_tracker: nil)
+      def initialize(settings, serializer, logger, code_tracker: nil, telemetry: nil)
         @settings = settings
         @serializer = serializer
         @logger = logger
+        @telemetry = telemetry
         @code_tracker = code_tracker
 
         @lock = Mutex.new
@@ -66,6 +67,7 @@ def initialize(settings, serializer, logger, code_tracker: nil)
       attr_reader :settings
       attr_reader :serializer
       attr_reader :logger
+      attr_reader :telemetry
       attr_reader :code_tracker
 
       # This is a substitute for Thread::Backtrace::Location
@@ -172,12 +174,12 @@ def hook_line(probe, &block)
         # we use mock objects and the methods may be mocked with
         # individual invocations, yielding different return values on
         # different calls to the same method.
-        permit_untargeted_trace_points = settings.dynamic_instrumentation.untargeted_trace_points
+        permit_untargeted_trace_points = settings.dynamic_instrumentation.internal.untargeted_trace_points
 
         iseq = nil
         if code_tracker
-          iseq = code_tracker.iseqs_for_path_suffix(probe.file).first # steep:ignore
-          unless iseq
+          ret = code_tracker.iseqs_for_path_suffix(probe.file) # steep:ignore
+          unless ret
             if permit_untargeted_trace_points
               # Continue withoout targeting the trace point.
               # This is going to cause a serious performance penalty for
@@ -204,6 +206,10 @@ def hook_line(probe, &block)
           raise Error::DITargetNotDefined, "File not in code tracker registry: #{probe.file}"
         end
 
+        if ret
+          actual_path, iseq = ret
+        end
+
         # If trace point is not targeted, we only need one trace point per file.
         # Creating a trace point for each probe does work but the performance
         # penalty will be taken for each trace point defined in the file.
@@ -217,18 +223,26 @@ def hook_line(probe, &block)
         # this optimization just yet and create a trace point for each probe.
 
         tp = TracePoint.new(:line) do |tp|
-          # If trace point is not targeted, we must verify that the invocation
-          # is the file & line that we want, because untargeted trace points
-          # are invoked for *each* line of Ruby executed.
-          if iseq || tp.lineno == probe.line_no && probe.file_matches?(tp.path)
-            if rate_limiter.nil? || rate_limiter.allow?
-              # & is to stop steep complaints, block is always present here.
-              block&.call(probe: probe, trace_point: tp, caller_locations: caller_locations)
+          begin
+            # If trace point is not targeted, we must verify that the invocation
+            # is the file & line that we want, because untargeted trace points
+            # are invoked for *each* line of Ruby executed.
+            if iseq || tp.lineno == probe.line_no && probe.file_matches?(tp.path)
+              if rate_limiter.nil? || rate_limiter.allow?
+                # & is to stop steep complaints, block is always present here.
+                block&.call(probe: probe, trace_point: tp, caller_locations: caller_locations)
+              end
             end
+          rescue => exc
+            raise if settings.dynamic_instrumentation.internal.propagate_all_exceptions
+            logger.warn("Unhandled exception in line trace point: #{exc.class}: #{exc}")
+            telemetry&.report(exc, description: "Unhandled exception in line trace point")
+            # TODO test this path
           end
         rescue => exc
           raise if settings.dynamic_instrumentation.propagate_all_exceptions
           logger.warn("Unhandled exception in line trace point: #{exc.class}: #{exc}")
+          telemetry&.report(exc, description: "Unhandled exception in line trace point")
           # TODO test this path
         end
 
@@ -244,6 +258,8 @@ def hook_line(probe, &block)
           end
 
           probe.instrumentation_trace_point = tp
+          # actual_path could be nil if we don't use targeted trace points.
+          probe.instrumented_path = actual_path
 
           if iseq
             tp.enable(target: iseq, target_line: line_no)
diff --git a/lib/datadog/di/probe.rb b/lib/datadog/di/probe.rb
index 877c3ade846..5dd5cf88a72 100644
--- a/lib/datadog/di/probe.rb
+++ b/lib/datadog/di/probe.rb
@@ -71,6 +71,8 @@ def initialize(id:, type:,
 
         @rate_limit = rate_limit || (@capture_snapshot ? 1 : 5000)
         @rate_limiter = Datadog::Core::TokenBucket.new(@rate_limit)
+
+        @emitting_notified = false
       end
 
       attr_reader :id
@@ -157,6 +159,16 @@ def file_matches?(path)
       # Line trace point for line probes. Normally this would be a targeted
       # trace point.
       attr_accessor :instrumentation_trace_point
+
+      # Actual path to the file instrumented by the probe, for line probes,
+      # when code tracking is available and line trace point is targeted.
+      # For untargeted line trace points instrumented path will be nil.
+      attr_accessor :instrumented_path
+
+      attr_writer :emitting_notified
+      def emitting_notified?
+        !!@emitting_notified
+      end
     end
   end
 end
diff --git a/lib/datadog/di/probe_notifier_worker.rb b/lib/datadog/di/probe_notifier_worker.rb
index 98e49e5f0f7..6ce2ccdf340 100644
--- a/lib/datadog/di/probe_notifier_worker.rb
+++ b/lib/datadog/di/probe_notifier_worker.rb
@@ -23,12 +23,9 @@ module DI
     #
     # @api private
     class ProbeNotifierWorker
-      # Minimum interval between submissions.
-      # TODO make this into an internal setting and increase default to 2 or 3.
-      MIN_SEND_INTERVAL = 1
-
-      def initialize(settings, transport, logger)
+      def initialize(settings, transport, logger, telemetry: nil)
         @settings = settings
+        @telemetry = telemetry
         @status_queue = []
         @snapshot_queue = []
         @transport = transport
@@ -39,10 +36,12 @@ def initialize(settings, transport, logger)
         @sleep_remaining = nil
         @wake_scheduled = false
         @thread = nil
+        @flush = 0
       end
 
       attr_reader :settings
       attr_reader :logger
+      attr_reader :telemetry
 
       def start
         return if @thread
@@ -53,33 +52,38 @@ def start
             # and then quit?
             break if @stop_requested
 
-            sleep_remaining = @lock.synchronize do
-              if sleep_remaining && sleep_remaining > 0
-                # Recalculate how much sleep time is remaining, then sleep that long.
-                set_sleep_remaining
-              else
-                0
+            # If a flush was requested, send immediately and do not
+            # wait for the cooldown period.
+            if @lock.synchronize { @flush } == 0
+              sleep_remaining = @lock.synchronize do
+                if sleep_remaining && sleep_remaining > 0
+                  # Recalculate how much sleep time is remaining, then sleep that long.
+                  set_sleep_remaining
+                else
+                  0
+                end
               end
-            end
 
-            if sleep_remaining > 0
-              # Do not need to update @wake_scheduled here because
-              # wake-up is already scheduled for the earliest possible time.
-              wake.wait(sleep_remaining)
-              next
+              if sleep_remaining > 0
+                # Do not need to update @wake_scheduled here because
+                # wake-up is already scheduled for the earliest possible time.
+                wake.wait(sleep_remaining)
+                next
+              end
             end
 
             begin
               more = maybe_send
             rescue => exc
-              raise if settings.dynamic_instrumentation.propagate_all_exceptions
+              raise if settings.dynamic_instrumentation.internal.propagate_all_exceptions
 
               logger.warn("Error in probe notifier worker: #{exc.class}: #{exc} (at #{exc.backtrace.first})")
+              telemetry&.report(exc, description: "Error in probe notifier worker")
             end
             @lock.synchronize do
               @wake_scheduled = more
             end
-            wake.wait(more ? MIN_SEND_INTERVAL : nil)
+            wake.wait(more ? min_send_interval : nil)
           end
         end
       end
@@ -106,26 +110,40 @@ def stop(timeout = 1)
       # therefore, it should only be called when there is no parallel
       # activity (in another thread) that causes more notifications
       # to be generated.
+      #
+      # This method is used by the test suite to wait until notifications have
+      # been sent out, and could be used for graceful stopping of the
+      # worker thread.
       def flush
-        loop do
-          if @thread.nil? || !@thread.alive?
-            return
-          end
+        @lock.synchronize do
+          @flush += 1
+        end
+        begin
+          loop do
+            if @thread.nil? || !@thread.alive?
+              return
+            end
 
-          io_in_progress, queues_empty = @lock.synchronize do
-            [io_in_progress?, status_queue.empty? && snapshot_queue.empty?]
-          end
+            io_in_progress, queues_empty = @lock.synchronize do
+              [io_in_progress?, status_queue.empty? && snapshot_queue.empty?]
+            end
 
-          if io_in_progress
-            # If we just call Thread.pass we could be in a busy loop -
-            # add a sleep.
-            sleep 0.25
-            next
-          elsif queues_empty
-            break
-          else
-            sleep 0.25
-            next
+            if io_in_progress
+              # If we just call Thread.pass we could be in a busy loop -
+              # add a sleep.
+              sleep 0.25
+              next
+            elsif queues_empty
+              break
+            else
+              wake.signal
+              sleep 0.25
+              next
+            end
+          end
+        ensure
+          @lock.synchronize do
+            @flush -= 1
           end
         end
       end
@@ -136,6 +154,11 @@ def flush
       attr_reader :wake
       attr_reader :thread
 
+      # Convenience method to keep line length reasonable in the rest of the file.
+      def min_send_interval
+        settings.dynamic_instrumentation.internal.min_send_interval
+      end
+
       # This method should be called while @lock is held.
       def io_in_progress?
         @io_in_progress
@@ -181,14 +204,14 @@ def io_in_progress?
         end
 
         # Determine how much longer the worker thread should sleep
-        # so as not to send in less than MIN_SEND_INTERVAL since the last send.
+        # so as not to send in less than min send interval since the last send.
         # Important: this method must be called when @lock is held.
         #
         # Returns the time remaining to sleep.
         def set_sleep_remaining
           now = Core::Utils::Time.get_time
           @sleep_remaining = if last_sent
-            [last_sent + MIN_SEND_INTERVAL - now, 0].max
+            [last_sent + min_send_interval - now, 0].max
           else
             0
           end
@@ -218,16 +241,20 @@ def set_sleep_remaining
                 @last_sent = time
               end
             rescue => exc
-              raise if settings.dynamic_instrumentation.propagate_all_exceptions
+              raise if settings.dynamic_instrumentation.internal.propagate_all_exceptions
               logger.warn("failed to send #{event_name}: #{exc.class}: #{exc} (at #{exc.backtrace.first})")
+              # Should we report this error to telemetry? Most likely failure
+              # to send is due to a network issue, and trying to send a
+              # telemetry message would also fail.
             end
           end
           batch.any? # steep:ignore
-        rescue ThreadError
+        rescue ThreadError => exc
           # Normally the queue should only be consumed in this method,
           # however if anyone consumes it elsewhere we don't want to block
           # while consuming it here. Rescue ThreadError and return.
-          logger.warn("unexpected #{event_name} queue underflow - consumed elsewhere?")
+          logger.warn("Unexpected #{event_name} queue underflow - consumed elsewhere?")
+          telemetry&.report(exc, description: "Unexpected #{event_name} queue underflow")
         ensure
           @lock.synchronize do
             @io_in_progress = false
diff --git a/sig/datadog/di/code_tracker.rbs b/sig/datadog/di/code_tracker.rbs
index c3aa0fae9e0..f09b14b8086 100644
--- a/sig/datadog/di/code_tracker.rbs
+++ b/sig/datadog/di/code_tracker.rbs
@@ -11,7 +11,7 @@ module Datadog
 
       def start: () -> void
       def active?: () -> bool
-      def iseqs_for_path_suffix: (String suffix) -> (::Array[RubyVM::InstructionSequence])
+      def iseqs_for_path_suffix: (String suffix) -> untyped
       def stop: () -> void
       def clear: () -> void
 
diff --git a/sig/datadog/di/error.rbs b/sig/datadog/di/error.rbs
index 3f3dddc7780..351fa7bb85a 100644
--- a/sig/datadog/di/error.rbs
+++ b/sig/datadog/di/error.rbs
@@ -7,6 +7,10 @@ module Datadog
       end
       class DITargetNotDefined < Error
       end
+      class ProbePreviouslyFailed < Error
+      end
+      class MultiplePathsMatch < Error
+      end
     end
   end
 end
diff --git a/sig/datadog/di/instrumenter.rbs b/sig/datadog/di/instrumenter.rbs
index f11b82d1a82..cfaa56cac64 100644
--- a/sig/datadog/di/instrumenter.rbs
+++ b/sig/datadog/di/instrumenter.rbs
@@ -16,10 +16,12 @@ module Datadog
       @code_tracker: CodeTracker
       
       @logger: Core::Logger
+      
+      @telemetry: Core::Telemetry::Component?
 
       @lock: Mutex
 
-      def initialize: (untyped settings, Serializer serializer, Core::Logger logger, ?code_tracker: CodeTracker?) -> void
+      def initialize: (untyped settings, Serializer serializer, Core::Logger logger, ?code_tracker: CodeTracker?, ?telemetry: Core::Telemetry::Component) -> void
 
       attr_reader settings: untyped
 
@@ -28,6 +30,8 @@ module Datadog
       attr_reader code_tracker: CodeTracker
       
       attr_reader logger: Core::Logger
+      
+      attr_reader telemetry: Core::Telemetry::Component?
 
       def hook_method: (Probe probe) ?{ (?) -> untyped } -> void
 
diff --git a/sig/datadog/di/probe.rbs b/sig/datadog/di/probe.rbs
index dd63ef4bc56..d492b650414 100644
--- a/sig/datadog/di/probe.rbs
+++ b/sig/datadog/di/probe.rbs
@@ -48,6 +48,7 @@ module Datadog
       
       attr_accessor instrumentation_module: Module?
       attr_accessor instrumentation_trace_point: TracePoint?
+      attr_accessor instrumented_path: String?
     end
   end
 end
diff --git a/sig/datadog/di/probe_notifier_worker.rbs b/sig/datadog/di/probe_notifier_worker.rbs
index e490ff036b2..1886aff2735 100644
--- a/sig/datadog/di/probe_notifier_worker.rbs
+++ b/sig/datadog/di/probe_notifier_worker.rbs
@@ -23,12 +23,16 @@ module Datadog
       @stop_requested: bool
       
       @logger: Core::Logger
+      
+      @telemetry: Core::Telemetry::Component?
 
-      def initialize: (untyped settings, Transport transport, Core::Logger logger) -> void
+      def initialize: (untyped settings, Transport transport, Core::Logger logger, ?telemetry: Core::Telemetry::Component) -> void
 
       attr_reader settings: untyped
       
       attr_reader logger: Core::Logger
+      
+      attr_reader telemetry: Core::Telemetry::Component?
 
       def start: () -> void
       def stop: (?::Integer timeout) -> void
@@ -56,6 +60,8 @@ module Datadog
       def maybe_send_status: () -> bool
 
       def maybe_send_snapshot: () -> bool
+      
+      def min_send_interval: () -> Float
     end
   end
 end
diff --git a/spec/datadog/di/code_tracker_spec.rb b/spec/datadog/di/code_tracker_spec.rb
index 1bdc3644cd3..3f4b921e7be 100644
--- a/spec/datadog/di/code_tracker_spec.rb
+++ b/spec/datadog/di/code_tracker_spec.rb
@@ -142,11 +142,11 @@
       end
 
       it "returns the exact match only" do
-        expect(tracker.iseqs_for_path_suffix(path)).to eq([path])
+        expect(tracker.iseqs_for_path_suffix(path)).to eq([path, path])
       end
     end
 
-    context "basename match" do
+    context "basename matches two paths" do
       let(:expected) do
         [
           File.join(File.dirname(__FILE__), "code_tracker_test_class_1.rb"),
@@ -154,14 +154,16 @@
         ]
       end
 
-      it "returns the exact match only" do
-        expect(tracker.iseqs_for_path_suffix("code_tracker_test_class_1.rb")).to eq(expected)
+      it "raises exception" do
+        expect do
+          tracker.iseqs_for_path_suffix("code_tracker_test_class_1.rb")
+        end.to raise_error(Datadog::DI::Error::MultiplePathsMatch)
       end
     end
 
     context "match not on path component boundary" do
-      it "returns no matches" do
-        expect(tracker.iseqs_for_path_suffix("1.rb")).to eq([])
+      it "returns nil" do
+        expect(tracker.iseqs_for_path_suffix("1.rb")).to be nil
       end
     end
   end
diff --git a/spec/datadog/di/configuration/settings_spec.rb b/spec/datadog/di/configuration/settings_spec.rb
index e0f5b75e3fb..54dc019c2f6 100644
--- a/spec/datadog/di/configuration/settings_spec.rb
+++ b/spec/datadog/di/configuration/settings_spec.rb
@@ -6,33 +6,45 @@
   describe "dynamic_instrumentation" do
     context "programmatic configuration" do
       [
-        ["enabled", true],
-        ["enabled", false],
-        ["untargeted_trace_points", true],
-        ["untargeted_trace_points", false],
-        ["propagate_all_exceptions", true],
-        ["propagate_all_exceptions", false],
-        ["redacted_identifiers", ["foo"]],
-        ["redacted_identifiers", []],
-        ["redacted_type_names", ["foo*", "bar"]],
-        ["redacted_type_names", []],
-        ["max_capture_depth", 5],
-        ["max_capture_collection_size", 10],
-        ["max_capture_string_length", 20],
-        ["max_capture_attribute_count", 4],
-      ].each do |(name_, value_)|
+        [nil, "enabled", true],
+        [nil, "enabled", false],
+        ["internal", "untargeted_trace_points", true],
+        ["internal", "untargeted_trace_points", false],
+        ["internal", "propagate_all_exceptions", true],
+        ["internal", "propagate_all_exceptions", false],
+        ['internal', 'min_send_interval', 5],
+        ['internal', 'development', true],
+        ['internal', 'development', false],
+        [nil, "redacted_identifiers", ["foo"]],
+        [nil, "redacted_identifiers", []],
+        [nil, "redacted_type_names", ["foo*", "bar"]],
+        [nil, "redacted_type_names", []],
+        [nil, "max_capture_depth", 5],
+        [nil, "max_capture_collection_size", 10],
+        [nil, "max_capture_string_length", 20],
+        [nil, "max_capture_attribute_count", 4],
+      ].each do |(scope_name_, name_, value_)|
         name = name_
+        scope_name = scope_name_
         value = value_.freeze
 
         context "when #{name} set to #{value}" do
           let(:value) { _value }
 
+          let(:scope) do
+            if scope_name
+              settings.dynamic_instrumentation.public_send(scope_name)
+            else
+              settings.dynamic_instrumentation
+            end
+          end
+
           before do
-            settings.dynamic_instrumentation.public_send("#{name}=", value)
+            scope.public_send("#{name}=", value)
           end
 
           it "returns the value back" do
-            expect(settings.dynamic_instrumentation.public_send(name)).to eq(value)
+            expect(scope.public_send(name)).to eq(value)
           end
         end
       end
diff --git a/spec/datadog/di/instrumenter_spec.rb b/spec/datadog/di/instrumenter_spec.rb
index 0f03dcd7126..f34c397f1e3 100644
--- a/spec/datadog/di/instrumenter_spec.rb
+++ b/spec/datadog/di/instrumenter_spec.rb
@@ -8,20 +8,12 @@
 
   let(:observed_calls) { [] }
 
-  let(:settings) do
-    double('settings').tap do |settings|
-      allow(settings).to receive(:dynamic_instrumentation).and_return(di_settings)
-    end
-  end
-
-  let(:di_settings) do
-    double('di settings').tap do |settings|
-      allow(settings).to receive(:enabled).and_return(true)
-      allow(settings).to receive(:untargeted_trace_points).and_return(false)
-      allow(settings).to receive(:max_capture_depth).and_return(2)
-      allow(settings).to receive(:redacted_type_names).and_return([])
-      allow(settings).to receive(:redacted_identifiers).and_return([])
-    end
+  mock_settings_for_di do |settings|
+    allow(settings.dynamic_instrumentation).to receive(:enabled).and_return(true)
+    allow(settings.dynamic_instrumentation.internal).to receive(:untargeted_trace_points).and_return(false)
+    allow(settings.dynamic_instrumentation).to receive(:max_capture_depth).and_return(2)
+    allow(settings.dynamic_instrumentation).to receive(:redacted_type_names).and_return([])
+    allow(settings.dynamic_instrumentation).to receive(:redacted_identifiers).and_return([])
   end
 
   let(:redactor) do
@@ -321,7 +313,7 @@
         before do
           # We need untargeted trace points for this test since the line
           # being instrumented has already been loaded.
-          expect(di_settings).to receive(:untargeted_trace_points).and_return(true)
+          expect(di_internal_settings).to receive(:untargeted_trace_points).and_return(true)
         end
 
         let(:code_tracker) { nil }
@@ -359,7 +351,7 @@
         let(:code_tracker) { Datadog::DI.code_tracker }
 
         before do
-          expect(di_settings).to receive(:untargeted_trace_points).and_return(false)
+          expect(di_internal_settings).to receive(:untargeted_trace_points).and_return(false)
           Datadog::DI.activate_tracking!
           code_tracker.clear
         end
@@ -389,7 +381,7 @@
       before do
         # We need untargeted trace points for this test since the line
         # being instrumented has already been loaded.
-        expect(di_settings).to receive(:untargeted_trace_points).and_return(true)
+        expect(di_internal_settings).to receive(:untargeted_trace_points).and_return(true)
       end
 
       let(:probe) do
@@ -416,7 +408,7 @@
       before do
         # We need untargeted trace points for this test since the line
         # being instrumented has already been loaded.
-        expect(di_settings).to receive(:untargeted_trace_points).and_return(true)
+        expect(di_internal_settings).to receive(:untargeted_trace_points).and_return(true)
       end
 
       let(:probe) do
@@ -472,7 +464,7 @@
       let(:code_tracker) { nil }
 
       before do
-        expect(di_settings).to receive(:untargeted_trace_points).at_least(:once).and_return(true)
+        expect(di_internal_settings).to receive(:untargeted_trace_points).at_least(:once).and_return(true)
       end
 
       # We do not currently de-duplicate.
diff --git a/spec/datadog/di/probe_notifier_worker_spec.rb b/spec/datadog/di/probe_notifier_worker_spec.rb
index bdf6735eaef..0c595d330cb 100644
--- a/spec/datadog/di/probe_notifier_worker_spec.rb
+++ b/spec/datadog/di/probe_notifier_worker_spec.rb
@@ -4,16 +4,11 @@
 RSpec.describe Datadog::DI::ProbeNotifierWorker do
   di_test
 
-  let(:settings) do
-    double('settings').tap do |settings|
-      allow(settings).to receive(:dynamic_instrumentation).and_return(di_settings)
-    end
-  end
-
-  let(:di_settings) do
-    double('di settings').tap do |settings|
-      allow(settings).to receive(:propagate_all_exceptions).and_return(false)
-    end
+  mock_settings_for_di do |settings|
+    allow(settings.dynamic_instrumentation).to receive(:enabled).and_return(true)
+    allow(settings.dynamic_instrumentation.internal).to receive(:propagate_all_exceptions).and_return(false)
+    # Reduce to 1 to have the test run faster
+    allow(settings.dynamic_instrumentation.internal).to receive(:min_send_interval).and_return(1)
   end
 
   let(:transport) do
@@ -81,22 +76,20 @@
         {hello: 'world'}
       end
 
-      xit 'sends the snapshot' do
+      it 'sends the snapshot' do
         expect(worker.send(:snapshot_queue)).to be_empty
 
         expect(transport).to receive(:send_snapshot).once.with([snapshot])
 
         worker.add_snapshot(snapshot)
 
-        # Since sending is asynchronous, we need to relinquish execution
-        # for the sending thread to run.
-        sleep(0.1)
+        worker.flush
 
         expect(worker.send(:snapshot_queue)).to eq([])
       end
 
       context 'when three snapshots are added in quick succession' do
-        xit 'sends two batches' do
+        it 'sends two batches' do
           expect(worker.send(:snapshot_queue)).to be_empty
 
           expect(transport).to receive(:send_snapshot).once.with([snapshot])
@@ -106,22 +99,15 @@
           worker.add_snapshot(snapshot)
           sleep 0.1
           worker.add_snapshot(snapshot)
-
-          # Since sending is asynchronous, we need to relinquish execution
-          # for the sending thread to run.
           sleep(0.1)
 
           # At this point the first snapshot should have been sent,
           # with the remaining two in the queue
           expect(worker.send(:snapshot_queue)).to eq([snapshot, snapshot])
 
-          sleep 0.4
-          # Still within the cooldown period
-          expect(worker.send(:snapshot_queue)).to eq([snapshot, snapshot])
-
           expect(transport).to receive(:send_snapshot).once.with([snapshot, snapshot])
 
-          sleep 0.5
+          worker.flush
           expect(worker.send(:snapshot_queue)).to eq([])
         end
       end
diff --git a/spec/datadog/di/spec_helper.rb b/spec/datadog/di/spec_helper.rb
index 5fe045e4cea..3ce872c2943 100644
--- a/spec/datadog/di/spec_helper.rb
+++ b/spec/datadog/di/spec_helper.rb
@@ -12,9 +12,34 @@ def di_test
         end
       end
     end
+
+    def mock_settings_for_di(&block)
+      let(:settings) do
+        double('settings').tap do |settings|
+          allow(settings).to receive(:dynamic_instrumentation).and_return(di_settings)
+          if block
+            instance_exec(settings, &block)
+          end
+        end
+      end
+
+      let(:di_settings) do
+        double('di settings').tap do |settings|
+          allow(settings).to receive(:internal).and_return(di_internal_settings)
+        end
+      end
+
+      let(:di_internal_settings) do
+        double('di internal settings')
+      end
+    end
+  end
+
+  module InstanceMethods
   end
 end
 
 RSpec.configure do |config|
   config.extend DIHelpers::ClassMethods
+  config.include DIHelpers::InstanceMethods
 end