DEBUG-2334 Remaining dynamic instrumentation code

benchmarks/di_instrument.rb

@@ -73,7 +73,7 @@ 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
+    settings.dynamic_instrumentation.internal.untargeted_trace_points = true
     redactor = Datadog::DI::Redactor.new(settings)
     serializer = Datadog::DI::Serializer.new(settings, redactor)
     logger = Logger.new(STDERR)

lib/datadog.rb

@@ -7,4 +7,5 @@
 # Load other products (must follow tracing)
 require_relative 'datadog/profiling'
 require_relative 'datadog/appsec'
+require_relative 'datadog/di'
 require_relative 'datadog/kit'

lib/datadog/core/configuration.rb

@@ -258,8 +258,12 @@ def build_components(settings)
       def replace_components!(settings, old)
         components = Components.new(settings)
 
+        old_state = {
+          remote: old.remote&.started?,
+        }
+
         old.shutdown!(components)
-        components.startup!(settings)
+        components.startup!(settings, old_state: old_state)
         components
       end
 

lib/datadog/core/configuration/components.rb

@@ -13,6 +13,7 @@
 require_relative '../../tracing/component'
 require_relative '../../profiling/component'
 require_relative '../../appsec/component'
+require_relative '../../di/component'
 require_relative '../crashtracking/component'
 
 module Datadog
@@ -83,6 +84,7 @@ def build_crashtracker(settings, agent_settings, logger:)
           :telemetry,
           :tracer,
           :crashtracker,
+          :dynamic_instrumentation,
           :appsec
 
         def initialize(settings)
@@ -110,12 +112,13 @@ def initialize(settings)
           @runtime_metrics = self.class.build_runtime_metrics_worker(settings)
           @health_metrics = self.class.build_health_metrics(settings)
           @appsec = Datadog::AppSec::Component.build_appsec_component(settings, telemetry: telemetry)
+          @dynamic_instrumentation = Datadog::DI::Component.build(settings, agent_settings, telemetry: telemetry)
 
           self.class.configure_tracing(settings)
         end
 
         # Starts up components
-        def startup!(settings)
+        def startup!(settings, old_state: nil)
           if settings.profiling.enabled
             if profiler
               profiler.start
@@ -126,6 +129,11 @@ def startup!(settings)
             end
           end
 
+          if settings.remote.enabled && old_state&.[](:remote)
+            # remote should be defined here
+            remote.start
+          end
+
           Core::Diagnostics::EnvironmentLogger.collect_and_log!(@environment_logger_extra)
         end
 

lib/datadog/core/remote/client.rb

@@ -30,87 +30,96 @@ def sync
           response = transport.send_config(payload)
 
           if response.ok?
-            # when response is completely empty, do nothing as in: leave as is
-            if response.empty?
-              Datadog.logger.debug { 'remote: empty response => NOOP' }
+            process_response(response)
+          elsif response.internal_error?
+            raise TransportError, response.to_s
+          end
+        end
+        # rubocop:enable Metrics/AbcSize,Metrics/PerceivedComplexity,Metrics/MethodLength,Metrics/CyclomaticComplexity
 
-              return
-            end
+        private
 
-            begin
-              paths = response.client_configs.map do |path|
-                Configuration::Path.parse(path)
-              end
+        def process_response(response)
+          # when response is completely empty, do nothing as in: leave as is
+          if response.empty?
+            Datadog.logger.debug { 'remote: empty response => NOOP' }
 
-              targets = Configuration::TargetMap.parse(response.targets)
+            return
+          end
 
-              contents = Configuration::ContentList.parse(response.target_files)
-            rescue Remote::Configuration::Path::ParseError => e
-              raise SyncError, e.message
+          begin
+            paths = response.client_configs.map do |path|
+              Configuration::Path.parse(path)
             end
 
-            # To make sure steep does not complain
-            return unless paths && targets && contents
+            targets = Configuration::TargetMap.parse(response.targets)
 
-            # TODO: sometimes it can strangely be so that paths.empty?
-            # TODO: sometimes it can strangely be so that targets.empty?
+            contents = Configuration::ContentList.parse(response.target_files)
+          rescue Remote::Configuration::Path::ParseError => e
+            raise SyncError, e.message
+          end
 
-            changes = repository.transaction do |current, transaction|
-              # paths to be removed: previously applied paths minus ingress paths
-              (current.paths - paths).each { |p| transaction.delete(p) }
+          # To make sure steep does not complain
+          return unless paths && targets && contents
 
-              # go through each ingress path
-              paths.each do |path|
-                # match target with path
-                target = targets[path]
+          # TODO: sometimes it can strangely be so that paths.empty?
+          # TODO: sometimes it can strangely be so that targets.empty?
 
-                # abort entirely if matching target not found
-                raise SyncError, "no target for path '#{path}'" if target.nil?
+          apply_config(paths, targets, contents)
+        end
 
-                # new paths are not in previously applied paths
-                new = !current.paths.include?(path)
+        def apply_config(paths, targets, contents)
 
-                # updated paths are in previously applied paths
-                # but the content hash changed
-                changed = current.paths.include?(path) && !current.contents.find_content(path, target)
+          changes = repository.transaction do |current, transaction|
+            # paths to be removed: previously applied paths minus ingress paths
+            (current.paths - paths).each { |p| transaction.delete(p) }
 
-                # skip if unchanged
-                same = !new && !changed
+            # go through each ingress path
+            paths.each do |path|
+              # match target with path
+              target = targets[path]
 
-                next if same
+              # abort entirely if matching target not found
+              raise SyncError, "no target for path '#{path}'" if target.nil?
 
-                # match content with path and target
-                content = contents.find_content(path, target)
+              # new paths are not in previously applied paths
+              new = !current.paths.include?(path)
 
-                # abort entirely if matching content not found
-                raise SyncError, "no valid content for target at path '#{path}'" if content.nil?
+              # updated paths are in previously applied paths
+              # but the content hash changed
+              changed = current.paths.include?(path) && !current.contents.find_content(path, target)
 
-                # to be added or updated << config
-                # TODO: metadata (hash, version, etc...)
-                transaction.insert(path, target, content) if new
-                transaction.update(path, target, content) if changed
-              end
+              # skip if unchanged
+              same = !new && !changed
 
-              # save backend opaque backend state
-              transaction.set(opaque_backend_state: targets.opaque_backend_state)
-              transaction.set(targets_version: targets.version)
+              next if same
 
-              # upon transaction end, new list of applied config + metadata (add, change, remove) will be saved
-              # TODO: also remove stale config (matching removed) from cache (client configs is exhaustive list of paths)
-            end
+              # match content with path and target
+              content = contents.find_content(path, target)
 
-            if changes.empty?
-              Datadog.logger.debug { 'remote: no changes' }
-            else
-              dispatcher.dispatch(changes, repository)
+              # abort entirely if matching content not found
+              raise SyncError, "no valid content for target at path '#{path}'" if content.nil?
+
+              # to be added or updated << config
+              # TODO: metadata (hash, version, etc...)
+              transaction.insert(path, target, content) if new
+              transaction.update(path, target, content) if changed
             end
-          elsif response.internal_error?
-            raise TransportError, response.to_s
+
+            # save backend opaque backend state
+            transaction.set(opaque_backend_state: targets.opaque_backend_state)
+            transaction.set(targets_version: targets.version)
+
+            # upon transaction end, new list of applied config + metadata (add, change, remove) will be saved
+            # TODO: also remove stale config (matching removed) from cache (client configs is exhaustive list of paths)
           end
-        end
-        # rubocop:enable Metrics/AbcSize,Metrics/PerceivedComplexity,Metrics/MethodLength,Metrics/CyclomaticComplexity
 
-        private
+          if changes.empty?
+            Datadog.logger.debug { 'remote: no changes' }
+          else
+            dispatcher.dispatch(changes, repository)
+          end
+        end
 
         def payload # rubocop:disable Metrics/MethodLength
           state = repository.state

lib/datadog/core/remote/client/capabilities.rb

@@ -32,6 +32,12 @@ def register(settings)
               register_receivers(Datadog::AppSec::Remote.receivers(@telemetry))
             end
 
+            if settings.respond_to?(:dynamic_instrumentation) && settings.dynamic_instrumentation.enabled
+              register_capabilities(Datadog::DI::Remote.capabilities)
+              register_products(Datadog::DI::Remote.products)
+              register_receivers(Datadog::DI::Remote.receivers(@telemetry))
+            end
+
             register_capabilities(Datadog::Tracing::Remote.capabilities)
             register_products(Datadog::Tracing::Remote.products)
             register_receivers(Datadog::Tracing::Remote.receivers(@telemetry))

lib/datadog/core/workers/polling.rb

@@ -23,6 +23,15 @@ def perform(*args)
           end
         end
 
+        # Stops the worker thread.
+        #
+        # Signals the worker loop to stop and waits up to the timeout value
+        # (in seconds) for the thread to end.
+        # Then, if force_stop is true, terminates the thread by calling
+        # Thread#terminate.
+        #
+        # Returns true if the thread was terminated, false if the
+        # thread continues to run.
         def stop(force_stop = false, timeout = DEFAULT_SHUTDOWN_TIMEOUT)
           if running?
             # Attempt graceful stop and wait
@@ -34,12 +43,9 @@ def stop(force_stop = false, timeout = DEFAULT_SHUTDOWN_TIMEOUT)
                 "Timeout while waiting for worker to finish gracefully, forcing termination for: #{self}"
               end
               terminate
-            else
-              graceful
             end
-          else
-            false
           end
+          !running?
         end
 
         def enabled?

lib/datadog/di.rb

@@ -1,21 +1,44 @@
 # frozen_string_literal: true
 
 require_relative 'di/error'
-require_relative 'di/configuration'
 require_relative 'di/code_tracker'
+require_relative 'di/component'
+require_relative 'di/configuration'
 require_relative 'di/extensions'
 require_relative 'di/instrumenter'
 require_relative 'di/probe'
+require_relative 'di/probe_builder'
+require_relative 'di/probe_manager'
+require_relative 'di/probe_notification_builder'
+require_relative 'di/probe_notifier_worker'
 require_relative 'di/redactor'
+require_relative 'di/remote'
 require_relative 'di/serializer'
 require_relative 'di/transport'
-require_relative 'di/utils'
+
+if defined?(ActiveRecord::Base)
+  # The third-party library integrations need to be loaded after the
+  # third-party libraries are loaded. Tracing and appsec use Railtie
+  # to delay integrations until all of the application's dependencies
+  # are loaded, when running under Rails. We should do the same here in
+  # principle, however DI currently only has an ActiveRecord integration
+  # and AR should be loaded before any application code is loaded, being
+  # part of Rails, therefore for now we should be OK to just require the
+  # AR integration from here.
+  require_relative 'di/contrib/active_record'
+end
 
 module Datadog
   # Namespace for Datadog dynamic instrumentation.
   #
   # @api private
   module DI
+    class << self
+      def enabled?
+        Datadog.configuration.dynamic_instrumentation.enabled
+      end
+    end
+
     # Expose DI to global shared objects
     Extensions.activate!
 
@@ -33,6 +56,8 @@ class << self
       # existing mappings in the registry
       def activate_tracking!
         (@code_tracker ||= CodeTracker.new).start
+        # & is demanded by steep, code tracker is always not nil here.
+        #code_tracker&.start
       end
 
       # Deactivates code tracking. In normal usage of DI this method should
@@ -52,6 +77,14 @@ def deactivate_tracking!
       def code_tracking_active?
         code_tracker&.active? || false
       end
+
+      def component
+        Datadog.send(:components).dynamic_instrumentation
+      end
     end
   end
 end
+
+# Activate code tracking by default because line trace points will not work
+# without it.
+Datadog::DI.activate_tracking!