event validation, optimize serializable events creation

lib/datadog/ci/test_visibility/serializers/base.rb

@@ -5,6 +5,10 @@ module CI
     module TestVisibility
       module Serializers
         class Base
+          MINIMUM_TIMESTAMP_NANO = 946684800000000000
+          MINIMUM_DURATION_NANO = 0
+          MAXIMUM_DURATION_NANO = 9223372036854775807
+
           attr_reader :trace, :span
 
           def initialize(trace, span)
@@ -34,6 +38,11 @@ def to_msgpack(packer = nil)
             end
           end
 
+          # validates according to citestcycle json schema
+          def valid?
+            required_fields_present? && valid_start_time? && valid_duration?
+          end
+
           def content_fields
             []
           end
@@ -82,11 +91,11 @@ def service
           end
 
           def start
-            time_nano(@span.start_time)
+            @start ||= time_nano(@span.start_time)
           end
 
           def duration
-            duration_nano(@span.duration)
+            @duration ||= duration_nano(@span.duration)
           end
 
           def meta
@@ -113,21 +122,35 @@ def self.calculate_content_map_size(fields_list)
 
           private
 
+          def valid_start_time?
+            !start.nil? && start >= MINIMUM_TIMESTAMP_NANO
+          end
+
+          def valid_duration?
+            !duration.nil? && duration >= MINIMUM_DURATION_NANO && duration <= MAXIMUM_DURATION_NANO
+          end
+
+          def required_fields_present?
+            required_fields.all? { |field| !send(field).nil? }
+          end
+
+          def required_fields
+            []
+          end
+
           def write_field(packer, field_name, method = nil)
             method ||= field_name
 
             packer.write(field_name)
             packer.write(send(method))
           end
 
-          # Used for serialization
-          # @return [Integer] in nanoseconds since Epoch
+          # in nanoseconds since Epoch
           def time_nano(time)
             time.to_i * 1000000000 + time.nsec
           end
 
-          # Used for serialization
-          # @return [Integer] in nanoseconds since Epoch
+          # in nanoseconds
           def duration_nano(duration)
             (duration * 1e9).to_i
           end

lib/datadog/ci/test_visibility/serializers/span.rb

@@ -17,6 +17,16 @@ class Span < Base
 
           CONTENT_MAP_SIZE = calculate_content_map_size(CONTENT_FIELDS)
 
+          REQUIRED_FIELDS = [
+            "trace_id",
+            "span_id",
+            "error",
+            "name",
+            "resource",
+            "start",
+            "duration"
+          ].freeze
+
           def content_fields
             CONTENT_FIELDS
           end
@@ -28,6 +38,12 @@ def content_map_size
           def type
             "span"
           end
+
+          private
+
+          def required_fields
+            REQUIRED_FIELDS
+          end
         end
       end
     end

lib/datadog/ci/test_visibility/serializers/test_v1.rb

@@ -18,6 +18,16 @@ class TestV1 < Base
 
           CONTENT_MAP_SIZE = calculate_content_map_size(CONTENT_FIELDS)
 
+          REQUIRED_FIELDS = [
+            "trace_id",
+            "span_id",
+            "error",
+            "name",
+            "resource",
+            "start",
+            "duration"
+          ].freeze
+
           def content_fields
             CONTENT_FIELDS
           end
@@ -37,6 +47,12 @@ def name
           def resource
             "#{@span.get_tag(Ext::Test::TAG_SUITE)}.#{@span.get_tag(Ext::Test::TAG_NAME)}"
           end
+
+          private
+
+          def required_fields
+            REQUIRED_FIELDS
+          end
         end
       end
     end

lib/datadog/ci/test_visibility/transport.rb

@@ -2,6 +2,7 @@
 
 require "msgpack"
 require "datadog/core/encoding"
+require "datadog/core/environment/identity"
 # use it to chunk payloads by size
 # require "datadog/core/chunker"
 
@@ -24,20 +25,16 @@ def initialize(api_key:, site: "datadoghq.com", serializer: Datadog::CI::TestVis
         end
 
         def send_traces(traces)
-          # convert traces to events and construct payload
-          # TODO: encode events immediately?
-          events = traces.flat_map { |trace| @serializer.convert_trace_to_serializable_events(trace) }
+          return [] if traces.nil? || traces.empty?
 
-          # move this to validation
-          events = events.filter { |event| event.start >= 946684800000000000 && event.duration > 0 }
+          events = serialize_traces(traces)
 
           if events.empty?
             Datadog.logger.debug("[TestVisibility::Transport] empty events list, skipping send")
             return []
           end
 
           payload = Payload.new(events)
-
           encoded_payload = encoder.encode(payload)
 
           response = @http.request(
@@ -55,6 +52,22 @@ def send_traces(traces)
 
         private
 
+        def serialize_traces(traces)
+          # TODO: replace map.filter with filter_map when 1.0 is released
+          traces.flat_map do |trace|
+            trace.spans.map do |span|
+              event = @serializer.convert_span_to_serializable_event(trace, span)
+
+              if event.valid?
+                event
+              else
+                Datadog.logger.debug { "Invalid span skipped: #{span}" }
+                nil
+              end
+            end.filter { |event| !event.nil? }
+          end
+        end
+
         def encoder
           Datadog::Core::Encoding::MsgpackEncoder
         end
@@ -79,15 +92,11 @@ def to_msgpack(packer)
             packer.write("*")
             packer.write_map_header(3)
 
-            # TODO: implement our own identity?
-            first_event = @events.first
-            if first_event
-              packer.write("runtime-id")
-              packer.write(first_event.runtime_id)
-            end
+            packer.write("runtime-id")
+            packer.write(Datadog::Core::Environment::Identity.id)
 
             packer.write("language")
-            packer.write("ruby")
+            packer.write(Datadog::Core::Environment::Identity.lang)
 
             packer.write("library_version")
             packer.write(Datadog::CI::VERSION::STRING)

sig/datadog/ci/test_visibility/serializers/base.rbs

@@ -3,7 +3,13 @@ module Datadog
     module TestVisibility
       module Serializers
         class Base
+          MINIMUM_TIMESTAMP_NANO: 946684800000000000
+          MINIMUM_DURATION_NANO: 0
+          MAXIMUM_DURATION_NANO: 9223372036854775807
+
           @content_fields_count: Integer
+          @start: Integer
+          @duration: Integer
 
           attr_reader trace: Datadog::Tracing::TraceSegment
           attr_reader span: Datadog::Tracing::Span
@@ -12,6 +18,7 @@ module Datadog
 
           def to_msgpack: (?untyped? packer) -> untyped
 
+          def valid?: () -> bool
           def content_fields: () -> ::Array[String | Hash[String, String]]
           def content_map_size: () -> Integer
 
@@ -49,6 +56,11 @@ module Datadog
 
           private
 
+          def valid_start_time?: () -> bool
+          def valid_duration?: () -> bool
+          def required_fields_present?: () -> bool
+          def required_fields: () -> Array[String]
+
           def write_field: (untyped packer, String field_name, ?String? method) -> untyped
           def time_nano: (Time time) -> Integer
           def duration_nano: (Float duration) -> Integer

sig/datadog/ci/test_visibility/serializers/span.rbs

@@ -5,7 +5,9 @@ module Datadog
         class Span < Base
           CONTENT_FIELDS: Array[String | Hash[String, String]]
           CONTENT_MAP_SIZE: Integer
+          REQUIRED_FIELDS: Array[String]
 
+          def required_fields: () -> Array[String]
           def content_fields: () -> Array[String | Hash[String, String]]
           def content_map_size: () -> Integer
           def type: () -> "span"

sig/datadog/ci/test_visibility/serializers/test_v1.rbs

@@ -5,7 +5,9 @@ module Datadog
         class TestV1 < Base
           CONTENT_FIELDS: Array[String | Hash[String, String]]
           CONTENT_MAP_SIZE: Integer
+          REQUIRED_FIELDS: Array[String]
 
+          def required_fields: () -> Array[String]
           def content_fields: () -> Array[String | Hash[String, String]]
           def content_map_size: () -> Integer
 

sig/datadog/ci/test_visibility/transport.rbs

@@ -12,6 +12,8 @@ module Datadog
 
         private
 
+        def serialize_traces: (Array[Datadog::Tracing::TraceSegment] traces) -> ::Array[Datadog::CI::TestVisibility::Serializers::Base]
+
         def encoder: () -> singleton(Datadog::Core::Encoding::MsgpackEncoder)
 
         class Payload

spec/datadog/ci/test_visibility/serializers/span_spec.rb

@@ -50,4 +50,18 @@
       end
     end
   end
+
+  describe "valid?" do
+    context "required fields" do
+      context "when not present" do
+        before do
+          produce_test_trace(with_http_span: true)
+
+          tracer_span.name = nil
+        end
+
+        it { is_expected.not_to be_valid }
+      end
+    end
+  end
 end

spec/datadog/ci/test_visibility/serializers/test_v1_spec.rb

@@ -73,4 +73,50 @@
       end
     end
   end
+
+  describe "#valid?" do
+    context "duration" do
+      before do
+        produce_test_trace
+        span.duration = duration
+      end
+
+      context "when positive number" do
+        let(:duration) { 42 }
+
+        it { is_expected.to be_valid }
+      end
+
+      context "when negative number" do
+        let(:duration) { -1 }
+
+        it { is_expected.not_to be_valid }
+      end
+
+      context "when too big" do
+        let(:duration) { Datadog::CI::TestVisibility::Serializers::Base::MAXIMUM_DURATION_NANO + 1 }
+
+        it { is_expected.not_to be_valid }
+      end
+    end
+
+    context "start" do
+      before do
+        produce_test_trace
+        span.start_time = start_time
+      end
+
+      context "when now" do
+        let(:start_time) { Time.now }
+
+        it { is_expected.to be_valid }
+      end
+
+      context "when far in the past" do
+        let(:start_time) { Time.at(0) }
+
+        it { is_expected.not_to be_valid }
+      end
+    end
+  end
 end

spec/datadog/ci/test_visibility/transport_spec.rb

@@ -40,7 +40,8 @@
           expect(payload["version"]).to eq(1)
 
           metadata = payload["metadata"]["*"]
-          expect(metadata).to include("runtime-id", "language", "library_version")
+          expect(metadata).to include("runtime-id", "library_version")
+          expect(metadata["language"]).to eq("ruby")
 
           events = payload["events"]
           expect(events.count).to eq(1)
@@ -84,17 +85,22 @@
             events = payload["events"]
             expect(events.count).to eq(expected_events_count)
 
-            p events
             span_events = events.filter { |e| e["type"] == "span" }
             expect(span_events.count).to eq(1)
           end
         end
       end
     end
 
-    context "when there are no events" do
+    context "when all events are invalid" do
+      before do
+        produce_test_trace
+
+        span.start_time = Time.at(0)
+      end
+
       it "does not send anything" do
-        subject.send_traces([])
+        subject.send_traces(traces)
 
         expect(http).not_to have_received(:request)
       end

vendor/rbs/ddtrace/0/datadog/core/environment/identity.rbs

@@ -0,0 +1,10 @@
+module Datadog
+  module Core
+    module Environment
+      module Identity
+        def self.lang: () -> "ruby"
+        def self.id: () -> String
+      end
+    end
+  end
+end