Update the queue-granules task to fetch collection configs for each granule from S3

.eslint-ratchet-high-water-mark

@@ -1 +1 @@
-1306
+1145

.eslintrc.json

@@ -51,16 +51,7 @@
     "generator-star-spacing": "off",
     "import/no-extraneous-dependencies": "off",
     "import/newline-after-import": "off",
-    "no-warning-comments": [
-      2,
-      {
-        "terms": [
-          "TODO",
-          "fixme"
-        ],
-        "location": "anywhere"
-      }
-    ],
+    "no-warning-comments": "off",
     "no-unused-vars": [
       "error",
       { "argsIgnorePattern": "^_" }
@@ -97,6 +88,7 @@
         "ignorePattern": "(https?:|JSON\\.parse|[Uu]rl =)"
       }
     ],
-    "arrow-parens": ["error", "always"]
+    "arrow-parens": ["error", "always"],
+    "prefer-destructuring": "off"
   }
 }

CHANGELOG.md

@@ -13,6 +13,35 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   - fixed pdr-status-check schema, the failed execution contains arn and reason
 - **CUMULUS-206** make sure homepage and repository urls exist in package.json files of tasks and packages
 
+### Changed
+- [CUMULUS-450](https://bugs.earthdata.nasa.gov/browse/CUMULUS-450) - Updated
+  the config schema of the **queue-granules** task
+  - The config no longer takes a "collection" property
+  - The config now takes an "internalBucket" property
+  - The config now takes a "stackName" property
+- [CUMULUS-450](https://bugs.earthdata.nasa.gov/browse/CUMULUS-450) - Updated
+  the config schema of the **parse-pdr** task
+  - The config no longer takes a "buckets" property
+  - The config no longer takes a "collection" property
+  - The config now takes an "internalBucket" property
+  - The "stack", "provider", and "internalBucket" config properties are now
+    required
+
+### Removed
+- Removed the `findTmpTestDataDirectory()` function from
+  `@cumulus/common/test-utils`
+
+### Fixed
+- [CUMULUS-450](https://bugs.earthdata.nasa.gov/browse/CUMULUS-450)
+  - The **queue-granules** task now enqueues a **sync-granule** task with the
+    correct collection config for that granule based on the granule's
+    data-type. It had previously been using the collection config from the
+    config of the **queue-granules** task, which was a problem if the granules
+    being queued belonged to different data-types.
+  - The **parse-pdr** task now handles the case where a PDR contains granules
+    with different data types, and uses the correct granuleIdExtraction for
+    each granule.
+
 ### Added
 - **CUMULUS-448** Add code coverage checking using [nyc](https://github.com/istanbuljs/nyc).
 - **CUMULUS-469** Added a lambda to the API package to prototype creating an S3 bucket policy for direct, in-region S3 access for the prototype bucket

packages/api/models/collections.js

@@ -1,5 +1,6 @@
 'use strict';
 
+const { CollectionConfigStore } = require('@cumulus/common');
 const { S3 } = require('@cumulus/ingest/aws');
 const Manager = require('./base');
 const collectionSchema = require('./schemas').collection;
@@ -44,9 +45,9 @@ class Collection extends Manager {
   }
 
   async create(item) {
-    // write the record to S3
-    const key = `${process.env.stackName}/collections/${item.name}.json`;
-    await S3.put(process.env.internal, key, JSON.stringify(item));
+    const collectionConfigStore =
+      new CollectionConfigStore(process.env.internal, process.env.stackName);
+    await collectionConfigStore.put(item.name, item);
 
     return super.create(item);
   }

packages/common/collection-config-store.js

@@ -0,0 +1,87 @@
+'use strict';
+
+const { s3 } = require('./aws');
+
+/**
+ * Store and retrieve collection configs in S3
+ */
+class CollectionConfigStore {
+  /**
+   * Initialize a CollectionConfigFetcher instance
+   *
+   * @param {string} bucket - the bucket where collection configs are stored
+   * @param {string} stackName - the Cumulus deployment stack name
+   */
+  constructor(bucket, stackName) {
+    this.bucket = bucket;
+    this.stackName = stackName;
+    this.cache = {};
+  }
+
+  /**
+   * Fetch a collection config from S3 (or cache if available)
+   *
+   * @param {string} dataType - the name of the collection config to fetch
+   * @returns {Object} the fetched collection config
+   */
+  async get(dataType) {
+    // Check to see if the collection config has already been cached
+    if (!this.cache[dataType]) {
+      let response;
+      try {
+        // Attempt to fetch the collection config from S3
+        response = await s3().getObject({
+          Bucket: this.bucket,
+          Key: this.configKey(dataType)
+        }).promise();
+      }
+      catch (err) {
+        if (err.code === 'NoSuchKey') {
+          throw new Error(`A collection config for data type "${dataType}" was not found.`);
+        }
+
+        if (err.code === 'NoSuchBucket') {
+          throw new Error(`Collection config bucket does not exist: ${this.bucket}`);
+        }
+
+        throw err;
+      }
+
+      // Store the fetched collection config to the cache
+      this.cache[dataType] = JSON.parse(response.Body.toString());
+    }
+
+    return this.cache[dataType];
+  }
+
+  /**
+   * Store a collection config to S3
+   *
+   * @param {string} dataType - the name of the collection config to store
+   * @param {Object} config - the collection config to store
+   * @returns {Promise<null>} resolves when the collection config has been written
+   *   to S3
+   */
+  async put(dataType, config) {
+    this.cache[dataType] = config;
+
+    return s3().putObject({
+      Bucket: this.bucket,
+      Key: `${this.stackName}/collections/${dataType}.json`,
+      Body: JSON.stringify(config)
+    }).promise().then(() => null); // Don't leak implementation details to the caller
+  }
+
+  /**
+   * Return the S3 key pointing to the collection config
+   *
+   * @param {string} dataType - the datatype
+   * @returns {string} the S3 key where the collection config is located
+   *
+   * @private
+   */
+  configKey(dataType) {
+    return `${this.stackName}/collections/${dataType}.json`;
+  }
+}
+module.exports = CollectionConfigStore;

packages/common/index.js

@@ -3,3 +3,4 @@
 exports.log = require('./log');
 exports.aws = require('./aws');
 exports.task = require('./task');
+exports.CollectionConfigStore = require('./collection-config-store');

packages/common/package.json

@@ -16,6 +16,7 @@
     "url": "https://github.com/cumulus-nasa/cumulus"
   },
   "scripts": {
+    "build": "true",
     "lint": "eslint .",
     "mocha": "mocha-webpack test/**/*.js",
     "mocha-coverage": "nyc mocha-webpack test/**/*.js",
@@ -58,7 +59,7 @@
     "ajv-cli": "^1.1.1",
     "async": "^2.0.0",
     "aws-sdk": "^2.4.11",
-    "babel-core": "^6.13.2",
+    "babel-core": "^6.26.0",
     "babel-eslint": "^8.2.2",
     "babel-polyfill": "^6.13.0",
     "babel-preset-env": "^1.6.1",
@@ -87,7 +88,6 @@
     "webpack-node-externals": "^1.5.4"
   },
   "devDependencies": {
-    "ajv": "^5.5.2",
     "ava": "^0.25.0",
     "nyc": "^11.6.0"
   }

packages/common/test-utils.js

@@ -177,17 +177,6 @@ async function findGitRepoRootDirectory(dirname) {
 }
 exports.findGitRepoRootDirectory = findGitRepoRootDirectory;
 
-/**
- * Determine the path of the .tmp-test-data directory
- *
- * @returns {Promise.<string>} - the filesystem path of the .tmp-test-data directory
- */
-function findTmpTestDataDirectory() {
-  return exports.findGitRepoRootDirectory(process.cwd())
-    .then((gitRepoRoot) => path.join(gitRepoRoot, '.tmp-test-data'));
-}
-exports.findTmpTestDataDirectory = findTmpTestDataDirectory;
-
 /**
  * Determine the path of the packages/test-data directory
  *

packages/common/tests/.eslintrc.json

@@ -0,0 +1,5 @@
+{
+  "rules": {
+    "no-param-reassign": "off"
+  }
+}

packages/common/tests/collection-config-store.js

@@ -0,0 +1,115 @@
+'use strict';
+
+const test = require('ava');
+const { recursivelyDeleteS3Bucket, s3 } = require('../aws');
+const { randomString } = require('../test-utils');
+const CollectionConfigStore = require('../collection-config-store');
+
+test.beforeEach(async (t) => {
+  t.context.stackName = randomString();
+  t.context.dataType = randomString();
+  t.context.collectionConfig = { name: randomString() };
+
+  t.context.bucket = randomString();
+  await s3().createBucket({ Bucket: t.context.bucket }).promise();
+
+  // Utility function to return the S3 key of a collection config
+  t.context.collectionConfigKey = (dataType) =>
+    `${t.context.stackName}/collections/${dataType}.json`;
+});
+
+test.afterEach(async (t) => {
+  try {
+    await recursivelyDeleteS3Bucket(t.context.bucket);
+  }
+  catch (err) {
+    // Some tests delete the bucket before this "afterEach" hook is run
+    if (err.code !== 'NoSuchBucket') throw err;
+  }
+});
+
+test('get() fetches a collection config from S3', async (t) => {
+  await s3().putObject({
+    Bucket: t.context.bucket,
+    Key: t.context.collectionConfigKey(t.context.dataType),
+    Body: JSON.stringify(t.context.collectionConfig)
+  }).promise();
+
+  const collectionConfigStore = new CollectionConfigStore(t.context.bucket, t.context.stackName);
+  const fetchedCollectionConfig = await collectionConfigStore.get(t.context.dataType);
+
+  t.deepEqual(fetchedCollectionConfig, t.context.collectionConfig);
+});
+
+test('get() does not hit S3 for a cached collection config', async (t) => {
+  await s3().putObject({
+    Bucket: t.context.bucket,
+    Key: t.context.collectionConfigKey(t.context.dataType),
+    Body: JSON.stringify(t.context.collectionConfig)
+  }).promise();
+
+  const collectionConfigStore = new CollectionConfigStore(t.context.bucket, t.context.stackName);
+
+  // Fetch the collection config once so it's in the cache
+  await collectionConfigStore.get(t.context.dataType);
+
+  // Delete the S3 bucket so the config can't be fetched from S3
+  await recursivelyDeleteS3Bucket(t.context.bucket);
+
+  // This get() should use the cache
+  const fetchedCollectionConfig = await collectionConfigStore.get(t.context.dataType);
+
+  t.deepEqual(fetchedCollectionConfig, t.context.collectionConfig);
+});
+
+test('get() throws an exception if the collection config could not be found', async (t) => {
+  const invalidDataType = randomString();
+  const collectionConfigStore = new CollectionConfigStore(t.context.bucket, t.context.stackName);
+
+  try {
+    await collectionConfigStore.get(invalidDataType);
+    t.fail('Expected an error to be thrown');
+  }
+  catch (err) {
+    t.is(err.message, `A collection config for data type "${invalidDataType}" was not found.`);
+  }
+});
+
+test('get() throws an exception if the bucket does not exist', async (t) => {
+  const invalidBucket = randomString();
+  const collectionConfigStore = new CollectionConfigStore(invalidBucket, t.context.stackName);
+
+  try {
+    await collectionConfigStore.get(t.context.dataType);
+    t.fail('Expected an error to be thrown');
+  }
+  catch (err) {
+    t.is(err.message, `Collection config bucket does not exist: ${invalidBucket}`);
+  }
+});
+
+test('put() stores a collection config to S3', async (t) => {
+  const collectionConfigStore = new CollectionConfigStore(t.context.bucket, t.context.stackName);
+  await collectionConfigStore.put(t.context.dataType, t.context.collectionConfig);
+
+  const getObjectResponse = await s3().getObject({
+    Bucket: t.context.bucket,
+    Key: t.context.collectionConfigKey(t.context.dataType)
+  }).promise();
+
+  const storedCollectionConfig = JSON.parse(getObjectResponse.Body.toString());
+  t.deepEqual(storedCollectionConfig, t.context.collectionConfig);
+});
+
+test('put() updates the cache with the new collection config', async (t) => {
+  const collectionConfigStore = new CollectionConfigStore(t.context.bucket, t.context.stackName);
+  await collectionConfigStore.put(t.context.dataType, t.context.collectionConfig);
+
+  // Delete the S3 bucket so the config can't be fetched from S3
+  await recursivelyDeleteS3Bucket(t.context.bucket);
+
+  // This get() should use the cache
+  const fetchedCollectionConfig = await collectionConfigStore.get(t.context.dataType);
+
+  t.deepEqual(fetchedCollectionConfig, t.context.collectionConfig);
+});

packages/ingest/granule.js

@@ -9,10 +9,10 @@ const urljoin = require('url-join');
 const cksum = require('cksum');
 const checksum = require('checksum');
 const errors = require('@cumulus/common/errors');
-const sftpMixin = require('./sftp');
-const ftpMixin = require('./ftp').ftpMixin;
-const httpMixin = require('./http').httpMixin;
-const s3Mixin = require('./s3').s3Mixin;
+const { sftpMixin } = require('./sftp');
+const { ftpMixin } = require('./ftp');
+const { httpMixin } = require('./http');
+const { s3Mixin } = require('./s3');
 const { baseProtocol } = require('./protocol');
 
 class Discover {