[kafka_actions] Base64-encode binary headers and support header filtering (DataDog#23106)

* [kafka_actions] Base64-encode binary headers and support header filtering

Binary Kafka headers were displayed as unhelpful `<binary, N bytes>` placeholders.
Now they are base64-encoded with a `<base64>` prefix so the actual data is visible.

Header filtering is now supported in both filter systems:
- jq-style: `.headers.dbmOrgId == "2"`
- MessageFilter: `field: "header.dbmOrgId", operator: "eq", value: "2"`

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Remove unnecessary MessageFilter header filtering

MessageFilter is not used anywhere in the codebase. Only the jq-style
filter in check.py is used, so header filtering only needs to be there.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Use lazy dict for filter context to preserve partial deserialization

The filter context was eagerly accessing deserialized_msg.key and
deserialized_msg.value, defeating the lazy deserialization. Now uses
_LazyMessageDict which only triggers deserialization when the filter
actually accesses a field (e.g., filtering on .headers won't
deserialize key/value).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Add changelog entry for header improvements

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: piochelepiotr

kafka_actions/changelog.d/23106.added

@@ -0,0 +1 @@
+Base64-encode binary headers and support header filtering in jq-style filters

kafka_actions/datadog_checks/kafka_actions/check.py

@@ -13,6 +13,38 @@
 from .schema_registry import SchemaRegistryClient
 
 
+class _LazyMessageDict(dict):
+    """Dict-like wrapper around DeserializedMessage that only deserializes fields on access."""
+
+    _FIELD_MAP = {
+        'key': lambda m: m.key,
+        'value': lambda m: m.value,
+        'headers': lambda m: m.headers,
+        'topic': lambda m: m.topic,
+        'partition': lambda m: m.partition,
+        'offset': lambda m: m.offset,
+    }
+
+    def __init__(self, msg: DeserializedMessage):
+        super().__init__()
+        self._msg = msg
+
+    def __contains__(self, key):
+        return key in self._FIELD_MAP
+
+    def __getitem__(self, key):
+        getter = self._FIELD_MAP.get(key)
+        if getter is None:
+            raise KeyError(key)
+        return getter(self._msg)
+
+    def get(self, key, default=None):
+        getter = self._FIELD_MAP.get(key)
+        if getter is None:
+            return default
+        return getter(self._msg)
+
+
 class KafkaActionsCheck(AgentCheck):
     """
     Kafka Actions Check - Performs one-time actions on Kafka clusters.
@@ -337,18 +369,10 @@ def _evaluate_filter(self, filter_expression: str, deserialized_msg: Deserialize
             True if message matches filter, False otherwise
         """
         try:
-            msg_dict = {
-                'key': deserialized_msg.key,
-                'value': deserialized_msg.value,
-                'topic': deserialized_msg.topic,
-                'partition': deserialized_msg.partition,
-                'offset': deserialized_msg.offset,
-            }
-
-            # Simple jq expression parser
-            # Supports: .field.subfield == "value", .field > 100, .field contains "text"
-            # Supports: and, or operators
-            result = self._evaluate_jq_expression(filter_expression, msg_dict)
+            # Use a lazy dict so key/value are only deserialized when the filter accesses them
+            context = _LazyMessageDict(deserialized_msg)
+
+            result = self._evaluate_jq_expression(filter_expression, context)
             self.log.debug("Filter '%s' evaluated to: %s", filter_expression, result)
             return result
 

kafka_actions/datadog_checks/kafka_actions/message_deserializer.py

@@ -510,7 +510,7 @@ def headers(self) -> dict:
                 try:
                     headers[key] = value.decode('utf-8') if value else None
                 except UnicodeDecodeError:
-                    headers[key] = f"<binary, {len(value)} bytes>"
+                    headers[key] = f"<base64>{base64.b64encode(value).decode('ascii')}"
         return headers
 
     @staticmethod

kafka_actions/tests/test_message_deserializer.py

@@ -17,12 +17,13 @@
 class MockKafkaMessage:
     """Mock confluent_kafka.Message for testing."""
 
-    def __init__(self, key, value, topic='test-topic', partition=0, offset=0):
+    def __init__(self, key, value, topic='test-topic', partition=0, offset=0, headers=None):
         self._key = key
         self._value = value
         self._topic = topic
         self._partition = partition
         self._offset = offset
+        self._headers = headers
 
     def key(self):
         return self._key
@@ -43,7 +44,7 @@ def timestamp(self):
         return (1, 1732128000000)  # (timestamp_type, timestamp_ms)
 
     def headers(self):
-        return None
+        return self._headers
 
 
 class TestMessageDeserializer:
@@ -684,5 +685,67 @@ def test_deserialized_message_with_registry(self):
         assert msg.value_schema_id == 42
 
 
+class TestHeaderSerialization:
+    """Test header serialization in DeserializedMessage."""
+
+    def test_utf8_headers(self):
+        """Test that UTF-8 headers are decoded as strings."""
+        log = MagicMock()
+        deserializer = MessageDeserializer(log)
+        kafka_msg = MockKafkaMessage(
+            key=b'key',
+            value=b'{}',
+            headers=[('dbmOrgId', b'2'), ('traceparent', b'00-abc-def-00')],
+        )
+        config = {'key_format': 'string', 'value_format': 'json', 'value_uses_schema_registry': False}
+        msg = DeserializedMessage(kafka_msg, deserializer, config)
+        assert msg.headers == {'dbmOrgId': '2', 'traceparent': '00-abc-def-00'}
+
+    def test_binary_headers_base64_encoded(self):
+        """Test that binary headers are base64-encoded with prefix."""
+        log = MagicMock()
+        deserializer = MessageDeserializer(log)
+        binary_value = bytes([0x80, 0xFF, 0x00, 0x01, 0xDE, 0xAD, 0xBE, 0xEF])
+        kafka_msg = MockKafkaMessage(
+            key=b'key',
+            value=b'{}',
+            headers=[('koutrisngId', binary_value)],
+        )
+        config = {'key_format': 'string', 'value_format': 'json', 'value_uses_schema_registry': False}
+        msg = DeserializedMessage(kafka_msg, deserializer, config)
+        import base64
+
+        expected = f"<base64>{base64.b64encode(binary_value).decode('ascii')}"
+        assert msg.headers['koutrisngId'] == expected
+
+    def test_mixed_headers(self):
+        """Test mix of UTF-8 and binary headers."""
+        log = MagicMock()
+        deserializer = MessageDeserializer(log)
+        binary_value = bytes([0x80, 0xFF])
+        kafka_msg = MockKafkaMessage(
+            key=b'key',
+            value=b'{}',
+            headers=[('text', b'hello'), ('binary', binary_value)],
+        )
+        config = {'key_format': 'string', 'value_format': 'json', 'value_uses_schema_registry': False}
+        msg = DeserializedMessage(kafka_msg, deserializer, config)
+        assert msg.headers['text'] == 'hello'
+        assert msg.headers['binary'].startswith('<base64>')
+
+    def test_null_header_value(self):
+        """Test that null header values are preserved as None."""
+        log = MagicMock()
+        deserializer = MessageDeserializer(log)
+        kafka_msg = MockKafkaMessage(
+            key=b'key',
+            value=b'{}',
+            headers=[('empty', None)],
+        )
+        config = {'key_format': 'string', 'value_format': 'json', 'value_uses_schema_registry': False}
+        msg = DeserializedMessage(kafka_msg, deserializer, config)
+        assert msg.headers['empty'] is None
+
+
 if __name__ == '__main__':
     pytest.main([__file__, '-v'])

kafka_actions/tests/test_message_filtering.py

@@ -17,12 +17,13 @@
 class MockKafkaMessage:
     """Mock confluent_kafka.Message for testing."""
 
-    def __init__(self, key, value, topic='test-topic', partition=0, offset=0):
+    def __init__(self, key, value, topic='test-topic', partition=0, offset=0, headers=None):
         self._key = key
         self._value = value
         self._topic = topic
         self._partition = partition
         self._offset = offset
+        self._headers = headers
 
     def key(self):
         return self._key
@@ -43,7 +44,7 @@ def timestamp(self):
         return (1, 1732128000000)
 
     def headers(self):
-        return None
+        return self._headers
 
 
 class TestFilteringOperators:
@@ -534,6 +535,28 @@ def test_null_literal(self):
         msg = self.create_message({'optional_field': None})
         assert check._evaluate_filter('.value.optional_field == null', msg) is True
 
+    def test_jq_filter_on_header(self):
+        """Test jq-style .headers.key filter."""
+        instance = {
+            'remote_config_id': 'test',
+            'kafka_connect_str': 'localhost:9092',
+            'read_messages': {'cluster': 'test', 'topic': 'test'},
+        }
+        check = KafkaActionsCheck('kafka_actions', {}, [instance])
+
+        kafka_msg = MockKafkaMessage(
+            key=b'test-key',
+            value=json.dumps({'status': 'active'}).encode('utf-8'),
+            headers=[('dbmOrgId', b'2'), ('traceparent', b'00-abc-def-00')],
+        )
+        config = {'key_format': 'string', 'value_format': 'json', 'value_uses_schema_registry': False}
+        msg = DeserializedMessage(kafka_msg, self.deserializer, config)
+
+        assert check._evaluate_filter('.headers.dbmOrgId == "2"', msg) is True
+        assert check._evaluate_filter('.headers.dbmOrgId == "3"', msg) is False
+        assert check._evaluate_filter('.headers.traceparent contains "abc"', msg) is True
+        assert check._evaluate_filter('.headers.nonexistent == "x"', msg) is False
+
 
 if __name__ == '__main__':
     pytest.main([__file__, '-vv'])