From nsidc cumulus 3220: Create a new send-pan task

CHANGELOG.md

@@ -6,9 +6,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 ## Unreleased
 
-- **CUMULUS-3188**
-  - Updated QueueGranules to support queueing granules that meet the required API granule schema.
-  - Added optional additional properties to queue-granules input schema
+### Added
+
+- **CUMULUS-3220**
+  - Created a new send-pan task
 - **CUMULUS-3287**
   - Added variable to allow the aws_ecs_task_definition health check to be configurable.
   - Added clarity to how the bucket field needs to be configured for the move-granules task definition
@@ -19,6 +20,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 - **CUMULUS-2985**
   - Changed `onetime` rules RuleTrigger to only execute when the state is `ENABLED` and updated documentation to reflect the change
   - Changed the `invokeRerun` function to only re-run enabled rules
+- **CUMULUS-3188**
+  - Updated QueueGranules to support queueing granules that meet the required API granule schema.
+  - Added optional additional properties to queue-granules input schema
 - **CUMULUS-3252**
   - Updated example/cumulus-tf/orca.tf to use orca v8.0.1
   - Added cumulus task `@cumulus/orca-copy-to-archive-adapter`, and add the task to `tf-modules/ingest`

example/cumulus-tf/parse_pdr_workflow.asl.json

@@ -143,7 +143,7 @@
         {
           "Variable": "$.meta.isPdrFinished",
           "BooleanEquals": true,
-          "Next": "WorkflowSucceeded"
+          "Next": "SendPAN"
         }
       ],
       "Default": "WorkflowSucceeded"
@@ -193,6 +193,44 @@
       "Seconds": 10,
       "Next": "CheckStatus"
     },
+    "SendPAN": {
+      "Parameters": {
+        "cma": {
+          "event.$": "$",
+          "ReplaceConfig": {
+            "FullMessage": true
+          },
+          "task_config": {
+            "provider": "{$.meta.provider}",
+            "remoteDir": "{$.meta.collection.meta.panPath}"
+          }
+        }
+      },
+      "Type": "Task",
+      "Resource": "${send_pan_arn}",
+      "Retry": [
+        {
+          "ErrorEquals": [
+            "Lambda.ServiceException",
+            "Lambda.AWSLambdaException",
+            "Lambda.SdkClientException"
+          ],
+          "IntervalSeconds": 2,
+          "MaxAttempts": 6,
+          "BackoffRate": 2
+        }
+      ],
+      "Catch": [
+        {
+          "ErrorEquals": [
+            "States.ALL"
+          ],
+          "ResultPath": "$.exception",
+          "Next": "WorkflowFailed"
+        }
+      ],
+      "Next": "WorkflowSucceeded"
+    },
     "WorkflowFailed": {
       "Type": "Fail",
       "Cause": "Workflow failed"

example/cumulus-tf/parse_pdr_workflow.tf

@@ -14,6 +14,7 @@ module "parse_pdr_workflow" {
       parse_pdr_task_arn: module.cumulus.parse_pdr_task.task_arn,
       pdr_status_check_task_arn: module.cumulus.pdr_status_check_task.task_arn,
       queue_granules_task_arn: module.cumulus.queue_granules_task.task_arn,
+      send_pan_arn: module.cumulus.send_pan_task.task_arn,
       sf_sqs_report_task_arn: module.cumulus.sf_sqs_report_task.task_arn,
       start_sf_queue_url: module.cumulus.start_sf_queue_url
     }

example/data/collections/s3_MOD09GQ_006/s3_MOD09GQ_006.json

@@ -10,7 +10,8 @@
   "url_path": "{cmrMetadata.Granule.Collection.ShortName}___{cmrMetadata.Granule.Collection.VersionId}/{substring(file.fileName, 0, 3)}",
   "sampleFileName": "MOD09GQ.A2017025.h21v00.006.2017034065104.hdf",
   "meta": {
-    "s3MultipartChunksizeMb": 16
+    "s3MultipartChunksizeMb": 16,
+    "panPath": "integration-test/pans/MOD09GQ___006"
   },
   "files": [
     {

example/spec/parallel/ingest/ingestFromPdrSpec.js

@@ -23,8 +23,9 @@
 
 const flatten = require('lodash/flatten');
 const cryptoRandomString = require('crypto-random-string');
+const path = require('path');
 
-const { deleteS3Object, s3ObjectExists } = require('@cumulus/aws-client/S3');
+const { buildS3Uri, deleteS3Object, s3ObjectExists } = require('@cumulus/aws-client/S3');
 const { s3 } = require('@cumulus/aws-client/services');
 const { LambdaStep } = require('@cumulus/integration-tests/sfnStep');
 const { getPdr } = require('@cumulus/api-client/pdrs');
@@ -478,7 +479,7 @@ describe('Ingesting from PDR', () => {
 
             // the output of the CheckStatus is used to determine the task of choice
             const checkStatusTaskName = 'CheckStatus';
-            const successStepName = 'WorkflowSucceeded';
+            const successStepName = 'SendPAN';
             const pdrStatusReportTaskName = 'PdrStatusReport';
 
             let choiceVerified = false;
@@ -516,6 +517,32 @@ describe('Ingesting from PDR', () => {
           }
         });
       });
+
+      describe('SendPan lambda function', () => {
+        let lambdaOutput;
+        beforeAll(async () => {
+          try {
+            lambdaOutput = await lambdaStep.getStepOutput(parsePdrExecutionArn, 'SendPan');
+          } catch (error) {
+            beforeAllFailed = error;
+          }
+        });
+
+        // SfSnsReport lambda is used in the workflow multiple times, apparently, only the first output
+        it('has expected pan output', async () => {
+          if (beforeAllFailed) fail(beforeAllFailed);
+          const panName = lambdaOutput.payload.pdr.name.replace(/\.pdr/gi, '.pan');
+          const panKey = path.join(addedCollections[0].meta.panPath, panName);
+          const expectedPanUri = buildS3Uri(config.bucket, panKey);
+          expect(lambdaOutput.payload.pan.uri).toEqual(expectedPanUri);
+          const panExists = await s3ObjectExists({
+            Bucket: config.bucket,
+            Key: panKey,
+          });
+          expect(panExists).toEqual(true);
+          await deleteS3Object(config.bucket, panKey);
+        });
+      });
     });
 
     describe('the reporting lambda has received the CloudWatch step function event and', () => {

example/spec/parallel/ingest/ingestPdrWithNodeNameSpec.js

@@ -171,10 +171,6 @@ describe('Ingesting from PDR', () => {
       deleteFolder(config.bucket, testDataFolder),
       cleanupCollections(config.stackName, config.bucket, collectionsDir, testSuffix),
       cleanupProviders(config.stackName, config.bucket, providersDir, testSuffix),
-      apiTestUtils.deletePdr({
-        prefix: config.stackName,
-        pdr: pdrFilename,
-      }),
     ]).catch(console.error);
 
     await providersApi.deleteProvider({
@@ -512,7 +508,7 @@ describe('Ingesting from PDR', () => {
 
             // the output of the CheckStatus is used to determine the task of choice
             const checkStatusTaskName = 'CheckStatus';
-            const successStepName = 'WorkflowSucceeded';
+            const successStepName = 'SendPAN';
             const pdrStatusReportTaskName = 'PdrStatusReport';
 
             let choiceVerified = false;

packages/api/index.js

@@ -4,3 +4,4 @@ exports.models = require('./models');
 exports.testUtils = require('./lib/testUtils');
 exports.tokenUtils = require('./lib/token');
 exports.snsRuleHelpers = require('./lib/snsRuleHelpers');
+exports.pdrHelpers = require('./lib/pdrHelpers');

packages/api/lib/pdrHelpers.js

@@ -0,0 +1,36 @@
+'use strict';
+
+const pvl = require('@cumulus/pvl');
+
+/**
+ * Generate PAN message
+ *
+ * @returns {string} the PAN message
+ */
+function generatePAN() {
+  return pvl.jsToPVL(
+    new pvl.models.PVLRoot()
+      .add('MESSAGE_TYPE', new pvl.models.PVLTextString('SHORTPAN'))
+      .add('DISPOSITION', new pvl.models.PVLTextString('SUCCESSFUL'))
+      .add('TIME_STAMP', new pvl.models.PVLDateTime(new Date()))
+  );
+}
+
+/**
+ * Generate a PDRD message with a given err
+ *
+ * @param {object} err - the error object
+ * @returns {string} the PDRD message
+ */
+function generatePDRD(err) {
+  return pvl.jsToPVL(
+    new pvl.models.PVLRoot()
+      .add('MESSAGE_TYPE', new pvl.models.PVLTextString('SHORTPDRD'))
+      .add('DISPOSITION', new pvl.models.PVLTextString(err.message))
+  );
+}
+
+module.exports = {
+  generatePAN,
+  generatePDRD,
+};

packages/ingest/src/HttpProviderClient.js

@@ -9,6 +9,7 @@ const Crawler = require('simplecrawler');
 const got = require('got');
 const { CookieJar } = require('tough-cookie');
 const { promisify } = require('util');
+const stream = require('node:stream');
 
 const {
   buildS3Uri,
@@ -206,7 +207,7 @@ class HttpProviderClient {
   /**
    * Download a remote file to disk
    *
-   * @param {Object} params
+   * @param {object} params
    * @param {string} params.remotePath - the full path to the remote file to be fetched
    * @param {string} params.localPath - the full local destination file path
    * @returns {Promise.<string>} - the path that the file was saved to
@@ -244,7 +245,7 @@ class HttpProviderClient {
   /**
    * Download the remote file to a given s3 location
    *
-   * @param {Object} params
+   * @param {object} params
    * @param {string} params.fileRemotePath - the full path to the remote file to be fetched
    * @param {string} params.destinationBucket - destination s3 bucket of the file
    * @param {string} params.destinationKey - destination s3 key of the file
@@ -291,6 +292,39 @@ class HttpProviderClient {
     return { s3uri, etag };
   }
 
+  /**
+   * Upload a file
+   *
+   * @param {object} params
+   * @param {string} params.localPath - the full local file path
+   * @param {string} params.uploadPath - the full remote file path for uploading file to
+   * @returns {Promise<string>} the uri of the uploaded file
+   */
+  async upload(params) {
+    const { localPath, uploadPath } = params;
+    log.info(params);
+    await this.setUpGotOptions();
+    await this.downloadTLSCertificate();
+    const options = {
+      protocol: this.protocol,
+      host: this.host,
+      port: this.port,
+      path: uploadPath,
+      method: 'POST',
+    };
+
+    const remoteUrl = buildURL(options);
+    got.stream.options = options;
+    await promisify(pipeline)(
+      fs.createReadStream(localPath),
+      await got.stream.post(remoteUrl),
+      new stream.PassThrough()
+    );
+
+    log.info(`Finishing uploading ${localPath} to ${remoteUrl}`);
+    return remoteUrl;
+  }
+
   /* eslint-disable no-empty-function */
   async connect() {}
 

packages/ingest/src/S3ProviderClient.ts

@@ -15,7 +15,7 @@ class S3ProviderClient implements ProviderClient {
   /**
    * Download a remote file to disk
    *
-   * @param {Object} params
+   * @param {object} params
    * @param {string} params.remotePath - the full path to the remote file to be fetched
    * @param {string} params.localPath - the full local destination file path
    * @param {string} params.remoteAltBucket - alternate per-file bucket override to this.bucket
@@ -79,8 +79,8 @@ class S3ProviderClient implements ProviderClient {
   /**
    * Download the remote file to a given s3 location
    *
-   * @param {Object} params - the full path to the remote file to be fetched
-   * @param {string} params.sourceKey - the full path to the remote file to be fetched
+   * @param {object} params
+   * @param {string} params.fileRemotePath - the full path to the remote file to be fetched
    * @param {string} params.bucket - destination s3 bucket of the file
    * @param {string} params.destinationBucket - destination s3 bucket of the file
    * @param {string} params.destinationKey - destination s3 key of the file
@@ -145,6 +145,34 @@ class S3ProviderClient implements ProviderClient {
     }
   }
 
+  /**
+   * Upload a file
+   *
+   * @param {object} params
+   * @param {string} params.localPath - the full local file path
+   * @param {string} params.uploadPath - the full remote file path for uploading file to
+   * @param {string} [params.remoteAltBucket] - alternate per-file bucket override to this.bucket
+   * @returns {Promise<string>} the uri of the uploaded file
+   */
+  async upload(params: {
+    localPath: string,
+    uploadPath: string,
+    remoteAltBucket?: string,
+  }) {
+    const { localPath, uploadPath, remoteAltBucket } = params;
+    log.info(params);
+    const remoteBucket = remoteAltBucket || this.bucket;
+    await S3.putFile(
+      remoteBucket,
+      uploadPath,
+      localPath
+    );
+
+    const s3Uri = S3.buildS3Uri(remoteBucket, uploadPath);
+    log.info(`Finishing uploading ${localPath} to ${s3Uri}`);
+    return s3Uri;
+  }
+
   /* eslint-disable @typescript-eslint/no-empty-function */
   async connect(): Promise<void> {}
 

packages/ingest/test/test-HttpProviderClient.js

@@ -312,3 +312,18 @@ test.serial('list fails if client wait time is set less than the response delay'
     }
   );
 });
+
+test.serial('upload() attemps to upload a file', async (t) => {
+  const localPath = path.join(tmpdir(), randomString());
+  t.teardown(() => fs.unlinkSync(localPath));
+  const uploadPath = path.join(randomString(), 'destinationfile.txt');
+  fs.writeFileSync(localPath, randomString());
+
+  const { httpProviderClient } = t.context;
+  nock('http://localhost:3030')
+    .post(path.join('/', uploadPath))
+    .reply(200);
+
+  await httpProviderClient.upload({ localPath, uploadPath });
+  t.true(nock.isDone());
+});

packages/ingest/test/test-S3ProviderClient.js

@@ -1,6 +1,7 @@
 'use strict';
 
-const { basename, dirname } = require('path');
+const path = require('path');
+const { tmpdir } = require('os');
 const fs = require('fs');
 const test = require('ava');
 
@@ -50,16 +51,16 @@ test.serial('S3ProviderClient.list lists objects from the bucket root with paths
 
   const files = await s3ProviderClient.list('');
   t.is(files.length, 1);
-  t.is(files[0].name, basename(t.context.sourceKey));
-  t.is(files[0].path, dirname(t.context.sourceKey));
+  t.is(files[0].name, path.basename(t.context.sourceKey));
+  t.is(files[0].path, path.dirname(t.context.sourceKey));
 });
 
 test.serial('S3ProviderClient.list lists objects under a path in a bucket', async (t) => {
   const s3ProviderClient = new S3ProviderClient({ bucket: t.context.sourceBucket });
 
   const files = await s3ProviderClient.list(t.context.sourcePrefix);
   t.is(files.length, 1);
-  t.is(files[0].name, basename(t.context.sourceKey));
+  t.is(files[0].name, path.basename(t.context.sourceKey));
 });
 
 test.serial('S3ProviderClient.download downloads a file to local disk', async (t) => {
@@ -161,3 +162,18 @@ test.serial('S3ProviderClient.sync throws an error if the source file does not e
     }
   );
 });
+
+test.serial('S3ProviderClient.upload uploads a file', async (t) => {
+  const s3ProviderClient = new S3ProviderClient({ bucket: t.context.sourceBucket });
+  const localPath = path.join(tmpdir(), randomString());
+  t.teardown(() => fs.unlinkSync(localPath));
+  const uploadPath = path.join(randomString(), 'destinationfile.txt');
+
+  fs.writeFileSync(localPath, t.context.fileContent);
+  await s3ProviderClient.upload({ localPath, uploadPath });
+
+  t.is(
+    await S3.getTextObject(t.context.sourceBucket, uploadPath),
+    t.context.fileContent
+  );
+});