log_firehose

Author: JamesW1-NHS

lambdas/delta_backend/src/delta.py

@@ -1,6 +1,5 @@
 import decimal
 import json
-import logging
 import os
 import time
 from datetime import UTC, datetime, timedelta
@@ -9,19 +8,16 @@
 from boto3.dynamodb.conditions import Attr
 from botocore.exceptions import ClientError
 
+from common.clients import STREAM_NAME, logger
+from common.log_firehose import send_log_to_firehose
 from common.mappings import ActionFlag, EventName, Operation
 from converter import Converter
-from log_firehose import FirehoseLogger
 
 failure_queue_url = os.environ["AWS_SQS_QUEUE_URL"]
 delta_table_name = os.environ["DELTA_TABLE_NAME"]
 delta_source = os.environ["SOURCE"]
 delta_ttl_days = os.environ["DELTA_TTL_DAYS"]
 region_name = "eu-west-2"
-logging.basicConfig()
-logger = logging.getLogger()
-logger.setLevel("INFO")
-firehose_logger = FirehoseLogger()
 
 delta_table = None
 
@@ -87,14 +83,6 @@ def get_creation_and_expiry_times(creation_timestamp: float) -> (str, int):
     return creation_datetime.isoformat(), expiry_timestamp
 
 
-def send_firehose(log_data):
-    try:
-        firehose_log = {"event": log_data}
-        firehose_logger.send_log(firehose_log)
-    except Exception:
-        logger.exception("Error sending log to Firehose")
-
-
 def handle_dynamodb_response(response, error_records):
     match response:
         case {"ResponseMetadata": {"HTTPStatusCode": 200}} if error_records:
@@ -248,14 +236,13 @@ def handler(event, _context):
             success, operation_outcome = process_record(record)
             overall_success = overall_success and success
             end = time.time()
-            send_firehose(
-                {
-                    "function_name": "delta_sync",
-                    "operation_outcome": operation_outcome,
-                    "date_time": datetime_str,
-                    "time_taken": f"{round(end - start, 5)}s",
-                }
-            )
+            log_data = {
+                "function_name": "delta_sync",
+                "operation_outcome": operation_outcome,
+                "date_time": datetime_str,
+                "time_taken": f"{round(end - start, 5)}s",
+            }
+            send_log_to_firehose(STREAM_NAME, log_data)
     except Exception:
         overall_success = False
         operation_outcome = {
@@ -265,7 +252,8 @@ def handler(event, _context):
         }
         logger.exception(operation_outcome["diagnostics"])
         send_message(event)  # Send failed records to DLQ
-        send_firehose({"function_name": "delta_sync", "operation_outcome": operation_outcome})
+        log_data = {"function_name": "delta_sync", "operation_outcome": operation_outcome}
+        send_log_to_firehose(STREAM_NAME, log_data)
 
     if not overall_success:
         send_message(event)

lambdas/delta_backend/src/log_firehose.py

@@ -75,13 +75,13 @@ def setUp(self):
         self.logger_exception_patcher = patch("logging.Logger.exception")
         self.mock_logger_exception = self.logger_exception_patcher.start()
 
-        self.firehose_logger_patcher = patch("delta.firehose_logger")
-        self.mock_firehose_logger = self.firehose_logger_patcher.start()
+        self.send_log_to_firehose_patcher = patch("delta.send_log_to_firehose")
+        self.mock_send_log_to_firehose = self.send_log_to_firehose_patcher.start()
 
     def tearDown(self):
         self.logger_exception_patcher.stop()
         self.logger_info_patcher.stop()
-        self.mock_firehose_logger.stop()
+        self.mock_send_log_to_firehose.stop()
 
         self.mock.stop()
 

lambdas/delta_backend/tests/test_convert.py

@@ -42,8 +42,8 @@ def setUp(self):
         self.logger_error_patcher = patch("logging.Logger.error")
         self.mock_logger_error = self.logger_error_patcher.start()
 
-        self.firehose_logger_patcher = patch("delta.firehose_logger")
-        self.mock_firehose_logger = self.firehose_logger_patcher.start()
+        self.send_log_to_firehose_patcher = patch("delta.send_log_to_firehose")
+        self.mock_send_log_to_firehose = self.send_log_to_firehose_patcher.start()
 
         self.sqs_client_patcher = patch("delta.sqs_client")
         self.mock_sqs_client = self.sqs_client_patcher.start()
@@ -56,7 +56,7 @@ def tearDown(self):
         self.logger_warning_patcher.stop()
         self.logger_error_patcher.stop()
         self.logger_info_patcher.stop()
-        self.mock_firehose_logger.stop()
+        self.mock_send_log_to_firehose.stop()
         self.sqs_client_patcher.stop()
         self.delta_table_patcher.stop()
 
@@ -105,7 +105,7 @@ def test_handler_success_insert(self):
             # Assert
             self.assertTrue(result)
             self.mock_delta_table.put_item.assert_called()
-            self.mock_firehose_logger.send_log.assert_called()  # check logged
+            self.mock_send_log_to_firehose.assert_called()  # check logged
             put_item_call_args = self.mock_delta_table.put_item.call_args  # check data written to DynamoDB
             put_item_data = put_item_call_args.kwargs["Item"]
             self.assertIn("Imms", put_item_data)
@@ -149,7 +149,7 @@ def test_handler_success_update(self):
         # Assert
         self.assertTrue(result)
         self.mock_delta_table.put_item.assert_called()
-        self.mock_firehose_logger.send_log.assert_called()  # check logged
+        self.mock_send_log_to_firehose.assert_called()  # check logged
         put_item_call_args = self.mock_delta_table.put_item.call_args  # check data written to DynamoDB
         put_item_data = put_item_call_args.kwargs["Item"]
         self.assertIn("Imms", put_item_data)
@@ -174,7 +174,7 @@ def test_handler_success_delete_physical(self):
         # Assert
         self.assertTrue(result)
         self.mock_delta_table.put_item.assert_called()
-        self.mock_firehose_logger.send_log.assert_called()  # check logged
+        self.mock_send_log_to_firehose.assert_called()  # check logged
         put_item_call_args = self.mock_delta_table.put_item.call_args  # check data written to DynamoDB
         put_item_data = put_item_call_args.kwargs["Item"]
         self.assertIn("Imms", put_item_data)
@@ -198,7 +198,7 @@ def test_handler_success_delete_logical(self):
         # Assert
         self.assertTrue(result)
         self.mock_delta_table.put_item.assert_called()
-        self.mock_firehose_logger.send_log.assert_called()  # check logged
+        self.mock_send_log_to_firehose.assert_called()  # check logged
         put_item_call_args = self.mock_delta_table.put_item.call_args  # check data written to DynamoDB
         put_item_data = put_item_call_args.kwargs["Item"]
         self.assertIn("Imms", put_item_data)
@@ -217,7 +217,7 @@ def test_dps_record_skipped(self, mock_logger_info):
 
         # Check logging and Firehose were called
         mock_logger_info.assert_called_with("Record from DPS skipped")
-        self.mock_firehose_logger.send_log.assert_called()
+        self.mock_send_log_to_firehose.assert_called()
         self.mock_sqs_client.send_message.assert_not_called()
 
     @patch("delta.Converter")
@@ -237,15 +237,15 @@ def test_partial_success_with_errors(self, mock_converter):
         self.assertTrue(response)
         # Check logging and Firehose were called
         self.mock_logger_info.assert_called()
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, 1)
-        self.mock_firehose_logger.send_log.assert_called_once()
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, 1)
+        self.mock_send_log_to_firehose.assert_called_once()
 
-        # Get the actual argument passed to send_log
-        args, kwargs = self.mock_firehose_logger.send_log.call_args
-        sent_payload = args[0]  # First positional arg
+        # Get the actual argument passed to send_log_to_firehose
+        args, kwargs = self.mock_send_log_to_firehose.call_args
+        sent_payload = args[1]  # Second positional arg
 
         # Navigate to the specific message
-        status_desc = sent_payload["event"]["operation_outcome"]["statusDesc"]
+        status_desc = sent_payload["operation_outcome"]["statusDesc"]
 
         # Assert the expected message is present
         self.assertIn(
@@ -275,7 +275,7 @@ def test_send_message_multi_records_diverse(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, len(records_config))
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, len(records_config))
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, len(records_config))
 
     def test_send_message_skipped_records_diverse(self):
         """Check skipped records sent to firehose but not to DynamoDB"""
@@ -301,7 +301,7 @@ def test_send_message_skipped_records_diverse(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, len(records_config))
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, len(records_config))
 
     def test_send_message_multi_create(self):
         # Arrange
@@ -319,7 +319,7 @@ def test_send_message_multi_create(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, 3)
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, 3)
 
     def test_send_message_multi_update(self):
         # Arrange
@@ -337,7 +337,7 @@ def test_send_message_multi_update(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, 3)
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, 3)
 
     def test_send_message_multi_logical_delete(self):
         # Arrange
@@ -371,7 +371,7 @@ def test_send_message_multi_logical_delete(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, 3)
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, 3)
 
     def test_send_message_multi_physical_delete(self):
         # Arrange
@@ -389,7 +389,7 @@ def test_send_message_multi_physical_delete(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, 3)
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, 3)
 
     def test_single_error_in_multi(self):
         # Arrange
@@ -412,7 +412,7 @@ def test_single_error_in_multi(self):
         # Assert
         self.assertFalse(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, 3)
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, 3)
         self.assertEqual(self.mock_logger_error.call_count, 1)
 
     def test_single_exception_in_multi(self):
@@ -437,7 +437,7 @@ def test_single_exception_in_multi(self):
         # Assert
         self.assertFalse(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, len(records_config))
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, len(records_config))
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, len(records_config))
 
     def test_single_duplicate_in_multi(self):
         # Arrange
@@ -460,16 +460,16 @@ def test_single_duplicate_in_multi(self):
         # Assert
         self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, len(records_config))
-        self.assertEqual(self.mock_firehose_logger.send_log.call_count, len(records_config))
+        self.assertEqual(self.mock_send_log_to_firehose.call_count, len(records_config))
 
     @patch("delta.process_record")
-    @patch("delta.send_firehose")
-    def test_handler_calls_process_record_for_each_event(self, mock_send_firehose, mock_process_record):
+    @patch("delta.send_log_to_firehose")
+    def test_handler_calls_process_record_for_each_event(self, mock_send_log_to_firehose, mock_process_record):
         # Arrange
         event = {"Records": [{"a": "record1"}, {"a": "record2"}, {"a": "record3"}]}
         # Mock process_record to always return True
         mock_process_record.return_value = True, {}
-        mock_send_firehose.return_value = None
+        mock_send_log_to_firehose.return_value = None
 
         # Act
         result = handler(event, {})
@@ -480,15 +480,15 @@ def test_handler_calls_process_record_for_each_event(self, mock_send_firehose, m
 
     # TODO depedency injection needed here
     @patch("delta.process_record")
-    @patch("delta.send_firehose")
-    def test_handler_sends_all_to_firehose(self, mock_send_firehose, mock_process_record):
+    @patch("delta.send_log_to_firehose")
+    def test_handler_sends_all_to_firehose(self, mock_send_log_to_firehose, mock_process_record):
         # Arrange
 
         # event with 3 records
         event = {"Records": [{"a": "record1"}, {"a": "record2"}, {"a": "record3"}]}
         return_ok = (True, {})
         return_fail = (False, {})
-        mock_send_firehose.return_value = None
+        mock_send_log_to_firehose.return_value = None
         mock_process_record.side_effect = [return_ok, return_fail, return_ok]
 
         # Act
@@ -498,7 +498,7 @@ def test_handler_sends_all_to_firehose(self, mock_send_firehose, mock_process_re
         self.assertFalse(result)
         self.assertEqual(mock_process_record.call_count, len(event["Records"]))
         # check that all records were sent to firehose
-        self.assertEqual(mock_send_firehose.call_count, len(event["Records"]))
+        self.assertEqual(mock_send_log_to_firehose.call_count, len(event["Records"]))
 
 
 class DeltaRecordProcessorTestCase(unittest.TestCase):