VED-714 rearchitect to prevent batch race conditions (#770)

Author: dlzhry2nhs

.github/workflows/sonarcloud.yml

@@ -46,9 +46,22 @@ jobs:
           poetry run coverage run -m unittest discover || echo "filenameprocessor tests failed" >> ../failed_tests.txt
           poetry run coverage xml -o ../filenameprocessor-coverage.xml
 
+      - name: Run unittest with batchprocessorfilter-coverage
+        working-directory: batch_processor_filter
+        id: batchprocessorfilter
+        env:
+          PYTHONPATH: ${{ github.workspace }}/batch_processor_filter/src:${{ github.workspace }}/batch_processor_filter/tests
+        continue-on-error: true
+        run: |
+          poetry install
+          poetry run coverage run -m unittest discover || echo "batchprocessorfilter tests failed" >> ../failed_tests.txt
+          poetry run coverage xml -o ../batchprocessorfilter-coverage.xml
+
       - name: Run unittest with recordprocessor-coverage
         working-directory: recordprocessor
         id: recordprocessor
+        env:
+          PYTHONPATH: ${{ github.workspace }}/recordprocessor/src:${{ github.workspace }}/recordprocessor/tests
         continue-on-error: true
         run: |
           poetry install

Makefile

@@ -1,6 +1,6 @@
 SHELL=/usr/bin/env bash -euo pipefail
 
-PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS = ack_backend backend delta_backend filenameprocessor mesh_processor recordprocessor redis_sync lambdas/id_sync lambdas/shared mns_subscription
+PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS = ack_backend backend batch_processor_filter delta_backend filenameprocessor mesh_processor recordprocessor redis_sync lambdas/id_sync lambdas/shared mns_subscription
 PYTHON_PROJECT_DIRS = e2e e2e_batch $(PYTHON_PROJECT_DIRS_WITH_UNIT_TESTS)
 
 #Installs dependencies using poetry.

ack_backend/src/audit_table.py

@@ -1,27 +1,8 @@
 """Add the filename to the audit table and check for duplicates."""
 
-from typing import Union
-from boto3.dynamodb.conditions import Key
-from clients import dynamodb_client, dynamodb_resource, logger
+from clients import dynamodb_client, logger
 from errors import UnhandledAuditTableError
-from constants import AUDIT_TABLE_NAME, AUDIT_TABLE_QUEUE_NAME_GSI, FileStatus, AuditTableKeys
-
-
-def get_next_queued_file_details(queue_name: str) -> Union[dict, None]:
-    """
-    Checks for queued files.
-    Returns a dictionary containing the details of the oldest queued file, or returns None if no queued files are found.
-    """
-    queued_files_found_in_audit_table: dict = dynamodb_resource.Table(AUDIT_TABLE_NAME).query(
-        IndexName=AUDIT_TABLE_QUEUE_NAME_GSI,
-        KeyConditionExpression=Key(AuditTableKeys.QUEUE_NAME).eq(queue_name)
-        & Key(AuditTableKeys.STATUS).eq(FileStatus.QUEUED),
-    )
-
-    queued_files_details: list = queued_files_found_in_audit_table["Items"]
-
-    # Return the oldest queued file
-    return sorted(queued_files_details, key=lambda x: x["timestamp"])[0] if queued_files_details else None
+from constants import AUDIT_TABLE_NAME, FileStatus, AuditTableKeys
 
 
 def change_audit_table_status_to_processed(file_key: str, message_id: str) -> None:

ack_backend/src/clients.py

@@ -6,7 +6,6 @@
 REGION_NAME = "eu-west-2"
 
 firehose_client = boto3_client("firehose", region_name=REGION_NAME)
-lambda_client = boto3_client('lambda', region_name=REGION_NAME)
 dynamodb_client = boto3_client("dynamodb", region_name=REGION_NAME)
 
 dynamodb_resource = boto3_resource("dynamodb", region_name=REGION_NAME)

ack_backend/src/constants.py

@@ -3,9 +3,6 @@
 import os
 
 AUDIT_TABLE_NAME = os.getenv("AUDIT_TABLE_NAME")
-FILE_NAME_PROC_LAMBDA_NAME = os.getenv("FILE_NAME_PROC_LAMBDA_NAME")
-AUDIT_TABLE_FILENAME_GSI = "filename_index"
-AUDIT_TABLE_QUEUE_NAME_GSI = "queue_name_index"
 
 def get_source_bucket_name() -> str:
     """Get the SOURCE_BUCKET_NAME environment from environment variables."""

ack_backend/src/update_ack_file.py

@@ -1,12 +1,11 @@
 """Functions for uploading the data to the ack file"""
 
-import json
 from io import StringIO, BytesIO
 from typing import Union, Optional
 from botocore.exceptions import ClientError
-from constants import ACK_HEADERS, get_source_bucket_name, get_ack_bucket_name, FILE_NAME_PROC_LAMBDA_NAME
-from audit_table import change_audit_table_status_to_processed, get_next_queued_file_details
-from clients import get_s3_client, logger, lambda_client
+from constants import ACK_HEADERS, get_source_bucket_name, get_ack_bucket_name
+from audit_table import change_audit_table_status_to_processed
+from clients import get_s3_client, logger
 from utils_for_ack_lambda import get_row_count
 from logging_decorators import upload_ack_file_logging_decorator
 
@@ -96,12 +95,9 @@ def upload_ack_file(
         move_file(ack_bucket_name, temp_ack_file_key, archive_ack_file_key)
         move_file(source_bucket_name, f"processing/{file_key}", f"archive/{file_key}")
 
-        # Update the audit table and invoke the filename lambda with next file in the queue (if one exists)
+        # Update the audit table
         change_audit_table_status_to_processed(file_key, message_id)
-        supplier_queue = f"{supplier}_{vaccine_type}"
-        next_queued_file_details = get_next_queued_file_details(supplier_queue)
-        if next_queued_file_details:
-            invoke_filename_lambda(next_queued_file_details["filename"], next_queued_file_details["message_id"])
+
         # Ingestion of this file is complete
         result = {
             "message_id": message_id,
@@ -149,27 +145,3 @@ def move_file(bucket_name: str, source_file_key: str, destination_file_key: str)
     )
     s3_client.delete_object(Bucket=bucket_name, Key=source_file_key)
     logger.info("File moved from %s to %s", source_file_key, destination_file_key)
-
-
-def invoke_filename_lambda(file_key: str, message_id: str) -> None:
-    """Invokes the filenameprocessor lambda with the given file key and message id"""
-    try:
-        lambda_payload = {
-            "Records": [
-                {"s3": 
-                    {
-                        "bucket": {
-                            "name": get_source_bucket_name()
-                        },
-                        "object": {"key": file_key}
-                    },
-                    "message_id": message_id
-                }
-            ]
-        }
-        lambda_client.invoke(
-            FunctionName=FILE_NAME_PROC_LAMBDA_NAME, InvocationType="Event", Payload=json.dumps(lambda_payload)
-        )
-    except Exception as error:
-        logger.error("Error invoking filename lambda: %s", error)
-        raise

ack_backend/tests/test_audit_table.py

@@ -8,38 +8,13 @@ class TestAuditTable(unittest.TestCase):
     def setUp(self):
         self.logger_patcher = patch('audit_table.logger')
         self.mock_logger = self.logger_patcher.start()
-        self.dynamodb_resource_patcher = patch('audit_table.dynamodb_resource')
-        self.mock_dynamodb_resource = self.dynamodb_resource_patcher.start()
         self.dynamodb_client_patcher = patch('audit_table.dynamodb_client')
         self.mock_dynamodb_client = self.dynamodb_client_patcher.start()
 
     def tearDown(self):
         self.logger_patcher.stop()
-        self.dynamodb_resource_patcher.stop()
         self.dynamodb_client_patcher.stop()
 
-    def test_get_next_queued_file_details_returns_oldest(self):
-        # Arrange
-        mock_table = MagicMock()
-        self.mock_dynamodb_resource.Table.return_value = mock_table
-        mock_table.query.return_value = {
-            "Items": [
-                {"timestamp": 2, "my-key": "value2"},
-                {"timestamp": 1, "my-key": "value1"},
-            ]
-        }
-        # Act
-        result = audit_table.get_next_queued_file_details("queue1")
-        # Assert
-        self.assertEqual(result, {"timestamp": 1, "my-key": "value1"})
-
-    def test_get_next_queued_file_details_returns_none_if_empty(self):
-        mock_table = MagicMock()
-        self.mock_dynamodb_resource.Table.return_value = mock_table
-        mock_table.query.return_value = {"Items": []}
-        result = audit_table.get_next_queued_file_details("queue1")
-        self.assertIsNone(result)
-
     def test_change_audit_table_status_to_processed_success(self):
         # Should not raise
         self.mock_dynamodb_client.update_item.return_value = {}

ack_backend/tests/test_splunk_logging.py

@@ -85,7 +85,7 @@ def extract_all_call_args_for_logger_error(self, mock_logger):
 
     def expected_lambda_handler_logs(self, success: bool, number_of_rows, ingestion_complete=False, diagnostics=None):
         """Returns the expected logs for the lambda handler function."""
-        # Mocking of timings is such that the time taken is 2 seconds for each row, 
+        # Mocking of timings is such that the time taken is 2 seconds for each row,
         # plus 2 seconds for the handler if it succeeds (i.e. it calls update_ack_file) or 1 second if it doesn't;
         # plus an extra second if ingestion is complete
         if success:
@@ -321,8 +321,6 @@ def test_splunk_update_ack_file_not_logged(self):
             patch("logging_decorators.send_log_to_firehose") as mock_send_log_to_firehose,
             patch("logging_decorators.logger") as mock_logger,
             patch("update_ack_file.change_audit_table_status_to_processed") as mock_change_audit_table_status_to_processed,
-            patch("update_ack_file.get_next_queued_file_details"),
-            patch("update_ack_file.invoke_filename_lambda"),
         ):
             result = lambda_handler(generate_event(messages), context={})
 
@@ -362,8 +360,6 @@ def test_splunk_update_ack_file_logged(self):
             patch("logging_decorators.send_log_to_firehose") as mock_send_log_to_firehose,
             patch("logging_decorators.logger") as mock_logger,
             patch("update_ack_file.change_audit_table_status_to_processed") as mock_change_audit_table_status_to_processed,
-            patch("update_ack_file.get_next_queued_file_details"),
-            patch("update_ack_file.invoke_filename_lambda"),
         ):
             result = lambda_handler(generate_event(messages), context={})
 

ack_backend/tests/test_update_ack_file.py

@@ -2,8 +2,6 @@
 
 import unittest
 import os
-from unittest.mock import patch
-from io import StringIO
 from boto3 import client as boto3_client
 from moto import mock_s3
 

ack_backend/tests/test_update_ack_file_flow.py

@@ -1,9 +1,7 @@
-import unittest
-from unittest.mock import patch, MagicMock, call
+from unittest.mock import patch
 from io import StringIO
 
 import update_ack_file
-from update_ack_file import invoke_filename_lambda
 import unittest
 import boto3
 
@@ -42,22 +40,10 @@ def setUp(self):
         self.change_audit_status_patcher = patch('update_ack_file.change_audit_table_status_to_processed')
         self.mock_change_audit_status = self.change_audit_status_patcher.start()
 
-        self.get_next_queued_file_details_patcher = patch('update_ack_file.get_next_queued_file_details')
-        self.mock_get_next_queued_file_details = self.get_next_queued_file_details_patcher.start()
-
-        self.invoke_filename_lambda_patcher = patch('update_ack_file.invoke_filename_lambda')
-        self.mock_invoke_filename_lambda = self.invoke_filename_lambda_patcher.start()
-
-        self.lambda_client_patcher = patch('update_ack_file.lambda_client')
-        self.mock_lambda_client = self.lambda_client_patcher.start()
-        
     def tearDown(self):
         self.logger_patcher.stop()
         self.get_row_count_patcher.stop()
         self.change_audit_status_patcher.stop()
-        self.get_next_queued_file_details_patcher.stop()
-        self.invoke_filename_lambda_patcher.stop()
-        self.lambda_client_patcher.stop()
 
     def test_audit_table_updated_correctly(self):
         """ VED-167 - Test that the audit table has been updated correctly"""
@@ -116,39 +102,3 @@ def test_move_file(self):
 
         # Logger assertion (if logger is mocked)
         self.mock_logger.info.assert_called_with("File moved from %s to %s", file_key, dest_key)
-
-    def test_next_queued_file_triggers_lambda(self):
-        """ VED-167 Test that the next queued file details are used to re-invoke the lambda."""
-        # Setup
-        self.mock_get_row_count.side_effect = [3, 3]
-        next_file = "next_for_lambda.csv"
-        next_message_id = "msg-next-lambda"
-        supplier = "lambda-trigger-supplier"
-        vaccine_type = "vaccine-type"
-        queue_name = f"{supplier}_{vaccine_type}"
-        self.mock_get_next_queued_file_details.return_value = {"filename": next_file, "message_id": next_message_id}
-        accumulated_csv_content = StringIO("header1|header2\n")
-        ack_data_rows = [
-            {"a": 1, "b": 2, "row": "lambda1"},
-            {"a": 3, "b": 4, "row": "lambda2"},
-            {"a": 5, "b": 6, "row": "lambda3"}
-        ]
-        next_key="next_lambda_test.csv"
-        self.s3_client.put_object(
-            Bucket=self.source_bucket_name,
-            Key=f"processing/{next_key}",
-            Body="dummy content"
-        )
-        # Act
-        update_ack_file.upload_ack_file(
-            temp_ack_file_key=f"TempAck/{next_key}",
-            message_id="msg-lambda-trigger",
-            supplier=supplier,
-            vaccine_type=vaccine_type,
-            accumulated_csv_content=accumulated_csv_content,
-            ack_data_rows=ack_data_rows,
-            archive_ack_file_key=f"forwardedFile/{next_key}",
-            file_key=next_key
-        )
-        # Assert: Check that the next queued file was used to re-invoke the lambda
-        self.mock_get_next_queued_file_details.assert_called_once_with(queue_name)