DEBUG-3700 Replace telemetry transport with a core-based transport that supports UDS (#4630)

Author: p-datadog

.rubocop.yml

@@ -24,6 +24,11 @@ AllCops:
     - 'spec/**/**/interesting_backtrace_helper.rb' # This file needs quite a few bizarre code patterns by design
     - 'vendor/bundle/**/*'
     - 'spec/datadog/tracing/contrib/grpc/support/gen/**/*.rb' # Skip protoc autogenerated code
+    # Telemetry transport code is copied from DI and is formatted to
+    # standard rules. We are migrating to standard therefore avoid
+    # reformatting all of this code to rubocop's demands and then have it
+    # be different from telemetry.
+    - lib/datadog/core/telemetry/transport/**/*
     - lib/datadog/di/**/*
     - lib/datadog/di.rb
     - spec/datadog/di/**/*

lib/datadog/core/telemetry/component.rb

@@ -2,12 +2,13 @@
 
 require_relative 'emitter'
 require_relative 'event'
-require_relative 'http/transport'
 require_relative 'metrics_manager'
 require_relative 'worker'
 require_relative 'logging'
+require_relative 'transport/http'
 
 require_relative '../configuration/ext'
+require_relative '../configuration/agentless_settings_resolver'
 require_relative '../utils/forking'
 
 module Datadog
@@ -16,7 +17,7 @@ module Telemetry
       # Telemetry entrypoint, coordinates sending telemetry events at various points in app lifecycle.
       # Note: Telemetry does not spawn its worker thread in fork processes, thus no telemetry is sent in forked processes.
       class Component
-        attr_reader :enabled, :logger
+        attr_reader :enabled, :logger, :transport, :worker
 
         include Core::Utils::Forking
         include Telemetry::Logging
@@ -25,89 +26,83 @@ def self.build(settings, agent_settings, logger)
           enabled = settings.telemetry.enabled
           agentless_enabled = settings.telemetry.agentless_enabled
 
-          if !agentless_enabled && agent_settings.adapter != Datadog::Core::Configuration::Ext::Agent::HTTP::ADAPTER
-            enabled = false
-            logger.debug { "Telemetry disabled. Agent network adapter not supported: #{agent_settings.adapter}" }
-          end
-
           if agentless_enabled && settings.api_key.nil?
             enabled = false
-            logger.debug { 'Telemetry disabled. Agentless telemetry requires an DD_API_KEY variable to be set.' }
+            logger.debug { 'Telemetry disabled. Agentless telemetry requires a DD_API_KEY variable to be set.' }
           end
 
-          transport = if agentless_enabled
-                        Datadog::Core::Telemetry::Http::Transport.build_agentless_transport(
-                          api_key: settings.api_key,
-                          dd_site: settings.site,
-                          url_override: settings.telemetry.agentless_url_override
-                        )
-                      else
-                        Datadog::Core::Telemetry::Http::Transport.build_agent_transport(agent_settings)
-                      end
-
           Telemetry::Component.new(
-            http_transport: transport,
+            settings: settings,
+            agent_settings: agent_settings,
             enabled: enabled,
-            metrics_enabled: enabled && settings.telemetry.metrics_enabled,
-            heartbeat_interval_seconds: settings.telemetry.heartbeat_interval_seconds,
-            metrics_aggregation_interval_seconds: settings.telemetry.metrics_aggregation_interval_seconds,
-            dependency_collection: settings.telemetry.dependency_collection,
             logger: logger,
-            shutdown_timeout_seconds: settings.telemetry.shutdown_timeout_seconds,
-            log_collection_enabled: settings.telemetry.log_collection_enabled
           )
         end
 
         # @param enabled [Boolean] Determines whether telemetry events should be sent to the API
-        # @param metrics_enabled [Boolean] Determines whether telemetry metrics should be sent to the API
-        # @param heartbeat_interval_seconds [Float] How frequently heartbeats will be reported, in seconds.
-        # @param metrics_aggregation_interval_seconds [Float] How frequently metrics will be aggregated, in seconds.
-        # @param [Boolean] dependency_collection Whether to send the `app-dependencies-loaded` event
-        def initialize(
-          heartbeat_interval_seconds:,
-          metrics_aggregation_interval_seconds:,
-          dependency_collection:,
+        def initialize( # rubocop: disable Metrics/MethodLength
+          settings:,
+          agent_settings:,
           logger:,
-          http_transport:,
-          shutdown_timeout_seconds:,
-          enabled: true,
-          metrics_enabled: true,
-          log_collection_enabled: true
+          enabled:
         )
           @enabled = enabled
-          @log_collection_enabled = log_collection_enabled
+          @log_collection_enabled = settings.telemetry.log_collection_enabled
           @logger = logger
 
           @metrics_manager = MetricsManager.new(
-            enabled: enabled && metrics_enabled,
-            aggregation_interval: metrics_aggregation_interval_seconds
+            enabled: @enabled && settings.telemetry.metrics_enabled,
+            aggregation_interval: settings.telemetry.metrics_aggregation_interval_seconds,
           )
 
+          @stopped = false
+
+          return unless @enabled
+
+          @transport = if settings.telemetry.agentless_enabled
+                         agent_settings = Core::Configuration::AgentlessSettingsResolver.call(
+                           settings,
+                           host_prefix: 'instrumentation-telemetry-intake',
+                           url_override: settings.telemetry.agentless_url_override,
+                           url_override_source: 'c.telemetry.agentless_url_override',
+                           logger: logger,
+                         )
+                         Telemetry::Transport::HTTP.agentless_telemetry(
+                           agent_settings: agent_settings,
+                           logger: logger,
+                           # api_key should have already validated to be
+                           # not nil by +build+ method above.
+                           api_key: settings.api_key,
+                         )
+                       else
+                         Telemetry::Transport::HTTP.agent_telemetry(
+                           agent_settings: agent_settings, logger: logger,
+                         )
+                       end
+
           @worker = Telemetry::Worker.new(
             enabled: @enabled,
-            heartbeat_interval_seconds: heartbeat_interval_seconds,
-            metrics_aggregation_interval_seconds: metrics_aggregation_interval_seconds,
-            emitter: Emitter.new(http_transport: http_transport),
+            heartbeat_interval_seconds: settings.telemetry.heartbeat_interval_seconds,
+            metrics_aggregation_interval_seconds: settings.telemetry.metrics_aggregation_interval_seconds,
+            emitter: Emitter.new(@transport, logger: @logger),
             metrics_manager: @metrics_manager,
-            dependency_collection: dependency_collection,
+            dependency_collection: settings.telemetry.dependency_collection,
             logger: logger,
-            shutdown_timeout: shutdown_timeout_seconds
+            shutdown_timeout: settings.telemetry.shutdown_timeout_seconds,
           )
 
-          @stopped = false
-
           @worker.start
         end
 
         def disable!
           @enabled = false
-          @worker.enabled = false
+          @worker&.enabled = false
         end
 
         def stop!
           return if @stopped
 
-          @worker.stop(true)
+          @worker&.stop(true)
           @stopped = true
         end
 
@@ -135,6 +130,8 @@ def log!(event)
         #
         # @api private
         def flush
+          return if !@enabled || forked?
+
           @worker.flush
         end
 

lib/datadog/core/telemetry/emitter.rb

@@ -1,7 +1,6 @@
 # frozen_string_literal: true
 
 require_relative 'request'
-require_relative 'http/transport'
 require_relative '../transport/response'
 require_relative '../utils/sequence'
 require_relative '../utils/forking'
@@ -11,25 +10,26 @@ module Core
     module Telemetry
       # Class that emits telemetry events
       class Emitter
-        attr_reader :http_transport
+        attr_reader :transport, :logger
 
         extend Core::Utils::Forking
 
-        # @param http_transport [Datadog::Core::Telemetry::Http::Transport] Transport object that can be used to send
-        #   telemetry requests via the agent
-        def initialize(http_transport:)
-          @http_transport = http_transport
+        # @param transport [Datadog::Core::Telemetry::Transport::Telemetry::Transport]
+        #   Transport object that can be used to send telemetry requests
+        def initialize(transport, logger: Datadog.logger)
+          @transport = transport
+          @logger = logger
         end
 
         # Retrieves and emits a TelemetryRequest object based on the request type specified
         def request(event)
           seq_id = self.class.sequence.next
           payload = Request.build_payload(event, seq_id)
-          res = @http_transport.request(request_type: event.type, payload: payload.to_json)
-          Datadog.logger.debug { "Telemetry sent for event `#{event.type}` (response: #{res})" }
+          res = @transport.send_telemetry(request_type: event.type, payload: payload)
+          logger.debug { "Telemetry sent for event `#{event.type}` (response code: #{res.code})" }
           res
         rescue => e
-          Datadog.logger.debug("Unable to send telemetry request for event `#{event.type rescue 'unknown'}`: #{e}")
+          logger.debug { "Unable to send telemetry request for event `#{event.type rescue 'unknown'}`: #{e}" }
           Core::Transport::InternalErrorResponse.new(e)
         end
 

lib/datadog/core/telemetry/http/ext.rb

@@ -11,9 +11,9 @@ module Request
         class << self
           using Core::Utils::Hash::Refinement
 
-          def build_payload(event, seq_id)
+          def build_payload(event, seq_id, api_version: 'v2')
             hash = {
-              api_version: Http::Ext::API_VERSION,
+              api_version: api_version,
               application: application,
               debug: false,
               host: host,