[SDTEST-409] Send telemetry events in batches

lib/datadog/core/telemetry/event.rb

@@ -313,6 +313,28 @@ def type
             'distributions'
           end
         end
+
+        # Telemetry class for the 'message-batch' event
+        class MessageBatch
+          attr_reader :events
+
+          def type
+            'message-batch'
+          end
+
+          def initialize(events)
+            @events = events
+          end
+
+          def payload(seq_id)
+            @events.map do |event|
+              {
+                request_type: event.type,
+                payload: event.payload(seq_id),
+              }
+            end
+          end
+        end
       end
     end
   end

lib/datadog/core/telemetry/worker.rb

@@ -82,19 +82,17 @@ def perform(*events)
         end
 
         def flush_events(events)
-          return if events.nil?
+          return if events.nil? || events.empty?
           return if !enabled? || !sent_started_event?
 
           Datadog.logger.debug { "Sending #{events&.count} telemetry events" }
-          events.each do |event|
-            send_event(event)
-          end
+          send_event(Event::MessageBatch.new(events))
         end
 
         def heartbeat!
           return if !enabled? || !sent_started_event?
 
-          send_event(Event::AppHeartbeat.new)
+          enqueue(Event::AppHeartbeat.new)
         end
 
         def started!

sig/datadog/core/telemetry/event.rbs

@@ -3,7 +3,7 @@ module Datadog
     module Telemetry
       class Event
         class Base
-          def payload: (int seq_id) -> Hash[Symbol, untyped]
+          def payload: (int seq_id) -> (Hash[Symbol, untyped] | Array[Hash[Symbol, untyped]])
           def type: -> String?
         end
 
@@ -65,6 +65,13 @@ module Datadog
 
         class Distributions < GenerateMetrics
         end
+
+        class MessageBatch < Base
+          attr_reader events: Array[Datadog::Core::Telemetry::Event::Base]
+          @events: Array[Datadog::Core::Telemetry::Event::Base]
+
+          def initialize: (Array[Datadog::Core::Telemetry::Event::Base] events) -> void
+        end
       end
     end
   end

spec/datadog/core/telemetry/event_spec.rb

@@ -252,4 +252,25 @@ def contain_configuration(*array)
       )
     end
   end
+
+  context 'MessageBatch' do
+    let(:event) { described_class::MessageBatch.new(events) }
+
+    let(:events) { [described_class::AppClosing.new, described_class::AppHeartbeat.new] }
+
+    it do
+      is_expected.to eq(
+        [
+          {
+            request_type: 'app-closing',
+            payload: {}
+          },
+          {
+            request_type: 'app-heartbeat',
+            payload: {}
+          }
+        ]
+      )
+    end
+  end
 end

spec/datadog/core/telemetry/worker_spec.rb

@@ -40,8 +40,10 @@
       response
     end
 
-    allow(emitter).to receive(:request).with(an_instance_of(Datadog::Core::Telemetry::Event::AppHeartbeat)) do
-      @received_heartbeat = true
+    allow(emitter).to receive(:request).with(an_instance_of(Datadog::Core::Telemetry::Event::MessageBatch)) do |event|
+      subevent = event.events.first
+
+      @received_heartbeat = true if subevent.is_a?(Datadog::Core::Telemetry::Event::AppHeartbeat)
 
       response
     end
@@ -106,7 +108,7 @@
 
         it 'always sends heartbeat event after started event' do
           sent_hearbeat = false
-          allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::AppHeartbeat)) do
+          allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::MessageBatch)) do
             # app-started was already sent by now
             expect(worker.sent_started_event?).to be(true)
 
@@ -138,7 +140,7 @@
             end
 
             sent_hearbeat = false
-            allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::AppHeartbeat)) do
+            allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::MessageBatch)) do
               # app-started was already sent by now
               expect(@received_started).to be(true)
 
@@ -222,15 +224,20 @@
       context 'several workers running' do
         it 'sends single started event' do
           started_events = 0
+          mutex = Mutex.new
           allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::AppStarted)) do
             started_events += 1
 
             response
           end
 
           heartbeat_events = 0
-          allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::AppHeartbeat)) do
-            heartbeat_events += 1
+          allow(emitter).to receive(:request).with(kind_of(Datadog::Core::Telemetry::Event::MessageBatch)) do |event|
+            event.events.each do |subevent|
+              mutex.synchronize do
+                heartbeat_events += 1 if subevent.is_a?(Datadog::Core::Telemetry::Event::AppHeartbeat)
+              end
+            end
 
             response
           end
@@ -284,10 +291,15 @@
   describe '#enqueue' do
     it 'adds events to the buffer and flushes them later' do
       events_received = 0
+      mutex = Mutex.new
       allow(emitter).to receive(:request).with(
-        an_instance_of(Datadog::Core::Telemetry::Event::AppIntegrationsChange)
-      ) do
-        events_received += 1
+        an_instance_of(Datadog::Core::Telemetry::Event::MessageBatch)
+      ) do |event|
+        event.events.each do |subevent|
+          mutex.synchronize do
+            events_received += 1 if subevent.is_a?(Datadog::Core::Telemetry::Event::AppIntegrationsChange)
+          end
+        end
 
         response
       end