VED-1006 Minor improvement to delta error handling (#1099)

Author: dlzhry2nhs

infrastructure/instance/delta.tf

@@ -93,9 +93,6 @@ data "aws_iam_policy_document" "delta_policy_document" {
     templatefile("${local.policy_path}/dynamo_key_access.json", {
       "dynamo_encryption_key" : data.aws_kms_key.existing_dynamo_encryption_key.arn
     }),
-    templatefile("${local.policy_path}/aws_sns_topic.json", {
-      "aws_sns_topic_name" : aws_sns_topic.delta_sns.name
-    }),
     templatefile("${local.policy_path}/log_kinesis.json", {
       "kinesis_stream_name" : module.splunk.firehose_stream_name
     }),
@@ -159,7 +156,7 @@ resource "aws_lambda_event_source_mapping" "delta_trigger" {
   starting_position = "TRIM_HORIZON"
   destination_config {
     on_failure {
-      destination_arn = aws_sns_topic.delta_sns.arn
+      destination_arn = aws_sqs_queue.dlq.arn
     }
   }
   maximum_retry_attempts = 0
@@ -170,10 +167,6 @@ resource "aws_sqs_queue" "dlq" {
   name = "${local.short_prefix}-${local.dlq_name}"
 }
 
-resource "aws_sns_topic" "delta_sns" {
-  name = "${local.short_prefix}-${local.sns_name}"
-}
-
 resource "aws_cloudwatch_log_group" "delta_lambda" {
   name              = "/aws/lambda/${local.short_prefix}-${local.function_name}"
   retention_in_days = 30

infrastructure/instance/outputs.tf

@@ -14,11 +14,7 @@ output "aws_sqs_queue_name" {
   value = aws_sqs_queue.dlq.name
 }
 
-output "aws_sns_topic_name" {
-  value = aws_sns_topic.delta_sns.name
-}
-
 output "id_sync_queue_arn" {
-  description = "The ARN of the created SQS queue"
+  description = "The ARN of the ID Sync (MNS NHS Number change) SQS queue"
   value       = aws_sqs_queue.id_sync_queue.arn
 }

lambdas/delta_backend/src/delta.py

@@ -37,12 +37,9 @@ def get_delta_table():
     return delta_table
 
 
-def send_message(record, queue_url=failure_queue_url):
-    # Create a message
-    message_body = record
+def send_record_to_dlq(record: dict) -> None:
     try:
-        # Send the record to the queue
-        get_sqs_client().send_message(QueueUrl=queue_url, MessageBody=json.dumps(message_body))
+        get_sqs_client().send_message(QueueUrl=failure_queue_url, MessageBody=json.dumps(record))
         logger.info("Record saved successfully to the DLQ")
     except Exception:
         logger.exception("Error sending record to DLQ")
@@ -207,35 +204,23 @@ def process_record(record):
         return False, {"statusCode": "500", "statusDesc": "Exception", "diagnostics": e}
 
 
-def handler(event, _context):
-    overall_success = True
+def handler(event, _context) -> bool:
     logger.info("Starting Delta Handler")
-    try:
-        for record in event["Records"]:
-            datetime_str = datetime.now().isoformat()
-            start = time.time()
-            success, operation_outcome = process_record(record)
-            overall_success = overall_success and success
-            end = time.time()
-            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 = {
-            "statusCode": "500",
-            "statusDesc": "Exception",
-            "diagnostics": "Delta Lambda failure: Incorrect invocation of Lambda",
+
+    for record in event["Records"]:
+        record_ingestion_datetime = datetime.now().isoformat()
+        record_processing_start = time.time()
+        success, operation_outcome = process_record(record)
+        record_processing_end = time.time()
+        log_data = {
+            "function_name": "delta_sync",
+            "operation_outcome": operation_outcome,
+            "date_time": record_ingestion_datetime,
+            "time_taken": f"{round(record_processing_end - record_processing_start, 5)}s",
         }
-        logger.exception(operation_outcome["diagnostics"])
-        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)
+        if not success:
+            send_record_to_dlq(record)
 
-    return overall_success
+    return True

lambdas/delta_backend/tests/test_delta.py

@@ -10,13 +10,13 @@
 from delta import (
     handler,
     process_record,
-    send_message,
+    send_record_to_dlq,
 )
 from mappings import ActionFlag, EventName, Operation
 from utils_for_converter_tests import RecordConfig, ValuesForTests
 
-TEST_QUEUE_URL = "https://sqs.eu-west-2.amazonaws.com/123456789012/test-queue"
-os.environ["AWS_SQS_QUEUE_URL"] = TEST_QUEUE_URL
+TEST_DEAD_LETTER_QUEUE_URL = "https://sqs.eu-west-2.amazonaws.com/123456789012/test-queue"
+os.environ["AWS_SQS_QUEUE_URL"] = TEST_DEAD_LETTER_QUEUE_URL
 os.environ["DELTA_TABLE_NAME"] = "my_delta_table"
 os.environ["DELTA_TTL_DAYS"] = "14"
 os.environ["SOURCE"] = "my_source"
@@ -52,27 +52,23 @@ def setUp(self):
         self.mock_delta_table = self.delta_table_patcher.start()
 
     def tearDown(self):
-        self.logger_exception_patcher.stop()
-        self.logger_warning_patcher.stop()
-        self.logger_error_patcher.stop()
-        self.logger_info_patcher.stop()
-        self.mock_send_log_to_firehose.stop()
-        self.sqs_client_patcher.stop()
-        self.delta_table_patcher.stop()
+        patch.stopall()
 
-    def test_send_message_success(self):
+    def test_send_record_to_dlq_success(self):
         # Arrange
         self.mock_sqs_client.send_message.return_value = {"MessageId": "123"}
         record = {"key": "value"}
-        sqs_queue_url = "test-queue-url"
 
         # Act
-        send_message(record, sqs_queue_url)
+        send_record_to_dlq(record)
 
         # Assert
-        self.mock_sqs_client.send_message.assert_called_once_with(QueueUrl=sqs_queue_url, MessageBody=json.dumps(record))
+        self.mock_sqs_client.send_message.assert_called_once_with(
+            QueueUrl=TEST_DEAD_LETTER_QUEUE_URL, MessageBody=json.dumps(record)
+        )
+        self.mock_logger_info.assert_called_with("Record saved successfully to the DLQ")
 
-    def test_send_message_client_error(self):
+    def test_send_record_to_dlq_client_error(self):
         # Arrange
         record = {"key": "value"}
 
@@ -81,7 +77,7 @@ def test_send_message_client_error(self):
         self.mock_sqs_client.send_message.side_effect = ClientError(error_response, "SendMessage")
 
         # Act
-        send_message(record, "test-queue-url")
+        send_record_to_dlq(record)
 
         # Assert
         self.mock_logger_exception.assert_called_once_with("Error sending record to DLQ")
@@ -116,33 +112,35 @@ def test_handler_success_insert(self):
 
     def test_handler_exception(self):
         """Ensure that sqs_client exceptions do not cause the lambda handler itself to raise an exception"""
-
         # Arrange
         self.mock_sqs_client.send_message.side_effect = Exception("SQS error")
-        event = {"invalid_format": True}
+        self.mock_delta_table.put_item.return_value = FAIL_RESPONSE
+        event = ValuesForTests.get_event()
 
         # Act
         result = handler(event, None)
 
         # Assert
-        self.assertFalse(result)
+        self.assertTrue(result)
         self.mock_logger_exception.assert_has_calls(
             [
-                call("Delta Lambda failure: Incorrect invocation of Lambda"),
                 call("Error sending record to DLQ"),
             ]
         )
 
-    def test_handler_overall_failure(self):
+    def test_handler_raises_exception_if_called_with_unexpected_event(self):
+        """Tests that when the Lambda is invoked with an unexpected event format i.e. no "Records" key, then an
+        exception will be raised. The DDB Stream configuration will then ensure that the event is forwarded to the DLQ.
+        Note: this would only ever happen if we misconfigured the Lambda or tested manually with a bad event."""
         # Arrange
         event = {"invalid_format": True}
 
         # Act
-        result = handler(event, None)
+        with self.assertRaises(KeyError):
+            handler(event, None)
 
         # Assert
-        self.assertFalse(result)
-        self.mock_sqs_client.send_message.assert_called_with(QueueUrl=TEST_QUEUE_URL, MessageBody=json.dumps(event))
+        self.mock_sqs_client.send_message.assert_not_called()
 
     def test_handler_processing_failure(self):
         # Arrange
@@ -153,8 +151,10 @@ def test_handler_processing_failure(self):
         result = handler(event, None)
 
         # Assert
-        self.assertFalse(result)
-        self.mock_sqs_client.send_message.assert_called_with(QueueUrl=TEST_QUEUE_URL, MessageBody=json.dumps(event))
+        self.assertTrue(result)
+        self.mock_sqs_client.send_message.assert_called_with(
+            QueueUrl=TEST_DEAD_LETTER_QUEUE_URL, MessageBody=json.dumps(event["Records"][0])
+        )
 
     def test_handler_success_update(self):
         # Arrange
@@ -429,10 +429,11 @@ def test_single_error_in_multi(self):
         result = handler(event, None)
 
         # Assert
-        self.assertFalse(result)
+        self.assertTrue(result)
         self.assertEqual(self.mock_delta_table.put_item.call_count, 3)
         self.assertEqual(self.mock_send_log_to_firehose.call_count, 3)
         self.assertEqual(self.mock_logger_error.call_count, 1)
+        self.assertEqual(self.mock_sqs_client.send_message.call_count, 1)
 
     def test_single_exception_in_multi(self):
         # Arrange
@@ -454,7 +455,8 @@ def test_single_exception_in_multi(self):
         result = handler(event, None)
 
         # Assert
-        self.assertFalse(result)
+        self.assertTrue(result)
+        self.assertEqual(self.mock_sqs_client.send_message.call_count, 1)
         self.assertEqual(self.mock_delta_table.put_item.call_count, len(records_config))
         self.assertEqual(self.mock_send_log_to_firehose.call_count, len(records_config))
 
@@ -497,12 +499,9 @@ def test_handler_calls_process_record_for_each_event(self, mock_send_log_to_fire
         self.assertTrue(result)
         self.assertEqual(mock_process_record.call_count, len(event["Records"]))
 
-    # TODO depedency injection needed here
     @patch("delta.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, {})
@@ -514,10 +513,12 @@ def test_handler_sends_all_to_firehose(self, mock_send_log_to_firehose, mock_pro
         result = handler(event, {})
 
         # Assert
-        self.assertFalse(result)
+        self.assertTrue(result)
         self.assertEqual(mock_process_record.call_count, len(event["Records"]))
         # check that all records were sent to firehose
         self.assertEqual(mock_send_log_to_firehose.call_count, len(event["Records"]))
+        # Only send the failed record to SQS DLQ
+        self.assertEqual(self.mock_sqs_client.send_message.call_count, 1)
 
 
 class DeltaRecordProcessorTestCase(unittest.TestCase):
@@ -677,39 +678,3 @@ def test_returns_none_on_exception(self, mock_get_dynamodb_table):
         table = delta.get_delta_table()
         self.assertIsNone(table)
         self.mock_logger_error.assert_called()
-
-
-class TestSendMessage(unittest.TestCase):
-    def setUp(self):
-        self.get_sqs_client_patcher = patch("delta.get_sqs_client")
-        self.mock_get_sqs_client = self.get_sqs_client_patcher.start()
-        self.mock_sqs_client = MagicMock()
-        self.mock_get_sqs_client.return_value = self.mock_sqs_client
-
-        self.logger_info_patcher = patch("logging.Logger.info")
-        self.mock_logger_info = self.logger_info_patcher.start()
-        self.logger_error_patcher = patch("logging.Logger.error")
-        self.mock_logger_error = self.logger_error_patcher.start()
-
-    def tearDown(self):
-        self.get_sqs_client_patcher.stop()
-        self.logger_info_patcher.stop()
-        self.logger_error_patcher.stop()
-
-    def test_send_message_success(self):
-        record = {"a": "bbb"}
-        self.mock_sqs_client.send_message.return_value = {"MessageId": "123"}
-
-        delta.send_message(record)
-
-        self.mock_sqs_client.send_message.assert_called_once()
-        self.mock_logger_info.assert_any_call("Record saved successfully to the DLQ")
-        self.mock_logger_error.assert_not_called()
-
-    def test_send_message_client_error(self):
-        record = {"a": "bbb"}
-        self.mock_sqs_client.send_message.side_effect = Exception("SQS error")
-
-        delta.send_message(record, "test-queue-url")
-
-        self.mock_logger_error.assert_called()