implement endpoint_payload.events_count metric

lib/datadog/ci/test_optimisation/coverage/transport.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require_relative "event"
+require_relative "../../ext/telemetry"
 require_relative "../../transport/event_platform_transport"
 
 module Datadog
@@ -10,6 +11,10 @@ module Coverage
         class Transport < Datadog::CI::Transport::EventPlatformTransport
           private
 
+          def telemetry_endpoint_tag
+            Ext::Telemetry::Endpoint::CODE_COVERAGE
+          end
+
           def send_payload(encoded_payload)
             api.citestcov_request(
               path: Ext::Transport::TEST_COVERAGE_INTAKE_PATH,

lib/datadog/ci/test_visibility/transport.rb

@@ -3,6 +3,7 @@
 require "datadog/core/environment/identity"
 
 require_relative "serializers/factories/test_level"
+require_relative "../ext/telemetry"
 require_relative "../ext/transport"
 require_relative "../transport/event_platform_transport"
 
@@ -31,6 +32,10 @@ def send_traces(traces)
 
         private
 
+        def telemetry_endpoint_tag
+          Ext::Telemetry::Endpoint::TEST_CYCLE
+        end
+
         def send_payload(encoded_payload)
           api.citestcycle_request(
             path: Datadog::CI::Ext::Transport::TEST_VISIBILITY_INTAKE_PATH,

lib/datadog/ci/transport/event_platform_transport.rb

@@ -25,21 +25,23 @@ def send_events(events)
           return [] if events.nil? || events.empty?
 
           Datadog.logger.debug { "[#{self.class.name}] Sending #{events.count} events..." }
-          Telemetry.events_enqueued_for_serialization(events.count)
 
           encoded_events = encode_events(events)
           if encoded_events.empty?
             Datadog.logger.debug { "[#{self.class.name}] Empty encoded events list, skipping send" }
             return []
           end
 
+          Telemetry.events_enqueued_for_serialization(encoded_events.count)
+
           responses = []
 
           Datadog::Core::Chunker.chunk_by_size(encoded_events, max_payload_size).map do |chunk|
             encoded_payload = pack_events(chunk)
             Datadog.logger.debug do
               "[#{self.class.name}] Send chunk of #{chunk.count} events; payload size #{encoded_payload.size}"
             end
+            Telemetry.endpoint_payload_events_count(chunk.count, telemetry_endpoint_tag)
 
             response = send_payload(encoded_payload)
 
@@ -51,6 +53,10 @@ def send_events(events)
 
         private
 
+        def telemetry_endpoint_tag
+          raise NotImplementedError, "must be implemented by the subclass"
+        end
+
         def encoder
           Datadog::Core::Encoding::MsgpackEncoder
         end

lib/datadog/ci/transport/telemetry.rb

@@ -10,6 +10,14 @@ module Telemetry
         def self.events_enqueued_for_serialization(count)
           Utils::Telemetry.inc(Ext::Telemetry::METRIC_EVENTS_ENQUEUED, count)
         end
+
+        def self.endpoint_payload_events_count(count, endpoint)
+          Utils::Telemetry.distribution(
+            Ext::Telemetry::METRIC_ENDPOINT_PAYLOAD_EVENTS_COUNT,
+            count.to_f,
+            {Ext::Telemetry::TAG_ENDPOINT => endpoint}
+          )
+        end
       end
     end
   end

sig/datadog/ci/transport/event_platform_transport.rbs

@@ -16,6 +16,8 @@ module Datadog
 
         private
 
+        def telemetry_endpoint_tag: () -> String
+
         def send_payload: (String payload) -> ::Datadog::CI::Transport::HTTP::ResponseDecorator
 
         def encoder: () -> singleton(Datadog::Core::Encoding::MsgpackEncoder)

sig/datadog/ci/transport/telemetry.rbs

@@ -3,6 +3,8 @@ module Datadog
     module Transport
       module Telemetry
         def self.events_enqueued_for_serialization: (Integer count) -> void
+
+        def self.endpoint_payload_events_count: (Integer count, String endpoint) -> void
       end
     end
   end

spec/datadog/ci/test_optimisation/coverage/transport_spec.rb

@@ -49,6 +49,15 @@
       end
 
       it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 1
+      it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 1
+
+      it "tags event with code_coverage endpoint" do
+        subject
+
+        expect(telemetry_metric(:distribution, "endpoint_payload.events_count")).to(
+          have_attributes(tags: {"endpoint" => "code_coverage"})
+        )
+      end
     end
 
     context "multiple events" do
@@ -84,6 +93,7 @@
       end
 
       it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 2
+      it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 2
 
       context "when some events are invalid" do
         let(:events) do
@@ -118,18 +128,24 @@
             "coverage={\"file.rb\"=>true, \"file2.rb\"=>true}]"
           )
         end
+
+        it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 1
+        it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 1
       end
 
       context "when chunking is used" do
         # one coverage event is approximately 75 bytes
         let(:max_payload_size) { 100 }
 
-        it "filters out invalid events" do
+        it "splits events based on size" do
           responses = subject
 
           expect(api).to have_received(:citestcov_request).twice
           expect(responses.count).to eq(2)
         end
+
+        it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 2
+        it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 1
       end
 
       context "when max_payload-size is too small" do

spec/datadog/ci/test_visibility/transport_spec.rb

@@ -63,6 +63,15 @@
       end
 
       it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 1
+      it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 1
+
+      it "tags event with test_cycle endpoint" do
+        subject
+
+        expect(telemetry_metric(:distribution, "endpoint_payload.events_count")).to(
+          have_attributes(tags: {"endpoint" => "test_cycle"})
+        )
+      end
     end
 
     context "with dd_env defined" do
@@ -127,8 +136,8 @@
         end
       end
 
-      # telemetry reports the number of traces
-      it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 2
+      it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 4
+      it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 4
 
       context "when some spans are broken" do
         let(:expected_events_count) { 3 }
@@ -161,6 +170,9 @@
             "Errors: {\"start\"=>#<Set: {\"must be greater than or equal to 946684800000000000\"}>}"
           )
         end
+
+        it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 3
+        it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 3
       end
 
       context "when chunking is used" do
@@ -174,6 +186,9 @@
           expect(api).to have_received(:citestcycle_request).twice
           expect(responses.count).to eq(2)
         end
+
+        it_behaves_like "emits telemetry metric", :inc, "events_enqueued_for_serialization", 4
+        it_behaves_like "emits telemetry metric", :distribution, "endpoint_payload.events_count", 3
       end
 
       context "when max_payload-size is too small" do

spec/datadog/ci/transport/telemetry_spec.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require_relative "../../../../lib/datadog/ci/transport/telemetry"
+
 RSpec.describe Datadog::CI::Transport::Telemetry do
   describe ".events_enqueued_for_serialization" do
     subject(:events_enqueued_for_serialization) { described_class.events_enqueued_for_serialization(count) }
@@ -12,4 +14,21 @@
       events_enqueued_for_serialization
     end
   end
+
+  describe ".endpoint_payload_events_count" do
+    subject(:endpoint_payload_events_count) { described_class.endpoint_payload_events_count(count, endpoint) }
+
+    let(:count) { 1 }
+    let(:endpoint) { "citestcycle" }
+
+    it "tracks the endpoint payload events count distribution" do
+      expect(Datadog::CI::Utils::Telemetry).to receive(:distribution).with(
+        Datadog::CI::Ext::Telemetry::METRIC_ENDPOINT_PAYLOAD_EVENTS_COUNT,
+        count.to_f,
+        {Datadog::CI::Ext::Telemetry::TAG_ENDPOINT => endpoint}
+      )
+
+      endpoint_payload_events_count
+    end
+  end
 end

spec/support/contexts/telemetry_spy.rb

@@ -11,15 +11,16 @@ def telemetry_spy_value_suffix(value)
 # spy on telemetry metrics emitted
 RSpec.shared_context "Telemetry spy" do
   before do
-    @metrics = {}
+    @metrics = {
+      inc: [],
+      distribution: []
+    }
 
     allow(Datadog::CI::Utils::Telemetry).to receive(:inc) do |metric_name, count, tags|
-      @metrics[:inc] ||= []
       @metrics[:inc] << SpiedMetric.new(metric_name, count, tags)
     end
 
     allow(Datadog::CI::Utils::Telemetry).to receive(:distribution) do |metric_name, value, tags|
-      @metrics[:distribution] ||= []
       @metrics[:distribution] << SpiedMetric.new(metric_name, value, tags)
     end
   end