Use Data Streams events to send Kafka messages to Datadog (#22208)

* Use Data Streams events to send Kafka messages to Datadog

* simplify e2e test

Author: piochelepiotr

kafka_actions/changelog.d/22208.changed

@@ -0,0 +1 @@
+kafka_actions: Send Kafka messages to backend as data streams events instead of regular Datadog events

kafka_actions/datadog_checks/kafka_actions/check.py

@@ -144,6 +144,10 @@ def _emit_action_event(
     ):
         """Emit an event for action success or failure.
 
+        Events are sent in duplicate:
+        1. As a standard Datadog event
+        2. As a data-streams-message event
+
         Args:
             success: Whether the action succeeded
             action: Action name
@@ -155,8 +159,10 @@ def _emit_action_event(
         alert_type = 'success' if success else 'error'
 
         payload = {
+            'message_timestamp': int(time.time() * 1000),
             'action': action,
             'remote_config_id': self.remote_config_id,
+            'kafka_cluster_id': cluster,
             'status': 'success' if success else 'failure',
             'message': message,
         }
@@ -171,6 +177,7 @@ def _emit_action_event(
 
         tags = self._get_tags() + [f'kafka_cluster_id:{cluster}']
 
+        # Send as standard Datadog event (backwards compatibility; moving to Data Streams intake-only)
         event_payload = {
             'timestamp': int(time.time()),
             'event_type': f'kafka_action_{event_type}',
@@ -180,10 +187,15 @@ def _emit_action_event(
             'source_type_name': 'kafka',
             'aggregation_key': f'kafka_action_{action}_{self.remote_config_id}',
             'tags': tags,
+            'remote_config_id': self.remote_config_id,
+            'kafka_cluster_id': cluster,
         }
 
         self.event(event_payload)
 
+        # Send same payload to Data Streams track
+        self.event_platform_event(event_text, "data-streams-message")
+
     # =========================================================================
     # Action Handlers (RFC-Compliant)
     # =========================================================================
@@ -486,57 +498,32 @@ def _parse_literal(self, value_str: str):
         return value_str
 
     def _emit_message_event_deserialized(self, deserialized_msg: DeserializedMessage, cluster: str):
-        """Emit a deserialized Kafka message as a Datadog event.
+        """Emit a deserialized Kafka message to data-streams-message track.
 
         Args:
             deserialized_msg: DeserializedMessage object
             cluster: Kafka cluster identifier
         """
-        msg_dict = deserialized_msg.to_dict()
-
         event_data = {
+            'message_timestamp': int(time.time() * 1000),
+            'remote_config_id': self.remote_config_id,
+            'kafka_cluster_id': cluster,
             'topic': deserialized_msg.topic,
             'partition': deserialized_msg.partition,
             'offset': deserialized_msg.offset,
-            'key': msg_dict.get('key'),
-            'value': msg_dict.get('value'),
+            'key': deserialized_msg.key,
+            'value': deserialized_msg.value,
         }
 
-        if msg_dict.get('headers'):
-            event_data['headers'] = msg_dict['headers']
+        if deserialized_msg.headers:
+            event_data['headers'] = deserialized_msg.headers
 
         if deserialized_msg.value_schema_id:
             event_data['value_schema_id'] = deserialized_msg.value_schema_id
         if deserialized_msg.key_schema_id:
             event_data['key_schema_id'] = deserialized_msg.key_schema_id
 
-        event_text = json.dumps(event_data, indent=2)
-
-        event_tags = self._get_tags() + [
-            f'kafka_cluster_id:{cluster}',
-            f'topic:{deserialized_msg.topic}',
-            f'partition:{deserialized_msg.partition}',
-            f'offset:{deserialized_msg.offset}',
-        ]
-
-        event_title = f'Kafka Message: {deserialized_msg.topic}'
-
-        agg_key = (
-            f'kafka_{deserialized_msg.topic}_{deserialized_msg.partition}_'
-            f'{deserialized_msg.offset}_{self.remote_config_id}'
-        )
-
-        event_payload = {
-            'timestamp': int(time.time()),
-            'event_type': 'kafka_message',
-            'msg_title': event_title,
-            'msg_text': event_text,
-            'tags': event_tags,
-            'source_type_name': 'kafka',
-            'aggregation_key': agg_key,
-        }
-
-        self.event(event_payload)
+        self.event_platform_event(json.dumps(event_data), "data-streams-message")
 
     def _format_for_display(self, data) -> str:
         """Format data for display in event.

kafka_actions/tests/test_e2e.py

@@ -13,97 +13,38 @@
 
 
 def test_e2e_read_messages(dd_environment, kafka_instance, check, dd_run_check, aggregator):
-    """Test reading messages from Kafka."""
+    """Test end-to-end integration: check can connect to real Kafka and emit events correctly."""
     if not e2e_testing():
         pytest.skip("E2E tests require dd_environment fixture")
 
     # Verify cluster is available
     cluster_id = common.get_cluster_id()
     assert cluster_id is not None, "Kafka cluster is not available"
 
-    # Ensure the instance has the cluster ID set
     kafka_instance['read_messages']['cluster'] = cluster_id
 
     # Run the check
     check_instance = check(kafka_instance)
     dd_run_check(check_instance)
 
-    # Verify events were emitted
-    events = aggregator.events
-    assert len(events) > 0, f"Expected at least one event. Got {len(events)} events."
-
-    # Find the action success event
-    action_events = [e for e in events if 'kafka_action_' in e.get('event_type', '')]
+    # Verify standard Datadog event was emitted
+    action_events = [e for e in aggregator.events if 'kafka_action_' in e.get('event_type', '')]
     assert len(action_events) == 1, f"Expected 1 action event, got {len(action_events)}"
+    assert action_events[0]['event_type'] == 'kafka_action_success'
 
-    action_event = action_events[0]
-    assert action_event['event_type'] == 'kafka_action_success'
-    assert 'kafka_cluster_id' in [tag.split(':')[0] for tag in action_event['tags']]
-    assert 'remote_config_id:test-rc-id' in action_event['tags']
-
-    # Parse the msg_text as JSON
-    event_data = json.loads(action_event['msg_text'])
-    assert event_data['action'] == 'read_messages'
-    assert event_data['status'] == 'success'
-    assert 'stats' in event_data
-    stats = event_data['stats']
-    assert stats['messages_scanned'] > 0
-    assert stats['messages_sent'] > 0
+    # Verify events sent to data-streams-message track
+    data_streams_events = aggregator.get_event_platform_events("data-streams-message")
+    action_ds_events = [e for e in data_streams_events if 'action' in e]
+    message_events = [e for e in data_streams_events if 'topic' in e]
 
-    # Check that message events were emitted
-    message_events = [e for e in events if e.get('event_type') == 'kafka_message']
+    # Verify both action and message events were sent
+    assert len(action_ds_events) == 1, f"Expected 1 action event in data streams, got {len(action_ds_events)}"
     assert len(message_events) > 0, "Expected at least one Kafka message event"
-    assert len(message_events) == stats['messages_sent'], "Message event count should match stats"
-
-    # Verify message event structure
-    msg_event = message_events[0]
-    assert msg_event['source_type_name'] == 'kafka'
-    assert 'kafka_cluster_id' in [tag.split(':')[0] for tag in msg_event['tags']]
-    assert 'remote_config_id:test-rc-id' in msg_event['tags']
-
-    # Parse message event data
-    msg_data = json.loads(msg_event['msg_text'])
-    assert 'topic' in msg_data
-    assert 'partition' in msg_data
-    assert 'offset' in msg_data
-    assert 'key' in msg_data
-    assert 'value' in msg_data
-    assert msg_data['topic'] == 'test-topic'
 
-    # Find the action success event
-    action_events = [e for e in events if 'kafka_action_' in e.get('event_type', '')]
-    assert len(action_events) == 1, f"Expected 1 action event, got {len(action_events)}"
-
-    action_event = action_events[0]
-    assert action_event['event_type'] == 'kafka_action_success'
-    assert 'kafka_cluster_id' in [tag.split(':')[0] for tag in action_event['tags']]
-    assert 'remote_config_id:test-rc-id' in action_event['tags']
+    # Verify action event payload is consistent between both tracks
+    event_data = json.loads(action_events[0]['msg_text'])
+    assert action_ds_events[0] == event_data, "Action event should have same payload in both tracks"
 
-    # Parse the msg_text as JSON
-    event_data = json.loads(action_event['msg_text'])
-    assert event_data['action'] == 'read_messages'
-    assert event_data['status'] == 'success'
-    assert 'stats' in event_data
+    # Verify message count matches stats
     stats = event_data['stats']
-    assert stats['messages_scanned'] > 0
-    assert stats['messages_sent'] > 0
-
-    # Check that message events were emitted
-    message_events = [e for e in events if e.get('event_type') == 'kafka_message']
-    assert len(message_events) > 0, "Expected at least one Kafka message event"
     assert len(message_events) == stats['messages_sent'], "Message event count should match stats"
-
-    # Verify message event structure
-    msg_event = message_events[0]
-    assert msg_event['source_type_name'] == 'kafka'
-    assert 'kafka_cluster_id' in [tag.split(':')[0] for tag in msg_event['tags']]
-    assert 'remote_config_id:test-rc-id' in msg_event['tags']
-
-    # Parse message event data
-    msg_data = json.loads(msg_event['msg_text'])
-    assert 'topic' in msg_data
-    assert 'partition' in msg_data
-    assert 'offset' in msg_data
-    assert 'key' in msg_data
-    assert 'value' in msg_data
-    assert msg_data['topic'] == 'test-topic'

kafka_actions/tests/test_unit.py

@@ -103,18 +103,21 @@ def test_read_messages_with_filter(self, aggregator, dd_run_check):
             assert call_kwargs['start_offset'] == -1
             assert call_kwargs['max_messages'] == 100
 
-        events = [e for e in aggregator.events if e.get('event_type') == 'kafka_message']
-        assert len(events) == 2, f"Expected 2 message events (filtered), got {len(events)}"
+        all_events = aggregator.get_event_platform_events("data-streams-message")
+        # Filter for message events only (those with 'topic' field)
+        message_events = [e for e in all_events if 'topic' in e]
+        assert len(message_events) == 2, f"Expected 2 message events (filtered), got {len(message_events)}"
 
-        event1 = events[0]
-        assert 'test-topic' in event1['msg_title']
-        assert '"offset": 100' in event1['msg_text']
-        assert '"id": 1' in event1['msg_text']
-        assert '"status": "active"' in event1['msg_text']
+        event1 = message_events[0]
+        assert event1['topic'] == 'test-topic'
+        assert event1['offset'] == 100
+        assert event1['value']['id'] == 1
+        assert event1['value']['status'] == 'active'
+        assert event1['remote_config_id'] == 'test-read-messages-001'
 
-        event2 = events[1]
-        assert '"offset": 102' in event2['msg_text']
-        assert '"id": 3' in event2['msg_text']
+        event2 = message_events[1]
+        assert event2['offset'] == 102
+        assert event2['value']['id'] == 3
 
 
 class TestCreateTopicAction:
@@ -370,11 +373,14 @@ def test_read_messages_nested_field_filter(self, aggregator, dd_run_check):
 
             mock_consume.assert_called_once()
 
-        events = [e for e in aggregator.events if e.get('event_type') == 'kafka_message']
-        assert len(events) == 1, f"Expected 1 message event (filtered), got {len(events)}"
-        assert '"order_id": 1' in events[0]['msg_text']
-        assert '"country": "US"' in events[0]['msg_text']
-        assert '"tier": "gold"' in events[0]['msg_text']
+        all_events = aggregator.get_event_platform_events("data-streams-message")
+        # Filter for message events only (those with 'topic' field)
+        message_events = [e for e in all_events if 'topic' in e]
+        assert len(message_events) == 1, f"Expected 1 message event (filtered), got {len(message_events)}"
+        assert message_events[0]['value']['order_id'] == 1
+        assert message_events[0]['value']['user']['country'] == 'US'
+        assert message_events[0]['value']['user']['tier'] == 'gold'
+        assert message_events[0]['remote_config_id'] == 'test-nested-filter-001'
 
 
 if __name__ == '__main__':