CUMULUS-260 update pdr stats

CHANGELOG.md

@@ -16,6 +16,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   - `@cumulus/deployment` deploys DynamoDB streams for the Collections, Providers and Rules tables as well as a new lambda function called `dbIndexer`. The `dbIndexer` lambda has an event source mapping which listens to each of the DynamoDB streams. The dbIndexer lambda receives events referencing operations on the DynamoDB table and updates the elasticsearch cluster accordingly.
   - The `@cumulus/api` endpoints for collections, providers and rules _only_ query DynamoDB, with the exception of LIST endpoints and the collections' GET endpoint.
 
+- **CUMULUS-260: "PDR page on dashboard only shows zeros."** The PDR stats in LPDAAC are all 0s, even if the dashboard has been fixed to retrieve the correct fields.  The current version of pdr-status-check has a few issues.
+  - pdr is not included in the input/output schema.  It's available from the input event.  So the pdr status and stats are not updated when the ParsePdr workflow is complete.  Adding the pdr to the input/output of the task will fix this.
+  - pdr-status-check doesn't update pdr stats which prevent the real time pdr progress from showing up in the dashboard. To solve this, added lambda function sf-sns-report which is copied from @cumulus/api/lambdas/sf-sns-broadcase with modification, sf-sns-report can be used to report step function status anywhere inside a step function.  So add step sf-sns-report after each pdr-status-check, we will get the PDR status progress at real time.
+  - It's possible a execution is still in the queue and doesn't exist in sfn yet.  Added code to handle 'ExecutionDoesNotExist' error when checking the execution status.
+
 ### Updated
 - Broke up `kes.override.js` of @cumulus/deployment to multiple modules and moved to a new location
 - Expanded @cumulus/deployment test coverage

cumulus/tasks/pdr-status-check/index.js

@@ -78,7 +78,8 @@ function logStatus(output) {
  *     failed: [
  *       { arn: 'arn:456', reason: 'Workflow Aborted' }
  *     ],
- *     completed: []
+ *     completed: [],
+ *     pdr: {}
  *   }
  *
  * @param {Object} event - the event that came into checkPdrStatuses
@@ -117,7 +118,8 @@ function buildOutput(event, groupedExecutions) {
     isFinished: groupedExecutions.running.length === 0,
     running,
     failed,
-    completed
+    completed,
+    pdr: event.input.pdr
   };
 
   if (!output.isFinished) {
@@ -137,28 +139,36 @@ function buildOutput(event, groupedExecutions) {
  * @returns {Promise.<Object>} - an object describing the status of Step
  *   Function executions related to a PDR
  */
-function checkPdrStatuses(event) {
+async function checkPdrStatuses(event) {
   const runningExecutionArns = event.input.running || [];
 
-  const promisedExecutionDescriptions = runningExecutionArns.map((executionArn) =>
-    aws.sfn().describeExecution({ executionArn }).promise());
-
-  return Promise.all(promisedExecutionDescriptions)
-    .then(groupExecutionsByStatus)
-    .then((groupedExecutions) => {
-      const counter = getCounterFromEvent(event) + 1;
-      const exceededLimit = counter >= getLimitFromEvent(event);
+  const executions = [];
+  for (const executionArn of runningExecutionArns) {
+    try {
+      const execution = await aws.sfn().describeExecution({ executionArn }).promise();
+      executions.push(execution);
+    }
+    catch (e) {
+      // it's ok if a execution is still in the queue and has not be executed
+      if (e.code === 'ExecutionDoesNotExist') {
+        executions.push({ executionArn: executionArn, status: 'RUNNING' });
+      }
+      else throw e;
+    }
+  }
 
-      const executionsAllDone = groupedExecutions.running.length === 0;
+  const groupedExecutions = groupExecutionsByStatus(executions);
+  const counter = getCounterFromEvent(event) + 1;
+  const exceededLimit = counter >= getLimitFromEvent(event);
 
-      if (!executionsAllDone && exceededLimit) {
-        throw new IncompleteError(`PDR didn't complete after ${counter} checks`);
-      }
+  const executionsAllDone = groupedExecutions.running.length === 0;
+  if (!executionsAllDone && exceededLimit) {
+    throw new IncompleteError(`PDR didn't complete after ${counter} checks`);
+  }
 
-      const output = buildOutput(event, groupedExecutions);
-      if (!output.isFinished) logStatus(output);
-      return output;
-    });
+  const output = buildOutput(event, groupedExecutions);
+  if (!output.isFinished) logStatus(output);
+  return output;
 }
 exports.checkPdrStatuses = checkPdrStatuses;
 

cumulus/tasks/pdr-status-check/package.json

@@ -40,6 +40,7 @@
     ]
   },
   "dependencies": {
+    "@cumulus/api": "^1.1.1",
     "@cumulus/common": "^1.1.0",
     "@cumulus/cumulus-message-adapter-js": "^1.0.1",
     "@cumulus/ingest": "^1.1.1",

cumulus/tasks/pdr-status-check/schemas/input.json

@@ -23,6 +23,14 @@
     },
     "counter": { "type": "integer" },
     "limit": { "type": "integer" },
-    "isFinished": { "type": "boolean" }
+    "isFinished": { "type": "boolean" },
+    "pdr": {
+      "type": "object",
+      "required": ["name", "path"],
+      "properties": {
+        "name": { "type": "string" },
+        "path": { "type": "string" }
+      }
+    }
   }
 }

cumulus/tasks/pdr-status-check/schemas/output.json

@@ -23,6 +23,14 @@
     },
     "counter": { "type": "integer" },
     "limit": { "type": "integer" },
-    "isFinished": { "type": "boolean" }
+    "isFinished": { "type": "boolean" },
+    "pdr": {
+      "type": "object",
+      "required": ["name", "path"],
+      "properties": {
+        "name": { "type": "string" },
+        "path": { "type": "string" }
+      }
+    }
   }
 }

cumulus/tasks/pdr-status-check/tests/index.js

@@ -9,7 +9,8 @@ const { checkPdrStatuses } = require('../index');
 test('valid output when no running executions', (t) => {
   const event = {
     input: {
-      running: []
+      running: [],
+      pdr: { name: 'test.PDR', path: 'test-path' }
     }
   };
 
@@ -19,7 +20,8 @@ test('valid output when no running executions', (t) => {
         isFinished: true,
         running: [],
         failed: [],
-        completed: []
+        completed: [],
+        pdr: { name: 'test.PDR', path: 'test-path' }
       };
 
       t.deepEqual(output, expectedOutput);
@@ -41,7 +43,8 @@ test('error thrown when limit exceeded', (t) => {
     input: {
       running: ['arn:123'],
       counter: 2,
-      limit: 3
+      limit: 3,
+      pdr: { name: 'test.PDR', path: 'test-path' }
     }
   };
 
@@ -61,26 +64,35 @@ test('returns the correct results in the nominal case', (t) => {
     'arn:1': 'RUNNING',
     'arn:2': 'SUCCEEDED',
     'arn:3': 'FAILED',
-    'arn:4': 'ABORTED'
+    'arn:4': 'ABORTED',
+    'arn:7': null
+  };
+  const error = {
+    message: 'Execution Does Not Exist: arn',
+    code: 'ExecutionDoesNotExist'
   };
 
   const stubSfnClient = {
     describeExecution: ({ executionArn }) => ({
-      promise: () => Promise.resolve({
-        executionArn,
-        status: executionStatuses[executionArn]
-      })
+      promise: () => {
+        if (executionStatuses[executionArn] === null) return Promise.reject(error);
+        return Promise.resolve({
+          executionArn,
+          status: executionStatuses[executionArn]
+        });
+      }
     })
   };
   const stub = sinon.stub(aws, 'sfn').returns(stubSfnClient);
 
   const event = {
     input: {
-      running: ['arn:1', 'arn:2', 'arn:3', 'arn:4'],
+      running: ['arn:1', 'arn:2', 'arn:3', 'arn:4', 'arn:7'],
       completed: ['arn:5'],
       failed: [{ arn: 'arn:6', reason: 'OutOfCheese' }],
       counter: 5,
-      limit: 10
+      limit: 10,
+      pdr: { name: 'test.PDR', path: 'test-path' }
     }
   };
 
@@ -92,7 +104,7 @@ test('returns the correct results in the nominal case', (t) => {
       t.is(output.counter, 6);
       t.is(output.limit, 10);
 
-      t.deepEqual(output.running, ['arn:1']);
+      t.deepEqual(output.running, ['arn:1', 'arn:7']);
       t.deepEqual(output.completed.sort(), ['arn:2', 'arn:5'].sort());
 
       t.is(output.failed.length, 3);

cumulus/tasks/sf-sns-report/README.md

@@ -0,0 +1,55 @@
+# @cumulus/queue-pdrs
+
+[![CircleCI](https://circleci.com/gh/cumulus-nasa/cumulus.svg?style=svg)](https://circleci.com/gh/cumulus-nasa/cumulus)
+
+Broadcast an incoming Cumulus message to SNS.  This lambda function works with Cumulus Message Adapter, and it can be used anywhere in a step function workflow to report granule and PDR status.
+
+To report the PDR's progress as it's being processed, add the following step after the pdr-status-check:
+
+    PdrStatusReport:
+      CumulusConfig:
+        cumulus_message:
+          input: '{$}'
+          outputs:
+            - source: '{$.payload}'
+              destination: '{$.payload}'
+      Type: Task
+      Resource: ${SfSnsReportLambdaFunction.Arn}
+
+To report the start status of the step function:
+
+    StartAt: StatusReport
+    States:
+     StatusReport:
+      CumulusConfig:
+        cumulus_message:
+          input: '{$}'
+          outputs:
+            - source: '{$.payload}'
+              destination: '{$.payload}'
+      Type: Task
+      Resource: ${SfSnsReportLambdaFunction.Arn}
+
+To report the final status of the step function:
+
+    StopStatus:
+      CumulusConfig:
+        sfnEnd: true
+        stack: '{$.meta.stack}'
+        bucket: '{$.meta.buckets.internal}'
+        stateMachine: '{$.cumulus_meta.state_machine}'
+        executionTime: '{$.cumulus_meta.execution_name}'
+        cumulus_message:
+          input: '{$}'
+      Type: Task
+      Resource: ${SfSnsReportLambdaFunction.Arn}
+
+## What is Cumulus?
+
+Cumulus is a cloud-based data ingest, archive, distribution and management prototype for NASA's future Earth science data streams.
+
+[Cumulus Documentation](https://cumulus-nasa.github.io/)
+
+## Contributing
+
+See [Cumulus README](https://github.com/cumulus-nasa/cumulus/blob/master/README.md#installing-and-deploying)

cumulus/tasks/sf-sns-report/index.js

@@ -0,0 +1,127 @@
+'use strict';
+
+const get = require('lodash.get');
+const { StepFunction } = require('@cumulus/ingest/aws');
+const { setGranuleStatus, sns } = require('@cumulus/common/aws');
+const errors = require('@cumulus/common/errors');
+const cumulusMessageAdapter = require('@cumulus/cumulus-message-adapter-js');
+
+/**
+ * Determines if there was a valid exception in the input message
+ *
+ * @param {Object} event - aws event object
+ * @returns {boolean} true if there was an exception, false otherwise
+ */
+function eventFailed(event) {
+  if (event.exception) {
+    if (typeof event.exception === 'object') {
+      // this is needed to avoid flagging cases like "exception: {}" or "exception: 'none'"
+      if (Object.keys(event.exception).length > 0) {
+        return true;
+      }
+    }
+  }
+  // Error and error keys are not part of the cumulus message
+  // and if they appear in the message something is seriously wrong
+  else if (event.Error || event.error) {
+    return true;
+  }
+  return false;
+}
+
+/**
+ * if the cumulus message shows that a previous step failed,
+ * this function extract the error message from the cumulus message
+ * and fail the function with that information. This ensures that the
+ * Step Function workflow fails with the correct error info
+ *
+ * @param {Object} event - aws event object
+ * @returns {undefined} throws an error and does not return anything
+ */
+function makeLambdaFunctionFail(event) {
+  const error = get(event, 'exception.Error', get(event, 'error.Error'));
+  const cause = get(event, 'exception.Cause', get(event, 'error.Cause'));
+  if (error) {
+    if (errors[error]) {
+      throw new errors[error](cause);
+    }
+    else if (error === 'TypeError') {
+      throw new TypeError(cause);
+    }
+    throw new Error(cause);
+  }
+
+  throw new Error('Step Function failed for an unknown reason.');
+}
+
+/**
+ * Publishes incoming Cumulus Message in its entirety to
+ * a given SNS topic
+ *
+ * @param  {Object} event - a Cumulus Message that has been sent through the
+ * Cumulus Message Adapter. See schemas/input.json for detailed input schema.
+ * @param {Object} event.config - configuration object for the task
+ * @param {Object} event.config.sfnEnd - indicate if it's the last step of the step function
+ * @param {string} event.config.stack - the name of the deployment stack
+ * @param {string} event.config.bucket - S3 bucket
+ * @param {string} event.config.stateMachine - current state machine
+ * @param {string} event.config.executionTime - execution time
+ * @returns {Promise.<Object>} - AWS SNS response. see schemas/output.json for detailed output
+ * schema that is passed to the next task in the workflow
+ */
+async function publishSnsMessage(event) {
+  const config = get(event, 'config', []);
+  const message = get(event, 'input', []);
+
+  const finished = get(config, 'sfnEnd', false);
+  const topicArn = get(message, 'meta.topic_arn', null);
+  const failed = eventFailed(message);
+
+  if (topicArn) {
+    // if this is the sns call at the end of the execution
+    if (finished) {
+      message.meta.status = failed ? 'failed' : 'completed';
+      const granuleId = get(message, 'meta.granuleId', null);
+      if (granuleId) {
+        await setGranuleStatus(
+          granuleId,
+          get(config, 'stack', null),
+          get(config, 'bucket', null),
+          get(config, 'stateMachine', null),
+          get(config, 'executionName', null),
+          message.meta.status
+        );
+      }
+    }
+    else {
+      message.meta.status = 'running';
+    }
+
+    await sns().publish({
+      TopicArn: topicArn,
+      Message: JSON.stringify(message)
+    }).promise();
+  }
+
+  if (failed) {
+    makeLambdaFunctionFail(message);
+  }
+
+  return message;
+}
+
+exports.publishSnsMessage = publishSnsMessage;
+
+/**
+ * Lambda handler. It broadcasts an incoming Cumulus message to SNS
+ *
+ * @param {Object} event - a Cumulus Message
+ * @param {Object} context - an AWS Lambda context object
+ * @param {Function} callback - an AWS Lambda call back
+ * @returns {Promise} updated event object
+ */
+function handler(event, context, callback) {
+  return StepFunction.pullEvent(event).then((message) =>
+    cumulusMessageAdapter.runCumulusTask(publishSnsMessage, message, context, callback));
+}
+exports.handler = handler;