Dy/cumulus 4504 cnm to cma ci workflow integration test

CHANGELOG.md

@@ -222,6 +222,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 - **CUMULUS-4388**
   - Added cnm_to_cma task (lambda).
   - Original cnm_to_cma was written in Java.  Converted to Python.
+- **CUMULUS-4504**
+  - Added cnm_to_cma CI integration test
 - **CUMULUS-4382**
   - Migrated the granule-invalidator task to the `tasks` directory as part of a coreification task in support of providing rolling archive functionality.
 - **CUMULUS-4385**

example/cumulus-tf/cnm_workflow.asl.json

@@ -19,7 +19,7 @@
                   "destination": "{$.meta.cnm}"
                 },
                 {
-                  "source": "{$.output}",
+                  "source": "{$.output_granules}",
                   "destination": "{$.payload}"
                 }
               ]

example/spec/parallel/cnmWorkflow/KinesisTestTriggerSpec.js

@@ -5,7 +5,6 @@ const crypto = require('crypto');
 const cloneDeep = require('lodash/cloneDeep');
 const get = require('lodash/get');
 const isMatch = require('lodash/isMatch');
-const path = require('path');
 const replace = require('lodash/replace');
 const { getJsonS3Object } = require('@cumulus/aws-client/S3');
 const {
@@ -78,7 +77,6 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
   let executionStatus;
   let expectedSyncGranulesPayload;
   let expectedTranslatePayload;
-  let fileData;
   let filePrefix;
   let granuleId;
   let lambdaStep;
@@ -161,32 +159,23 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
             {
               source_bucket: testConfig.bucket,
               name: recordFile.name,
+              filename: recordFile.name,
               type: recordFile.type,
-              bucket: testConfig.bucket,
               path: testDataFolder,
-              url_path: recordFile.uri,
-              size: recordFile.size,
-              checksumType: recordFile.checksumType,
-              checksum: recordFile.checksum,
-              fileName: recordFile.name,
-              key: path.join(testDataFolder, recordFile.name),
             },
           ],
         },
       ],
     };
 
-    fileData = expectedTranslatePayload.granules[0].files[0];
     filePrefix = `file-staging/${testConfig.stackName}/${record.collection}___000/${crypto.createHash('md5').update(record.product.name).digest('hex')}`;
 
     const fileDataWithFilename = {
       bucket: testConfig.buckets.private.name,
       key: `${filePrefix}/${recordFile.name}`,
       fileName: recordFile.name,
-      size: fileData.size,
+      size: recordFile.size,
       type: recordFile.type,
-      checksumType: recordFile.checksumType,
-      checksum: recordFile.checksum,
       source: `${testDataFolder}/${recordFile.name}`,
     };
 
@@ -301,7 +290,7 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
     describe('the TranslateMessage Lambda', () => {
       let lambdaOutput;
       beforeAll(async () => {
-        lambdaOutput = await lambdaStep.getStepOutput(workflowExecution.executionArn, 'CNMToCMA');
+        lambdaOutput = await lambdaStep.getStepOutput(workflowExecution.executionArn, 'CnmToCma');
       });
 
       it('outputs the expectedTranslatePayload object', () => {
@@ -327,7 +316,7 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
       let startStep;
       let endStep;
       beforeAll(async () => {
-        startStep = await lambdaStep.getStepInput(workflowExecution.executionArn, 'CNMToCMA');
+        startStep = await lambdaStep.getStepInput(workflowExecution.executionArn, 'CnmToCma');
         endStep = await lambdaStep.getStepOutput(workflowExecution.executionArn, 'CnmResponse');
       });
 
@@ -456,7 +445,7 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
         prefix: testConfig.stackName,
         granuleId,
         collectionId: constructCollectionId(ruleOverride.collection.name, ruleOverride.collection.version),
-      });
+      }).catch(() => undefined);
     });
 
     it('executes but fails', () => {
@@ -472,23 +461,10 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
     describe('the CnmResponse Lambda', () => {
       let beforeAllFailed = false;
       let lambdaOutput;
-      let failedGranule;
 
       beforeAll(async () => {
         try {
           lambdaOutput = await lambdaStep.getStepOutput(failingWorkflowExecution.executionArn, 'CnmResponse');
-          failedGranule = await waitForApiRecord(
-            getGranule,
-            {
-              prefix: testConfig.stackName,
-              granuleId: record.product.name,
-              collectionId: constructCollectionId(ruleOverride.collection.name, ruleOverride.collection.version),
-            },
-            {
-              status: 'failed',
-              execution: getExecutionUrlFromArn(failingWorkflowExecution.executionArn),
-            }
-          );
         } catch (error) {
           beforeAllFailed = true;
           console.log('CnmResponse Lambda error:::', error);
@@ -520,10 +496,22 @@ describe('The Cloud Notification Mechanism Kinesis workflow', () => {
         }
       });
 
-      it('puts cnm message to cumulus message for granule record', () => {
-        const cnm = get(lambdaOutput, 'meta.granule.queryFields.cnm');
-        expect(isMatch(cnm, badRecord)).toBe(true);
-        expect(get(failedGranule, 'queryFields.cnm')).toEqual(cnm);
+      it('puts cnm message on the CnmResponse output', () => {
+        const cnm = get(lambdaOutput, 'meta.cnmResponse');
+        expect(cnm).toEqual(jasmine.objectContaining({
+          version: badRecord.version,
+          submissionTime: badRecord.submissionTime,
+          collection: badRecord.collection,
+          provider: badRecord.provider,
+          identifier: badRecord.identifier,
+          receivedTime: jasmine.any(String),
+          response: jasmine.objectContaining({
+            status: 'FAILURE',
+            errorCode: 'TRANSFER_ERROR',
+          }),
+          processCompleteTime: jasmine.any(String),
+        }));
+        expect(cnm.product).toBeUndefined();
       });
     });
   });

example/spec/parallel/cnmWorkflow/KinesisTestTriggerWithUniqueGranuleIdsSpec.js

@@ -5,7 +5,6 @@
 const cloneDeep = require('lodash/cloneDeep');
 const get = require('lodash/get');
 const isMatch = require('lodash/isMatch');
-const path = require('path');
 const replace = require('lodash/replace');
 const { getJsonS3Object } = require('@cumulus/aws-client/S3');
 const {
@@ -82,11 +81,10 @@
   let duplicateRuleOverride;
   let duplicateWorkflowExecution;
   let executionNamePrefix;
   let existingGranuleId;
   let expectedSyncGranulesPayload;
   let expectedTranslatePayload;
   let failingWorkflowExecution;
-  let fileData;
   let initialExecutionStatus;
   let initialRecord;
   let initialRuleDirectory;
@@ -200,31 +198,21 @@
             {
               source_bucket: testConfig.bucket,
               name: recordFile.name,
+              filename: recordFile.name,
               type: recordFile.type,
-              bucket: testConfig.bucket,
               path: testDataFolder,
-              url_path: recordFile.uri,
-              size: recordFile.size,
-              checksumType: recordFile.checksumType,
-              checksum: recordFile.checksum,
-              fileName: recordFile.name,
-              key: path.join(testDataFolder, recordFile.name),
             },
           ],
         },
       ],
     };
 
-    fileData = expectedTranslatePayload.granules[0].files[0];
-
     const fileDataWithFilename = {
       bucket: testConfig.buckets.private.name,
       key: 'key_placeholder',
       fileName: recordFile.name,
-      size: fileData.size,
+      size: recordFile.size,
       type: recordFile.type,
-      checksumType: recordFile.checksumType,
-      checksum: recordFile.checksum,
       source: `${testDataFolder}/${recordFile.name}`,
     };
 
@@ -331,7 +319,7 @@
     describe('the TranslateMessage Lambda', () => {
       let lambdaOutput;
       beforeAll(async () => {
-        lambdaOutput = await lambdaStep.getStepOutput(initialWorkflowExecution.executionArn, 'CNMToCMA');
+        lambdaOutput = await lambdaStep.getStepOutput(initialWorkflowExecution.executionArn, 'CnmToCma');
       });
 
       it('outputs the expectedTranslatePayload object', () => {
@@ -357,7 +345,7 @@
       let startStep;
       let endStep;
       beforeAll(async () => {
-        startStep = await lambdaStep.getStepInput(initialWorkflowExecution.executionArn, 'CNMToCMA');
+        startStep = await lambdaStep.getStepInput(initialWorkflowExecution.executionArn, 'CnmToCma');
         endStep = await lambdaStep.getStepOutput(initialWorkflowExecution.executionArn, 'CnmResponse');
       });
 
@@ -615,23 +603,10 @@
     describe('the CnmResponse Lambda', () => {
       let beforeAllFailed = false;
       let lambdaOutput;
-      let failedGranule;
 
       beforeAll(async () => {
         try {
           lambdaOutput = await lambdaStep.getStepOutput(failingWorkflowExecution.executionArn, 'CnmResponse');
-          failedGranule = await waitForApiRecord(
-            getGranule,
-            {
-              prefix: testConfig.stackName,
-              granuleId: uniqueGranuleIdError,
-              collectionId: constructCollectionId(initialRuleOverride.collection.name, initialRuleOverride.collection.version),
-            },
-            {
-              status: 'failed',
-              execution: getExecutionUrlFromArn(failingWorkflowExecution.executionArn),
-            }
-          );
         } catch (error) {
           beforeAllFailed = true;
           console.log('CnmResponse Lambda error:::', error);
@@ -643,10 +618,6 @@
         if (beforeAllFailed) fail('beforeAll() failed to prepare test suite');
       });
 
-      it('failed granule has the same granuleId as successful granule', () => {
-        expect(failedGranule.granuleId === existingGranuleId);
-      });
-
       it('sends the error to the CnmResponse task', async () => {
         const CnmResponseInput = await lambdaStep.getStepInput(failingWorkflowExecution.executionArn, 'CnmResponse');
         expect(CnmResponseInput.exception.Error).toEqual('FileNotFound');
@@ -667,10 +638,22 @@
         }
       });
 
-      it('puts cnm message to cumulus message for granule record', () => {
-        const cnm = get(lambdaOutput, 'meta.granule.queryFields.cnm');
-        expect(isMatch(cnm, badRecord)).toBe(true);
-        expect(get(failedGranule, 'queryFields.cnm')).toEqual(cnm);
+      it('puts cnm message on the CnmResponse output', () => {
+        const cnm = get(lambdaOutput, 'meta.cnmResponse');
+        expect(cnm).toEqual(jasmine.objectContaining({
+          version: badRecord.version,
+          submissionTime: badRecord.submissionTime,
+          collection: badRecord.collection,
+          provider: badRecord.provider,
+          identifier: badRecord.identifier,
+          receivedTime: jasmine.any(String),
+          response: jasmine.objectContaining({
+            status: 'FAILURE',
+            errorCode: 'TRANSFER_ERROR',
+          }),
+          processCompleteTime: jasmine.any(String),
+        }));
+        expect(cnm.product).toBeUndefined();
       });
     });
   });

example/spec/parallel/cnmWorkflow/data/records/L2_HR_PIXC_product_0001-of-4154.json

@@ -2,7 +2,7 @@
   "version": "1.5",
   "provider": "PODAAC_SWOT",
   "collection": "L2_HR_PIXC",
-  "submissionTime": "2017-09-30T03:42:29.791198",
+  "submissionTime": "2017-09-30T03:42:29.791198Z",
   "identifier": "<<REPLACED with random string>>",
   "product": {
     "name": "L2_HR_PIXC_product_0001-of-4154",

example/spec/parallel/cnmWorkflow/data/records/L2_HR_PIXC_product_0001-of-4154_dupe.json

@@ -2,7 +2,7 @@
   "version": "1.5",
   "provider": "PODAAC_SWOT",
   "collection": "L2_HR_PIXC",
-  "submissionTime": "2017-10-01T05:45:29.000000",
+  "submissionTime": "2017-10-01T05:45:29.000000Z",
   "identifier": "<<REPLACED with random string>>",
   "product": {
     "name": "L2_HR_PIXC_product_0001-of-4154",

example/spec/parallel/cnmWorkflow/data/records/ascat_20121029_010301_metopb_00588_eps_o_coa_2101_ovw.l2.json

@@ -2,7 +2,7 @@
   "version": "1.4",
   "provider": "NASA/JPL/PO.DAAC",
   "collection": "ASCATB-L2-Coastal/op",
-  "submissionTime": "2022-08-10T17:49:30.477188",
+  "submissionTime": "2022-08-10T17:49:30.477188Z",
   "identifier": "<<REPLACED with random string>>",
   "product": {
     "name": "ascat_20121029_010301_metopb_00588_eps_o_coa_2101_ovw.l2",

tasks/cnm-to-cma/deploy/README.md

@@ -0,0 +1,58 @@
+# aws-api-proxy deploy module
+
+This Terraform configuration deploys the `aws-api-proxy` Lambda task using the shared [cumulus-task](../../../tf-modules/cumulus-task/README.md) module.
+
+## What this deploy config does
+
+- Resolves a Lambda execution role from either:
+	- a regex (`lambda_processing_role_pattern`), or
+	- a direct ARN (`lambda_processing_role_arn`)
+- Resolves private application subnet IDs for the current region
+- Creates the `aws-api-proxy` Lambda function from `../dist/final/lambda.zip`
+- Applies timeout, memory, VPC, and tags settings
+
+## Prerequisites
+
+- Build the task package so `tasks/aws-api-proxy/dist/final/lambda.zip` exists
+- Terraform and AWS credentials configured for the target account/region
+- Either a role that matches `lambda_processing_role_pattern` or a direct `lambda_processing_role_arn`
+- A subnet with tag name `Private application <region>a subnet`
+
+## Usage
+
+From this directory:
+
+```bash
+cd tasks/aws-api-proxy/deploy
+
+# Required deploy-time variables
+export TF_VAR_prefix=<your_prefix>
+export TF_VAR_lambda_processing_role_pattern='^<your-prefix>-.*lambda-processing.*$'
+
+# OR provide a direct role ARN (do not set both)
+# export TF_VAR_lambda_processing_role_arn='arn:aws:iam::<account-id>:role/<role-name>'
+
+# Optional
+export TF_VAR_tags='{"Project":"cumulus"}'
+
+terraform init
+terraform apply
+```
+
+## Inputs
+
+| Name | Description | Type | Default | Required |
+| --- | --- | --- | --- | :---: |
+| lambda_processing_role_pattern | Regex pattern to match IAM role name when lambda_processing_role_arn is not provided | `string` | `""` | no |
+| lambda_processing_role_arn | The ARN of the IAM role to use for the Lambda function. If not provided, lambda_processing_role_pattern will be used to find a matching role. | `string` | `""` | no |
+| lambda_timeout | The timeout value for the Lambda function in seconds | `number` | n/a | yes |
+| lambda_memory_size | The memory size for the Lambda function in MB | `number` | n/a | yes |
+| security_group_id | Security group ID for Lambda VPC configuration | `string` | `""` | no |
+| prefix | The prefix for resource names | `string` | n/a | yes |
+| tags | A map of tags to apply to resources | `map(string)` | `{}` | no |
+
+Default values for task-specific settings are in [terraform.tfvars](./terraform.tfvars).
+
+## Notes
+
+- This deploy config includes Terraform `check` blocks that fail early if both role inputs are set (or both are empty), if the role pattern does not match exactly one role, or if no expected subnet is found.

tasks/cnm-to-cma/deploy/terraform.tfvars

@@ -0,0 +1,10 @@
+# This tfvars file is only used if this task is deployed in isolation and not referenced by other tasks.
+
+# This tfvars file contains non-sensitive customization that's specific to this task
+# Additional variables are specified at deploy-time as environment variables:
+# export TF_VAR_prefix=<your_prefix_here>
+# export TF_VAR_lambda_processing_role_pattern="^my-prefix-.*lambda-processing.*$"
+# Optionally, you can also specify tags to apply to resources created by this task:
+# export TF_VAR_tags='{"tag_key": "tag_value"}'
+lambda_timeout = 180 # 6 minutes in seconds
+lambda_memory_size = 512

tasks/cnm-to-cma/deploy/variables.tf

@@ -8,13 +8,13 @@ variable "lambda_subnet_ids" {
 }
 
 variable "lambda_timeout" {
-  description = "Timeout value for the Lambda function in seconds"
+  description = "The timeout value for the Lambda function in seconds"
   type        = number
   default     = 300
 }
 
 variable "lambda_memory_size" {
-  description = "Memory size for the Lambda function in MB"
+  description = "The memory size for the Lambda function in MB"
   type        = number
   default     = 512
 }

tasks/cnm-to-cma/src/cnm_to_cma/__init__.py

@@ -0,0 +1,7 @@
+"""CNM to CMA task."""
+
+from .task import lambda_handler
+
+__all__ = [
+    "lambda_handler",
+]