Serialize span events via a dedicated field (#4279)

* Increase type checking coverage

* Serialize span events via a dedicated field

* Revert unrelated test changes

* Skip test running against incorrect combinations

* Use kwarg

Author: marcotc

.github/forced-tests-list.json

@@ -1,3 +1,10 @@
 {
-  
+  "AGENT_NOT_SUPPORTING_SPAN_EVENTS":
+  [
+    "tests/test_span_events.py"
+  ],
+  "PARAMETRIC":
+  [
+    "tests/parametric/test_span_events.py"
+  ]
 }

Steepfile

@@ -137,10 +137,8 @@ target :datadog do
   ignore 'lib/datadog/tracing/transport/http/traces.rb'
   ignore 'lib/datadog/tracing/transport/io/client.rb'
   ignore 'lib/datadog/tracing/transport/io/traces.rb'
-  ignore 'lib/datadog/tracing/transport/serializable_trace.rb'
   ignore 'lib/datadog/tracing/transport/statistics.rb'
   ignore 'lib/datadog/tracing/transport/trace_formatter.rb'
-  ignore 'lib/datadog/tracing/transport/traces.rb'
   ignore 'lib/datadog/tracing/workers.rb'
   ignore 'lib/datadog/tracing/workers/trace_writer.rb'
   ignore 'lib/datadog/tracing/writer.rb'

lib/datadog/core/configuration/components.rb

@@ -16,6 +16,8 @@
 require_relative '../../di/component'
 require_relative '../crashtracking/component'
 
+require_relative '../environment/agent_info'
+
 module Datadog
   module Core
     module Configuration
@@ -85,7 +87,8 @@ def build_crashtracker(settings, agent_settings, logger:)
           :tracer,
           :crashtracker,
           :dynamic_instrumentation,
-          :appsec
+          :appsec,
+          :agent_info
 
         def initialize(settings)
           @logger = self.class.build_logger(settings)
@@ -96,6 +99,9 @@ def initialize(settings)
           # the Core resolver from within your product/component's namespace.
           agent_settings = AgentSettingsResolver.call(settings, logger: @logger)
 
+          # Exposes agent capability information for detection by any components
+          @agent_info = Core::Environment::AgentInfo.new(agent_settings)
+
           @telemetry = self.class.build_telemetry(settings, agent_settings, @logger)
 
           @remote = Remote::Component.build(settings, agent_settings, telemetry: telemetry)

lib/datadog/core/encoding.rb

@@ -10,6 +10,7 @@ module Encoding
       # Encoder interface that provides the logic to encode traces and service
       # @abstract
       module Encoder
+        # :nocov:
         def content_type
           raise NotImplementedError
         end
@@ -23,6 +24,13 @@ def join(encoded_elements)
         def encode(_)
           raise NotImplementedError
         end
+
+        # Deserializes a value serialized with {#encode}.
+        # This method is used for debugging purposes.
+        def decode(_)
+          raise NotImplementedError
+        end
+        # :nocov:
       end
 
       # Encoder for the JSON format
@@ -41,6 +49,10 @@ def encode(obj)
           JSON.dump(obj)
         end
 
+        def decode(obj)
+          JSON.parse(obj)
+        end
+
         def join(encoded_data)
           "[#{encoded_data.join(',')}]"
         end
@@ -62,6 +74,10 @@ def encode(obj)
           MessagePack.pack(obj)
         end
 
+        def decode(obj)
+          MessagePack.unpack(obj)
+        end
+
         def join(encoded_data)
           packer = MessagePack::Packer.new
           packer.write_array_header(encoded_data.size)

lib/datadog/core/environment/agent_info.rb

@@ -0,0 +1,77 @@
+# frozen_string_literal: true
+
+module Datadog
+  module Core
+    module Environment
+      # Retrieves the agent's `/info` endpoint data.
+      # This data can be used to determine the capabilities of the local Datadog agent.
+      #
+      # @example Example response payload
+      #   {
+      #     "version" : "7.57.2",
+      #     "git_commit" : "38ba0c7",
+      #     "endpoints" : [ "/v0.4/traces", "/v0.4/services", "/v0.7/traces", "/v0.7/config" ],
+      #     "client_drop_p0s" : true,
+      #     "span_meta_structs" : true,
+      #     "long_running_spans" : true,
+      #     "evp_proxy_allowed_headers" : [ "Content-Type", "Accept-Encoding", "Content-Encoding", "User-Agent" ],
+      #     "config" : {
+      #       "default_env" : "none",
+      #       "target_tps" : 10,
+      #       "max_eps" : 200,
+      #       "receiver_port" : 8126,
+      #       "receiver_socket" : "/var/run/datadog/apm.socket",
+      #       "connection_limit" : 0,
+      #       "receiver_timeout" : 0,
+      #       "max_request_bytes" : 26214400,
+      #       "statsd_port" : 8125,
+      #       "analyzed_spans_by_service" : { },
+      #       "obfuscation" : {
+      #         "elastic_search" : true,
+      #         "mongo" : true,
+      #         "sql_exec_plan" : false,
+      #         "sql_exec_plan_normalize" : false,
+      #         "http" : {
+      #           "remove_query_string" : false,
+      #           "remove_path_digits" : false
+      #         },
+      #         "remove_stack_traces" : false,
+      #         "redis" : {
+      #           "Enabled" : true,
+      #           "RemoveAllArgs" : false
+      #         },
+      #         "memcached" : {
+      #           "Enabled" : true,
+      #           "KeepCommand" : false
+      #         }
+      #       }
+      #     },
+      #     "peer_tags" : null
+      #   }
+      #
+      # @see https://github.com/DataDog/datadog-agent/blob/f07df0a3c1fca0c83b5a15f553bd994091b0c8ac/pkg/trace/api/info.go#L20
+      class AgentInfo
+        attr_reader :agent_settings
+
+        def initialize(agent_settings)
+          @agent_settings = agent_settings
+          @client = Remote::Transport::HTTP.root(agent_settings: agent_settings)
+        end
+
+        # Fetches the information from the agent.
+        # @return [Datadog::Core::Remote::Transport::HTTP::Negotiation::Response] the response from the agent
+        # @return [nil] if an error occurred while fetching the information
+        def fetch
+          res = @client.send_info
+          return unless res.ok?
+
+          res
+        end
+
+        def ==(other)
+          other.is_a?(self.class) && other.agent_settings == agent_settings
+        end
+      end
+    end
+  end
+end

lib/datadog/core/remote/transport/http/negotiation.rb

@@ -43,6 +43,7 @@ def initialize(http_response, options = {})
                 @version = options[:version]
                 @endpoints = options[:endpoints]
                 @config = options[:config]
+                @span_events = options[:span_events]
               end
             end
 

lib/datadog/core/remote/transport/negotiation.rb

@@ -32,7 +32,19 @@ class Request < Datadog::Core::Transport::Request
 
           # Negotiation response
           module Response
-            attr_reader :version, :endpoints, :config
+            # @!attribute [r] version
+            #   The version of the agent.
+            #   @return [String]
+            # @!attribute [r] endpoints
+            #   The HTTP endpoints the agent supports.
+            #   @return [Array<String>]
+            # @!attribute [r] config
+            #   The agent configuration. These are configured by the user when starting the agent, as well as any defaults.
+            #   @return [Hash]
+            # @!attribute [r] span_events
+            #   Whether the agent supports the top-level span events field in flushed spans.
+            #   @return [Boolean,nil]
+            attr_reader :version, :endpoints, :config, :span_events
           end
 
           # Negotiation transport

lib/datadog/tracing/transport/serializable_trace.rb

@@ -14,7 +14,7 @@ class SerializableTrace
 
         # @param trace [Datadog::Trace] the trace to serialize
         # @param native_events_supported [Boolean] whether the agent supports span events as a top-level field
-        def initialize(trace, native_events_supported = false)
+        def initialize(trace, native_events_supported:)
           @trace = trace
           @native_events_supported = native_events_supported
         end
@@ -29,13 +29,17 @@ def initialize(trace, native_events_supported = false)
         # @param packer [MessagePack::Packer] serialization buffer, can be +nil+ with JRuby
         def to_msgpack(packer = nil)
           # As of 1.3.3, JRuby implementation doesn't pass an existing packer
-          trace.spans.map { |s| SerializableSpan.new(s, @native_events_supported) }.to_msgpack(packer)
+          trace.spans.map do |s|
+            SerializableSpan.new(s, native_events_supported: @native_events_supported)
+          end.to_msgpack(packer)
         end
 
         # JSON serializer interface.
         # Used by older version of the transport.
         def to_json(*args)
-          trace.spans.map { |s| SerializableSpan.new(s, @native_events_supported).to_hash }.to_json(*args)
+          trace.spans.map do |s|
+            SerializableSpan.new(s, native_events_supported: @native_events_supported).to_hash
+          end.to_json(*args)
         end
       end
 
@@ -46,7 +50,7 @@ class SerializableSpan
 
         # @param span [Datadog::Span] the span to serialize
         # @param native_events_supported [Boolean] whether the agent supports span events as a top-level field
-        def initialize(span, native_events_supported)
+        def initialize(span, native_events_supported:)
           @span = span
           @trace_id = Tracing::Utils::TraceId.to_low_order(span.trace_id)
           @native_events_supported = native_events_supported

lib/datadog/tracing/transport/traces.rb

@@ -50,8 +50,9 @@ class Chunker
           #
           # @param encoder [Datadog::Core::Encoding::Encoder]
           # @param max_size [String] maximum acceptable payload size
-          def initialize(encoder, max_size: DEFAULT_MAX_PAYLOAD_SIZE)
+          def initialize(encoder, native_events_supported:, max_size: DEFAULT_MAX_PAYLOAD_SIZE)
             @encoder = encoder
+            @native_events_supported = native_events_supported
             @max_size = max_size
           end
 
@@ -77,7 +78,7 @@ def encode_in_chunks(traces)
           private
 
           def encode_one(trace)
-            encoded = Encoder.encode_trace(encoder, trace)
+            encoded = Encoder.encode_trace(encoder, trace, native_events_supported: @native_events_supported)
 
             if encoded.size > max_size
               # This single trace is too large, we can't flush it
@@ -95,17 +96,18 @@ def encode_one(trace)
         module Encoder
           module_function
 
-          def encode_trace(encoder, trace)
+          def encode_trace(encoder, trace, native_events_supported:)
             # Format the trace for transport
             TraceFormatter.format!(trace)
 
             # Make the trace serializable
-            serializable_trace = SerializableTrace.new(trace)
-
-            Datadog.logger.debug { "Flushing trace: #{JSON.dump(serializable_trace)}" }
+            serializable_trace = SerializableTrace.new(trace, native_events_supported: native_events_supported)
 
             # Encode the trace
-            encoder.encode(serializable_trace)
+            encoder.encode(serializable_trace).tap do |encoded|
+              # Print the actual serialized trace, since the encoder can change make non-trivial changes
+              Datadog.logger.debug { "Flushing trace: #{encoder.decode(encoded)}" }
+            end
           end
         end
 
@@ -126,7 +128,10 @@ def initialize(apis, default_api)
 
           def send_traces(traces)
             encoder = current_api.encoder
-            chunker = Datadog::Tracing::Transport::Traces::Chunker.new(encoder)
+            chunker = Datadog::Tracing::Transport::Traces::Chunker.new(
+              encoder,
+              native_events_supported: native_events_supported?
+            )
 
             responses = chunker.encode_in_chunks(traces.lazy).map do |encoded_traces, trace_count|
               request = Request.new(EncodedParcel.new(encoded_traces, trace_count))
@@ -188,6 +193,18 @@ def change_api!(api_id)
             @client = HTTP::Client.new(current_api)
           end
 
+          # Queries the agent for native span events serialization support.
+          # This changes how the serialization of span events performed.
+          def native_events_supported?
+            return @native_events_supported if defined?(@native_events_supported)
+
+            if (res = Datadog.send(:components).agent_info.fetch)
+              @native_events_supported = res.span_events == true
+            else
+              false
+            end
+          end
+
           # Raised when configured with an unknown API version
           class UnknownApiVersionError < StandardError
             attr_reader :version

sig/datadog/core/configuration/components.rbs

@@ -36,6 +36,8 @@ module Datadog
 
         attr_reader remote: Datadog::Core::Remote::Component
 
+        attr_reader agent_info: Datadog::Core::Environment::AgentInfo
+
         def initialize: (untyped settings) -> untyped
 
         def startup!: (untyped settings) -> untyped

sig/datadog/core/encoding.rbs

@@ -5,39 +5,43 @@ module Datadog
       # Encoder interface that provides the logic to encode traces and service
       # @abstract
       module Encoder
-        def content_type: () -> untyped
+        def content_type: () -> String
 
-        # Concatenates a list of elements previously encoded by +#encode+.
-        def join: (untyped encoded_elements) -> untyped
+        def encode: (untyped obj) -> String
 
-        # Serializes a single trace into a String suitable for network transmission.
-        def encode: (untyped _) -> untyped
+        def join: (Array[untyped] encoded_data) -> String
+
+        def decode: (String obj)-> untyped
       end
 
       # Encoder for the JSON format
       module JSONEncoder
         extend Encoder
 
-        CONTENT_TYPE: "application/json"
+        CONTENT_TYPE: String
+
+        def self?.content_type: () -> String
 
-        def self?.content_type: () -> untyped
+        def self?.encode: (untyped obj) -> String
 
-        def self?.encode: (untyped obj) -> untyped
+        def self?.join: (Array[untyped] encoded_data) -> String
 
-        def self?.join: (untyped encoded_data) -> ::String
+        def self?.decode: (String obj)-> untyped
       end
 
       # Encoder for the Msgpack format
       module MsgpackEncoder
         extend Encoder
 
-        CONTENT_TYPE: "application/msgpack"
+        CONTENT_TYPE: String
+
+        def self?.content_type: () -> String
 
-        def self?.content_type: () -> untyped
+        def self?.encode: (untyped obj) -> String
 
-        def self?.encode: (untyped obj) -> untyped
+        def self?.join: (Array[untyped] encoded_data) -> String
 
-        def self?.join: (untyped encoded_data) -> untyped
+        def self?.decode: (String obj)-> untyped
       end
     end
   end

sig/datadog/core/environment/agent_info.rbs

@@ -0,0 +1,13 @@
+module Datadog
+  module Core
+    module Environment
+      class AgentInfo
+        attr_reader agent_settings: Configuration::AgentSettingsResolver::AgentSettings
+
+        def initialize: (Configuration::AgentSettingsResolver::AgentSettings agent_settings) -> void
+
+        def fetch: -> Remote::Transport::HTTP::Negotiation::Response?
+      end
+    end
+  end
+end