[SDTEST-184] Agentless telemetry

lib/datadog/core/configuration/components.rb

@@ -57,17 +57,35 @@ def build_runtime_metrics_worker(settings)
 
           def build_telemetry(settings, agent_settings, logger)
             enabled = settings.telemetry.enabled
-            if agent_settings.adapter != Datadog::Core::Configuration::Ext::Agent::HTTP::ADAPTER
+            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.' }
+            end
+
+            transport = if agentless_enabled
+                          Datadog::Core::Telemetry::Http::Transport.build_agentless_transport(
+                            api_key: settings.api_key,
+                            dd_site: settings.site
+                          )
+                        else
+                          Datadog::Core::Telemetry::Http::Transport.build_agent_transport
+                        end
+
             Telemetry::Component.new(
+              http_transport: transport,
               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
+              dependency_collection: settings.telemetry.dependency_collection,
+              shutdown_timeout_seconds: settings.telemetry.shutdown_timeout_seconds,
             )
           end
         end

lib/datadog/core/configuration/settings.rb

@@ -663,6 +663,14 @@ def initialize(*_)
             o.type :bool
           end
 
+          # Enable agentless mode for telemetry: submit telemetry events directly to the intake without Datadog Agent.
+          #
+          # @return [Boolean]
+          option :agentless_enabled do |o|
+            o.type :bool
+            o.default false
+          end
+
           # Enable metrics collection for telemetry. Metrics collection only works when telemetry is enabled and
           # metrics are enabled.
           # @default `DD_TELEMETRY_METRICS_ENABLED` environment variable, otherwise `true`.
@@ -734,6 +742,14 @@ def initialize(*_)
             o.type :string, nilable: true
             o.env Core::Telemetry::Ext::ENV_INSTALL_TIME
           end
+
+          # Telemetry shutdown timeout in seconds
+          #
+          # @!visibility private
+          option :shutdown_timeout_seconds do |o|
+            o.type :float
+            o.default 1.0
+          end
         end
 
         # Remote configuration

lib/datadog/core/environment/ext.rb

@@ -40,6 +40,8 @@ module Ext
         TAG_SERVICE = 'service'
         TAG_VERSION = 'version'
 
+        DD_SITE_STAGING = 'datad0g.com'
+
         GEM_DATADOG_VERSION = Datadog::VERSION::STRING
       end
     end

lib/datadog/core/telemetry/component.rb

@@ -5,6 +5,7 @@
 require_relative 'metrics_manager'
 require_relative 'worker'
 require_relative '../utils/forking'
+require_relative '../workers/polling'
 
 module Datadog
   module Core
@@ -24,6 +25,8 @@ def initialize(
           heartbeat_interval_seconds:,
           metrics_aggregation_interval_seconds:,
           dependency_collection:,
+          http_transport:,
+          shutdown_timeout_seconds:,
           enabled: true,
           metrics_enabled: true
         )
@@ -39,9 +42,10 @@ def initialize(
             enabled: @enabled,
             heartbeat_interval_seconds: heartbeat_interval_seconds,
             metrics_aggregation_interval_seconds: metrics_aggregation_interval_seconds,
-            emitter: Emitter.new,
+            emitter: Emitter.new(http_transport: http_transport),
             metrics_manager: @metrics_manager,
-            dependency_collection: dependency_collection
+            dependency_collection: dependency_collection,
+            shutdown_timeout: shutdown_timeout_seconds
           )
           @worker.start
         end

lib/datadog/core/telemetry/emitter.rb

@@ -16,22 +16,20 @@ class Emitter
 
         # @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: Datadog::Core::Telemetry::Http::Transport.new)
+        def initialize(http_transport:)
           @http_transport = http_transport
         end
 
         # Retrieves and emits a TelemetryRequest object based on the request type specified
         def request(event)
-          begin
-            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}` (code: #{res.code.inspect})" }
-            res
-          rescue => e
-            Datadog.logger.debug("Unable to send telemetry request for event `#{event.type rescue 'unknown'}`: #{e}")
-            Telemetry::Http::InternalErrorResponse.new(e)
-          end
+          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}` (code: #{res.code.inspect})" }
+          res
+        rescue => e
+          Datadog.logger.debug("Unable to send telemetry request for event `#{event.type rescue 'unknown'}`: #{e}")
+          Telemetry::Http::InternalErrorResponse.new(e)
         end
 
         # Initializes a Sequence object to track seq_id if not already initialized; else returns stored

lib/datadog/core/telemetry/event.rb

@@ -18,7 +18,9 @@ class Base
           # The type of the event.
           # It must be one of the stings defined in the Telemetry V2
           # specification for event names.
-          def type; end
+          def type
+            raise NotImplementedError, 'Must be implemented by subclass'
+          end
 
           # The JSON payload for the event.
           def payload

lib/datadog/core/telemetry/http/adapters/net.rb

@@ -34,19 +34,17 @@ def open(&block)
             end
 
             def post(env)
-              begin
-                post = ::Net::HTTP::Post.new(env.path, env.headers)
-                post.body = env.body
-
-                http_response = open do |http|
-                  http.request(post)
-                end
-
-                Response.new(http_response)
-              rescue StandardError => e
-                Datadog.logger.debug("Unable to send telemetry event to agent: #{e}")
-                Telemetry::Http::InternalErrorResponse.new(e)
+              post = ::Net::HTTP::Post.new(env.path, env.headers)
+              post.body = env.body
+
+              http_response = open do |http|
+                http.request(post)
               end
+
+              Response.new(http_response)
+            rescue StandardError => e
+              Datadog.logger.debug("Unable to send telemetry event to agent: #{e}")
+              Telemetry::Http::InternalErrorResponse.new(e)
             end
 
             # Data structure for an HTTP Response

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

@@ -17,7 +17,10 @@ module Ext
           CONTENT_TYPE_APPLICATION_JSON = 'application/json'
           API_VERSION = 'v2'
 
+          AGENTLESS_HOST_PREFIX = 'instrumentation-telemetry-intake'
+
           AGENT_ENDPOINT = '/telemetry/proxy/api/v2/apmtelemetry'
+          AGENTLESS_ENDPOINT = '/api/v2/apmtelemetry'
         end
       end
     end

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

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require_relative '../../configuration/settings'
+require_relative '../../environment/ext'
 require_relative '../../transport/ext'
 require_relative 'env'
 require_relative 'ext'
@@ -13,18 +14,47 @@ module Http
         # Class to send telemetry data to Telemetry API
         # Currently only supports the HTTP protocol.
         class Transport
+          def self.build_agent_transport
+            agent_settings = Configuration::AgentSettingsResolver.call(Datadog.configuration)
+
+            Transport.new(
+              host: agent_settings.hostname,
+              port: agent_settings.port,
+              path: Http::Ext::AGENT_ENDPOINT
+            )
+          end
+
+          def self.build_agentless_transport(api_key:, dd_site:)
+            host =
+              if dd_site == Environment::Ext::DD_SITE_STAGING
+                # special case for staging - the url is constructed differently
+                'all-http-intake.logs.datad0g.com'
+              else
+                "#{Http::Ext::AGENTLESS_HOST_PREFIX}.#{dd_site}"
+              end
+
+            Transport.new(
+              host: host,
+              port: 443,
+              path: Http::Ext::AGENTLESS_ENDPOINT,
+              ssl: true,
+              api_key: api_key
+            )
+          end
+
           attr_reader \
             :host,
             :port,
             :ssl,
-            :path
+            :path,
+            :api_key
 
-          def initialize
-            agent_settings = Configuration::AgentSettingsResolver.call(Datadog.configuration)
-            @host = agent_settings.hostname
-            @port = agent_settings.port
-            @ssl = false
-            @path = Http::Ext::AGENT_ENDPOINT
+          def initialize(host:, port:, path:, ssl: false, api_key: nil)
+            @host = host
+            @port = port
+            @ssl = ssl
+            @path = path
+            @api_key = api_key
           end
 
           def request(request_type:, payload:)
@@ -38,7 +68,7 @@ def request(request_type:, payload:)
           private
 
           def headers(request_type:, api_version: Http::Ext::API_VERSION)
-            {
+            result = {
               Core::Transport::Ext::HTTP::HEADER_DD_INTERNAL_UNTRACED_REQUEST => '1',
               Ext::HEADER_CONTENT_TYPE => Http::Ext::CONTENT_TYPE_APPLICATION_JSON,
               Ext::HEADER_DD_TELEMETRY_API_VERSION => api_version,
@@ -49,6 +79,10 @@ def headers(request_type:, api_version: Http::Ext::API_VERSION)
               # Enable debug mode for telemetry
               # HEADER_TELEMETRY_DEBUG_ENABLED => 'true',
             }
+
+            result[Ext::HEADER_DD_API_KEY] = api_key unless api_key.nil?
+
+            result
           end
 
           def adapter

sig/datadog/core/environment/ext.rbs

@@ -2,21 +2,23 @@ module Datadog
   module Core
     module Environment
       module Ext
-        LANG: untyped
+        LANG: String
 
-        LANG_ENGINE: untyped
+        LANG_ENGINE: String
 
-        LANG_INTERPRETER: untyped
+        LANG_INTERPRETER: String
 
-        LANG_PLATFORM: untyped
+        LANG_PLATFORM: String
 
-        LANG_VERSION: untyped
+        LANG_VERSION: String
 
-        RUBY_ENGINE: untyped
+        RUBY_ENGINE: String
 
-        GEM_DATADOG_VERSION: untyped
+        GEM_DATADOG_VERSION: String
 
-        ENGINE_VERSION: untyped
+        ENGINE_VERSION: String
+
+        DD_SITE_STAGING: String
       end
     end
   end

sig/datadog/core/telemetry/component.rbs

@@ -11,7 +11,7 @@ module Datadog
 
         include Core::Utils::Forking
 
-        def initialize: (heartbeat_interval_seconds: Float, metrics_aggregation_interval_seconds: Float, dependency_collection: bool, ?enabled: bool, ?metrics_enabled: bool) -> void
+        def initialize: (http_transport: Datadog::Core::Telemetry::Http::Transport, heartbeat_interval_seconds: Float, metrics_aggregation_interval_seconds: Float, dependency_collection: bool, ?enabled: bool, ?metrics_enabled: bool, shutdown_timeout_seconds: Float) -> void
 
         def disable!: () -> void
 
@@ -22,6 +22,16 @@ module Datadog
         def stop!: () -> void
 
         def integrations_change!: () -> void
+
+        def inc: (String namespace, String metric_name, Datadog::Core::Telemetry::Metric::input_value value, ?tags: Datadog::Core::Telemetry::Metric::tags_input, ?common: bool) -> void
+
+        def dec: (String namespace, String metric_name, Datadog::Core::Telemetry::Metric::input_value value, ?tags: Datadog::Core::Telemetry::Metric::tags_input, ?common: bool) -> void
+
+        def gauge: (String namespace, String metric_name, Datadog::Core::Telemetry::Metric::input_value value, ?tags: Datadog::Core::Telemetry::Metric::tags_input, ?common: bool) -> void
+
+        def rate: (String namespace, String metric_name, Datadog::Core::Telemetry::Metric::input_value value, ?tags: Datadog::Core::Telemetry::Metric::tags_input, ?common: bool) -> void
+
+        def distribution: (String namespace, String metric_name, Datadog::Core::Telemetry::Metric::input_value value, ?tags: Datadog::Core::Telemetry::Metric::tags_input, ?common: bool) -> void
       end
     end
   end

sig/datadog/core/telemetry/emitter.rbs

@@ -7,7 +7,7 @@ module Datadog
         attr_reader http_transport: untyped
 
         extend Core::Utils::Forking
-        def initialize: (?http_transport: untyped) -> void
+        def initialize: (http_transport: untyped) -> void
         def request: (Datadog::Core::Telemetry::Event::Base event) -> Datadog::Core::Telemetry::Http::Adapters::Net::Response
         def self.sequence: () -> untyped
       end

sig/datadog/core/telemetry/event.rbs

@@ -10,7 +10,7 @@ module Datadog
 
         class Base
           def payload: () -> (Hash[Symbol, untyped] | Array[Hash[Symbol, untyped]])
-          def type: -> String?
+          def type: -> String
         end
 
         class AppStarted < Base

sig/datadog/core/telemetry/http/ext.rbs

@@ -23,7 +23,11 @@ module Datadog
 
           API_VERSION: "v2"
 
+          AGENTLESS_HOST_PREFIX: "instrumentation-telemetry-intake"
+
           AGENT_ENDPOINT: "/telemetry/proxy/api/v2/apmtelemetry"
+
+          AGENTLESS_ENDPOINT: "/api/v2/apmtelemetry"
         end
       end
     end

sig/datadog/core/telemetry/http/transport.rbs

@@ -13,7 +13,9 @@ module Datadog
 
           attr_reader path: String
 
-          def initialize: () -> void
+          attr_reader api_key: String?
+
+          def initialize: (host: String, port: Integer, path: String, ?ssl: bool, ?api_key: String?) -> void
 
           def request: (request_type: String, payload: String) -> Datadog::Core::Telemetry::Http::Adapters::Net::response
 

sig/datadog/core/telemetry/worker.rbs

@@ -21,7 +21,7 @@ module Datadog
         @ticks_per_heartbeat: Integer
         @current_ticks: Integer
 
-        def initialize: (?enabled: bool, heartbeat_interval_seconds: Float, metrics_aggregation_interval_seconds: Float, emitter: Emitter, metrics_manager: MetricsManager, ?shutdown_timeout: Integer, ?buffer_size: Integer, dependency_collection: bool) -> void
+        def initialize: (?enabled: bool, heartbeat_interval_seconds: Float, metrics_aggregation_interval_seconds: Float, emitter: Emitter, metrics_manager: MetricsManager, ?shutdown_timeout: Float, ?buffer_size: Integer, dependency_collection: bool) -> void
 
         def start: () -> void