fix: store failed log events on intake errors (#1131)

Author: patbos

aws/logs_monitoring/logs/datadog_http_client.py

@@ -6,7 +6,6 @@
 
 import logging
 import os
-from concurrent.futures import as_completed
 
 from requests_futures.sessions import FuturesSession
 
@@ -67,7 +66,6 @@ def __init__(
         self._timeout = timeout
         self._session = None
         self._ssl_validation = not skip_ssl_validation
-        self._futures = []
 
         if logger.isEnabledFor(logging.DEBUG):
             logger.debug(
@@ -81,13 +79,6 @@ def _connect(self):
         self._session.headers.update(self._HEADERS)
 
     def _close(self):
-        # Resolve all the futures and log exceptions if any
-        for future in as_completed(self._futures):
-            try:
-                future.result()
-            except Exception as e:
-                logger.error(f"Exception while forwarding logs: {e}")
-
         self._session.close()
 
     def send(self, logs):
@@ -101,11 +92,11 @@ def send(self, logs):
         if DD_USE_COMPRESSION:
             data = compress_logs(data, DD_COMPRESSION_LEVEL)
 
-        # FuturesSession returns immediately with a future object
-        future = self._session.post(
+        # Resolve the future here so callers can attribute failures to this batch.
+        response = self._session.post(
             self._url, data, timeout=self._timeout, verify=self._ssl_validation
-        )
-        self._futures.append(future)
+        ).result()
+        response.raise_for_status()
 
     def __enter__(self):
         self._connect()

aws/logs_monitoring/tests/test_datadog_http_client.py

@@ -0,0 +1,90 @@
+import unittest
+import sys
+import types
+from unittest.mock import MagicMock, patch
+
+botocore = types.ModuleType("botocore")
+botocore.config = types.ModuleType("botocore.config")
+botocore.config.Config = MagicMock()
+sys.modules["boto3"] = MagicMock()
+sys.modules["botocore"] = botocore
+sys.modules["botocore.config"] = botocore.config
+sys.modules["requests"] = MagicMock()
+sys.modules["requests_futures.sessions"] = MagicMock()
+
+
+class TestDatadogHTTPClient(unittest.TestCase):
+    def _client(self, session):
+        from logs.datadog_http_client import DatadogHTTPClient
+
+        scrubber = MagicMock()
+        scrubber.scrub.side_effect = lambda payload: payload
+
+        client = DatadogHTTPClient("example.com", 443, False, False, "apikey", scrubber)
+        client._session = session
+        return client
+
+    def test_send_raises_future_exception(self):
+        future = MagicMock()
+        future.result.side_effect = Exception("network error")
+        session = MagicMock()
+        session.post.return_value = future
+
+        with self.assertRaisesRegex(Exception, "network error"):
+            self._client(session).send(['{"message":"hello"}'])
+
+    def test_send_raises_http_error_response(self):
+        response = MagicMock()
+        response.raise_for_status.side_effect = Exception("403 Client Error")
+        future = MagicMock()
+        future.result.return_value = response
+        session = MagicMock()
+        session.post.return_value = future
+
+        with self.assertRaisesRegex(Exception, "403 Client Error"):
+            self._client(session).send(['{"message":"hello"}'])
+
+    def test_send_returns_after_successful_response(self):
+        response = MagicMock()
+        future = MagicMock()
+        future.result.return_value = response
+        session = MagicMock()
+        session.post.return_value = future
+
+        self._client(session).send(['{"message":"hello"}'])
+
+        response.raise_for_status.assert_called_once_with()
+
+
+class TestForwarderFailedLogs(unittest.TestCase):
+    @patch("forwarder.send_event_metric")
+    @patch("forwarder.DatadogHTTPClient")
+    @patch("forwarder.DD_STORE_FAILED_EVENTS", True)
+    def test_forward_logs_stores_failed_batch(self, mock_http_client, mock_send_metric):
+        from forwarder import Forwarder
+        from retry.enums import RetryPrefix
+
+        client = MagicMock()
+        client.__enter__.return_value = client
+        client.send.side_effect = Exception("send failed")
+        mock_http_client.return_value = client
+
+        forwarder = Forwarder.__new__(Forwarder)
+        forwarder.storage = MagicMock()
+        forwarder._scrubber = MagicMock()
+        forwarder._matcher = MagicMock()
+        forwarder._matcher.match.return_value = True
+        forwarder._batcher = MagicMock()
+        forwarder._batcher.batch.return_value = [['"hello"']]
+
+        forwarder._forward_logs(["hello"])
+
+        forwarder.storage.store_data.assert_called_once_with(
+            RetryPrefix.LOGS, ['"hello"']
+        )
+        mock_send_metric.assert_any_call("logs_failed", ['"hello"'])
+        mock_send_metric.assert_any_call("logs_forwarded", 0)
+
+
+if __name__ == "__main__":
+    unittest.main()