init: ingestion logging

Author: JamesW1-NHS

recordprocessor/src/batch_processor.py

@@ -8,9 +8,14 @@
 from send_to_kinesis import send_to_kinesis
 from clients import logger
 from file_level_validation import file_level_validation
+from logging_decorator import ingestion_logging_decorator
 from errors import NoOperationPermissions, InvalidHeaders
 
 
+@ingestion_logging_decorator
+def ingestion_progress(finished: bool, message_body: dict, row_count: int = 0, start_time: float = 0.0) -> str:
+    return("Ingestion finished" if finished else "Ingestion started")
+
 def process_csv_to_fhir(incoming_message_body: dict) -> None:
     """
     For each row of the csv, attempts to transform into FHIR format, sends a message to kinesis,
@@ -31,8 +36,12 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     csv_reader = interim_message_body.get("csv_dict_reader")
 
     target_disease = map_target_disease(vaccine)
-
     row_count = 0
+    ingestion_start_time = time.time()
+    logger.info(ingestion_progress(
+                    finished=False,
+                    message_body=interim_message_body))
+
     for row in csv_reader:
         row_count += 1
         row_id = f"{file_id}^{row_count}"
@@ -55,6 +64,12 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
 
         logger.info("Total rows processed: %s", row_count)
 
+    logger.info(ingestion_progress(
+                    finished=True,
+                    message_body=interim_message_body,
+                    row_count=row_count,
+                    start_time=ingestion_start_time))
+
 
 def main(event: str) -> None:
     """Process each row of the file"""

recordprocessor/src/logging_decorator.py

@@ -69,3 +69,31 @@ def wrapper(*args, **kwargs):
             raise
 
     return wrapper
+
+def ingestion_logging_decorator(func):
+    """
+    Sends the appropriate logs to Cloudwatch and Firehose based on ingestion started/finished.
+    """
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        incoming_message_body = kwargs.get("message_body")
+        base_log_data = {
+            "function_name": f"record_processor_{func.__name__}",
+            "date_time": str(datetime.now()),
+            "file_key": incoming_message_body.get("file_key"),
+            "message_id": incoming_message_body.get("message_id"),
+            "vaccine_type": incoming_message_body.get("vaccine"),
+            "supplier": incoming_message_body.get("supplier"),
+        }
+        finished = kwargs.get("finished")
+
+        message = func(*args, **kwargs)
+        additional_log_data = {"statusCode": 200, "message": message}
+        if finished:
+            additional_log_data["row_count"] = kwargs.get("row_count")
+        start_time = kwargs.get("start_time") if finished else time.time()
+        generate_and_send_logs(start_time, base_log_data, additional_log_data=additional_log_data)
+        return message
+
+    return wrapper

recordprocessor/tests/test_logging_decorator.py

@@ -253,3 +253,5 @@ def test_splunk_logger_unhandled_failure(self):
         mock_firehose_client.put_record.assert_called_once_with(
             DeliveryStreamName=Firehose.STREAM_NAME, Record=expected_firehose_record
         )
+
+# TODO: unit tests for ingestion decorator