VED-470: MESH processing fixes and improvements. (#667)

* VED-470: Add missing permission. Move processed files to an archive prefix.

* VED-470: Increase MESH processor CPU & memory for faster processing. Only process files from the inbound prefix. Set the content type from headers. Override new line handling as recommended in the documentation.

* VED-470: Add tests for content type parsing.

Author: mfjarvis

mesh_processor/src/converter.py

@@ -98,11 +98,19 @@ def transfer_multipart_content(bucket_name: str, file_key: str, boundary: bytes,
         if content_disposition:
             _, content_disposition_params = parse_header_value(content_disposition)
             filename = content_disposition_params.get("filename") or filename
+        content_type = headers.get("Content-Type") or "application/octet-stream"
 
         with open(
             f"s3://{DESTINATION_BUCKET_NAME}/streaming/{filename}",
             "wb",
-            transport_params={"client": s3_client}
+            transport_params={
+                "client": s3_client,
+                "client_kwargs": {
+                    "S3.Client.create_multipart_upload": {
+                        "ContentType": content_type
+                    }
+                }
+            }
         ) as output_file:
             stream_part_body(input_file, boundary, output_file)
 
@@ -133,6 +141,10 @@ def process_record(record: dict) -> None:
 
     logger.info(f"Transfer complete for {file_key}")
 
+    move_file(bucket_name, file_key, bucket_name, f"archive/{file_key}")
+
+    logger.info(f"Archived {file_key}")
+
 
 def lambda_handler(event: dict, _context: dict) -> dict:
     success = True

mesh_processor/tests/test_converter.py

@@ -48,11 +48,18 @@ def test_non_multipart_content_type(self):
         result = invoke_lambda("test-csv-file.csv")
         self.assertEqual(result["statusCode"], 200)
 
-        response = s3.get_object(Bucket="destination-bucket", Key="overridden-filename.csv")
-        body = response["Body"].read().decode("utf-8")
+        get_target_response = s3.get_object(Bucket="destination-bucket", Key="overridden-filename.csv")
+        body = get_target_response["Body"].read().decode("utf-8")
         assert body == "some CSV content"
 
-    def test_non_multipart_content_type_no_mesh_metadata(self):
+        with self.assertRaises(ClientError) as e:
+            s3.head_object(Bucket="source-bucket", Key="test-csv-file.csv")
+        self.assertEqual(e.exception.response["Error"]["Code"], "404")
+
+        head_archive_response = s3.head_object(Bucket="source-bucket", Key="archive/test-csv-file.csv")
+        assert head_archive_response["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+    def test_non_multipart_content_type_without_mesh_metadata(self):
         s3 = boto3.client("s3", region_name="eu-west-2")
         s3.put_object(
             Bucket="source-bucket",
@@ -142,8 +149,10 @@ def test_multipart_content_type(self):
                 response = s3.get_object(Bucket="destination-bucket", Key="test-csv-file.csv")
                 body = response["Body"].read().decode("utf-8")
                 assert body == "some CSV content"
+                content_type = response["ContentType"]
+                assert content_type == "text/csv"
 
-    def test_multipart_content_type_without_filename_from_headers(self):
+    def test_multipart_content_type_without_filename_in_headers(self):
         cases = [
             (
                 "no filename in header",
@@ -187,6 +196,33 @@ def test_multipart_content_type_without_filename_from_headers(self):
                 body = response["Body"].read().decode("utf-8")
                 assert body == "some CSV content"
 
+    def test_multipart_content_type_without_content_type_in_headers(self):
+        body = "\r\n".join([
+            "",
+            "--12345678",
+            'Content-Disposition: form-data; name="File"; filename="test-csv-file.csv"',
+            "",
+            "some CSV content",
+            "--12345678--",
+            ""
+        ])
+        s3 = boto3.client("s3", region_name="eu-west-2")
+        s3.put_object(
+            Bucket="source-bucket",
+            Key="test-dat-file.dat",
+            Body=body.encode("utf-8"),
+            ContentType="multipart/form-data; boundary=12345678",
+        )
+
+        result = invoke_lambda("test-dat-file.dat")
+        self.assertEqual(result["statusCode"], 200)
+
+        response = s3.get_object(Bucket="destination-bucket", Key="test-csv-file.csv")
+        body = response["Body"].read().decode("utf-8")
+        assert body == "some CSV content"
+        content_type = response["ContentType"]
+        assert content_type == "application/octet-stream"
+
     def test_multipart_content_type_with_unix_line_endings(self):
         body = "\r\n".join([
             "",

recordprocessor/src/utils_for_recordprocessor.py

@@ -19,7 +19,7 @@ def get_csv_content_dict_reader(file_key: str) -> DictReader:
     """Returns the requested file contents from the source bucket in the form of a DictReader"""
     response = s3_client.get_object(Bucket=os.getenv("SOURCE_BUCKET_NAME"), Key=file_key)
     binary_io = response["Body"]
-    text_io = TextIOWrapper(binary_io, encoding="utf-8")
+    text_io = TextIOWrapper(binary_io, encoding="utf-8", newline="")
     return DictReader(text_io, delimiter="|")
 
 

terraform/mesh_processor.tf

@@ -137,7 +137,8 @@ resource "aws_iam_policy" "mesh_processor_lambda_exec_policy" {
           "s3:GetObject",
           "s3:ListBucket",
           "s3:PutObject",
-          "s3:CopyObject"
+          "s3:CopyObject",
+          "s3:DeleteObject"
         ]
         Resource = [
           aws_s3_bucket.batch_data_source_bucket.arn,
@@ -213,6 +214,7 @@ resource "aws_lambda_function" "mesh_file_converter_lambda" {
   image_uri     = module.mesh_processor_docker_image[0].image_uri
   architectures = ["x86_64"]
   timeout       = 900
+  memory_size   = 1024
 
   environment {
     variables = {
@@ -240,6 +242,7 @@ resource "aws_s3_bucket_notification" "mesh_datasources_lambda_notification" {
   lambda_function {
     lambda_function_arn = aws_lambda_function.mesh_file_converter_lambda[0].arn
     events              = ["s3:ObjectCreated:*"]
+    filter_prefix       = "inbound/"
   }
 }