CUMULUS-2688: Update bulk operations to fetch granules by unique columns

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 ## Unreleased Phase 3
 
+### Breaking Changes
+
+- **CUMULUS-2688**
+  - Updated bulk operation logic to use collectionId in addition to granuleId to fetch granules.
+  - Tasks using the `bulk-operation` Lambda should provide collectionId and granuleId e.g. { granuleId: xxx, collectionId: xxx }
+
 ### Changed
 
 - **CUMULUS-2312** - RDS Migration Epic Phase 3

example/spec/parallel/testAPI/bulkDeleteFailureSpec.js

@@ -73,7 +73,7 @@ describe('POST /granules/bulkDelete with a failed bulk delete operation', () =>
     postBulkDeleteResponse = await bulkDeleteGranules({
       prefix: config.stackName,
       body: {
-        ids: [granule.granuleId],
+        granules: [{ granuleId: granule.granuleId, collectionId: constructCollectionId(collection.name, collection.version) }],
       },
     });
     postBulkDeleteBody = JSON.parse(postBulkDeleteResponse.body);

example/spec/parallel/testAPI/bulkDeleteSuccessSpec.js

@@ -26,7 +26,7 @@ const {
 const { getGranuleWithStatus } = require('@cumulus/integration-tests/Granules');
 const { createProvider } = require('@cumulus/integration-tests/Providers');
 const { createOneTimeRule } = require('@cumulus/integration-tests/Rules');
-
+const { constructCollectionId } = require('@cumulus/message/Collections');
 const {
   createTimestampedTestId,
   createTestSuffix,
@@ -163,7 +163,7 @@ describe('POST /granules/bulkDelete', () => {
           {
             prefix,
             body: {
-              ids: [granuleId],
+              granules: [{ granuleId, collectionId: constructCollectionId(collection.name, collection.version) }],
               // required to force removal of granules from CMR before deletion
               forceRemoveFromCmr: true,
             },

example/spec/parallel/testAPI/bulkOperationsSuccessSpec.js

@@ -26,6 +26,7 @@ const {
 const { getGranuleWithStatus } = require('@cumulus/integration-tests/Granules');
 const { createProvider } = require('@cumulus/integration-tests/Providers');
 const { createOneTimeRule } = require('@cumulus/integration-tests/Rules');
+const { constructCollectionId } = require('@cumulus/message/Collections');
 
 const {
   createTimestampedTestId,
@@ -167,7 +168,7 @@ describe('POST /granules/bulk', () => {
         postBulkGranulesResponse = await granules.bulkGranules({
           prefix,
           body: {
-            ids: [granuleId],
+            granules: [{ granuleId, collectionId: constructCollectionId(collection.name, collection.version) }],
             workflowName: 'HelloWorldWorkflow',
             queueUrl: scheduleQueueUrl,
           },

packages/api-client/tests/test-granules.js

@@ -68,6 +68,34 @@ test('getGranule calls the callback with the expected object when there is query
   }));
 });
 
+test('getGranule accepts an optional collectionId', async (t) => {
+  const expected = {
+    prefix: t.context.testPrefix,
+    payload: {
+      httpMethod: 'GET',
+      resource: '/{proxy+}',
+      path: `/granules/${t.context.collectionId}/${t.context.granuleId}`,
+    },
+  };
+
+  const callback = (configObject) => {
+    t.deepEqual(configObject, expected);
+    return Promise.resolve({
+      body: JSON.stringify({
+        granuleId: t.context.granuleId,
+        collectionId: t.context.collectionId,
+      }),
+    });
+  };
+
+  await t.notThrowsAsync(granulesApi.getGranule({
+    callback,
+    prefix: t.context.testPrefix,
+    granuleId: t.context.granuleId,
+    collectionId: t.context.collectionId,
+  }));
+});
+
 test('waitForGranules calls getGranules with the expected payload', async (t) => {
   const callback = ({ prefix, payload }) => {
     t.true(payload.path.endsWith(t.context.granuleId));

packages/api/endpoints/granules.js

@@ -15,6 +15,7 @@ const {
   ExecutionPgModel,
   getKnexClient,
   getUniqueGranuleByGranuleId,
+  getGranuleByUniqueColumns,
   GranulePgModel,
   translateApiGranuleToPostgresGranule,
   translatePostgresCollectionToApiCollection,
@@ -548,8 +549,17 @@ async function getByGranuleId(req, res) {
   const granuleId = req.params.granuleName;
 
   let granule;
+  let pgCollection;
   try {
-    granule = await getUniqueGranuleByGranuleId(knex, granuleId);
+    if (collectionId) {
+      pgCollection = await collectionPgModel.get(
+        knex, deconstructCollectionId(collectionId)
+      );
+
+      granule = await getGranuleByUniqueColumns(knex, granuleId, pgCollection.cumulus_id);
+    } else {
+      granule = await getUniqueGranuleByGranuleId(knex, granuleId);
+    }
   } catch (error) {
     if (error instanceof RecordDoesNotExist) {
       if (granule === undefined) {
@@ -582,8 +592,8 @@ async function bulkOperations(req, res) {
   let description;
   if (payload.query) {
     description = `Bulk run ${payload.workflowName} on ${payload.query.size} granules`;
-  } else if (payload.ids) {
-    description = `Bulk run ${payload.workflowName} on ${payload.ids.length} granules`;
+  } else if (payload.granules) {
+    description = `Bulk run ${payload.workflowName} on ${payload.granules.length} granules`;
   } else {
     description = `Bulk run on ${payload.workflowName}`;
   }
@@ -671,7 +681,7 @@ async function bulkDelete(req, res) {
 async function bulkReingest(req, res) {
   const payload = req.body;
   const numOfGranules = (payload.query && payload.query.size)
-    || (payload.ids && payload.ids.length);
+    || (payload.granules && payload.granules.length);
   const description = `Bulk granule reingest run on ${numOfGranules || ''} granules`;
 
   const asyncOperationId = uuidv4();

packages/api/lambdas/bulk-operation.js

@@ -5,38 +5,47 @@ const Logger = require('@cumulus/logger');
 const {
   GranulePgModel,
   getKnexClient,
-  getUniqueGranuleByGranuleId,
   translatePostgresGranuleToApiGranule,
+  getGranuleByUniqueColumns,
+  CollectionPgModel,
 } = require('@cumulus/db');
 const { RecordDoesNotExist } = require('@cumulus/errors');
 
+const { deconstructCollectionId } = require('@cumulus/message/Collections');
 const { chooseTargetExecution } = require('../lib/executions');
 const { deleteGranuleAndFiles } = require('../src/lib/granule-delete');
 const { unpublishGranule } = require('../lib/granule-remove-from-cmr');
 const { updateGranuleStatusToQueued } = require('../lib/writeRecords/write-granules');
-const { getGranuleIdsForPayload } = require('../lib/granules');
+const { getGranulesForPayload } = require('../lib/granules');
 const { reingestGranule, applyWorkflow } = require('../lib/ingest');
 
 const log = new Logger({ sender: '@cumulus/bulk-operation' });
 
 async function applyWorkflowToGranules({
-  granuleIds,
+  granules,
   workflowName,
   meta,
   queueUrl,
   granulePgModel = new GranulePgModel(),
+  collectionPgModel = new CollectionPgModel(),
   granuleTranslateMethod = translatePostgresGranuleToApiGranule,
   applyWorkflowHandler = applyWorkflow,
   updateGranulesToQueuedMethod = updateGranuleStatusToQueued,
   knex,
 }) {
   return await pMap(
-    granuleIds,
-    (async (granuleId) => {
+    granules,
+    (async (granule) => {
       try {
-        const pgGranule = await getUniqueGranuleByGranuleId(
+        const collection = await collectionPgModel.get(
           knex,
-          granuleId,
+          deconstructCollectionId(granule.collectionId)
+        );
+
+        const pgGranule = await getGranuleByUniqueColumns(
+          knex,
+          granule.granuleId,
+          collection.cumulus_id,
           granulePgModel
         );
         const apiGranule = await granuleTranslateMethod({
@@ -51,10 +60,10 @@ async function applyWorkflowToGranules({
           queueUrl,
           asyncOperationId: process.env.asyncOperationId,
         });
-        return granuleId;
+        return granule.granuleId;
       } catch (error) {
-        log.error(`Granule ${granuleId} encountered an error`, error);
-        return { granuleId, err: error };
+        log.error(`Granule ${granule.granuleId} encountered an error`, error);
+        return { granuleId: granule.granuleId, err: error };
       }
     })
   );
@@ -70,7 +79,8 @@ async function applyWorkflowToGranules({
  * @param {Object} [payload.query] - Optional parameter of query to send to ES
  * @param {string} [payload.index] - Optional parameter of ES index to query.
  * Must exist if payload.query exists.
- * @param {Object} [payload.ids] - Optional list of granule IDs to bulk operate on
+ * @param {Object} [payload.granules] - Optional list of granule unique IDs to bulk operate on
+ * e.g. { granuleId: xxx, collectionID: xxx }
  * @param {Function} [unpublishGranuleFunc] - Optional function to delete the
  * granule from CMR. Useful for testing.
  * @returns {Promise}
@@ -81,16 +91,28 @@ async function bulkGranuleDelete(
 ) {
   const deletedGranules = [];
   const forceRemoveFromCmr = payload.forceRemoveFromCmr === true;
-  const granuleIds = await getGranuleIdsForPayload(payload);
+  const granules = await getGranulesForPayload(payload);
   const knex = await getKnexClient();
 
   await pMap(
-    granuleIds,
-    async (granuleId) => {
+    granules,
+    async (granule) => {
       let pgGranule;
       const granulePgModel = new GranulePgModel();
+      const collectionPgModel = new CollectionPgModel();
+
+      const collection = await collectionPgModel.get(
+        knex,
+        deconstructCollectionId(granule.collectionId)
+      );
+
       try {
-        pgGranule = await getUniqueGranuleByGranuleId(knex, granuleId, granulePgModel);
+        pgGranule = await getGranuleByUniqueColumns(
+          knex,
+          granule.granuleId,
+          collection.cumulus_id,
+          granulePgModel
+        );
       } catch (error) {
         // PG Granule being undefined will be caught by deleteGranulesAndFiles
         if (error instanceof RecordDoesNotExist) {
@@ -112,7 +134,7 @@ async function bulkGranuleDelete(
         pgGranule,
       });
 
-      deletedGranules.push(granuleId);
+      deletedGranules.push(granule.granuleId);
     },
     {
       concurrency: 10, // is this necessary?
@@ -133,17 +155,18 @@ async function bulkGranuleDelete(
  * @param {Object} [payload.query] - Optional parameter of query to send to ES
  * @param {string} [payload.index] - Optional parameter of ES index to query.
  * Must exist if payload.query exists.
- * @param {Object} [payload.ids] - Optional list of granule IDs to bulk operate on
+ * @param {Object} [payload.granules] - Optional list of granule unique IDs to bulk operate on
+ * e.g. { granuleId: xxx, collectionID: xxx }
  * @param {function} [applyWorkflowHandler] - Optional handler for testing
  * @returns {Promise}
  */
 async function bulkGranule(payload, applyWorkflowHandler) {
   const knex = await getKnexClient();
   const { queueUrl, workflowName, meta } = payload;
-  const granuleIds = await getGranuleIdsForPayload(payload);
+  const granules = await getGranulesForPayload(payload);
   return await applyWorkflowToGranules({
     knex,
-    granuleIds,
+    granules,
     meta,
     queueUrl,
     workflowName,
@@ -155,22 +178,38 @@ async function bulkGranuleReingest(
   payload,
   reingestHandler = reingestGranule
 ) {
-  const granuleIds = await getGranuleIdsForPayload(payload);
-  log.info(`Starting bulkGranuleReingest for ${JSON.stringify(granuleIds)}`);
+  const granules = await getGranulesForPayload(payload);
+  log.info(`Starting bulkGranuleReingest for ${JSON.stringify(granules)}`);
   const knex = await getKnexClient();
 
   const workflowName = payload.workflowName;
   return await pMap(
-    granuleIds,
-    async (granuleId) => {
+    granules,
+    async (granule) => {
+      const granulePgModel = new GranulePgModel();
+      const collectionPgModel = new CollectionPgModel();
+
+      const collection = await collectionPgModel.get(
+        knex,
+        deconstructCollectionId(granule.collectionId)
+      );
+
       try {
-        const pgGranule = await getUniqueGranuleByGranuleId(knex, granuleId);
+        const pgGranule = await getGranuleByUniqueColumns(
+          knex,
+          granule.granuleId,
+          collection.cumulus_id,
+          granulePgModel
+        );
         const apiGranule = await translatePostgresGranuleToApiGranule({
           granulePgRecord: pgGranule,
           knexOrTransaction: knex,
         });
 
-        const targetExecution = await chooseTargetExecution({ granuleId, workflowName });
+        const targetExecution = await chooseTargetExecution({
+          granuleId: granule.granuleId,
+          workflowName,
+        });
         const apiGranuleToReingest = {
           ...apiGranule,
           ...(targetExecution && { execution: targetExecution }),
@@ -180,10 +219,10 @@ async function bulkGranuleReingest(
           apiGranule: apiGranuleToReingest,
           asyncOperationId: process.env.asyncOperationId,
         });
-        return granuleId;
+        return granule.granuleId;
       } catch (error) {
-        log.error(`Granule ${granuleId} encountered an error`, error);
-        return { granuleId, err: error };
+        log.error(`Granule ${granule.granuleId} encountered an error`, error);
+        return { granuleId: granule.granuleId, err: error };
       }
     },
     {

packages/api/lib/granules.js

@@ -320,7 +320,7 @@ async function getGranuleIdsForPayload(payload) {
  * @param {Object} [payload.query] - Optional parameter of query to send to ES
  * @param {string} [payload.index] - Optional parameter of ES index to query.
  * Must exist if payload.query exists.
- * @returns {Promise<Array<Object>>}
+ * @returns {Promise<Array<ApiGranule>>}
  */
 async function getGranulesForPayload(payload) {
   const { granules, index, query } = payload;

packages/api/lib/request.js

@@ -35,19 +35,20 @@ async function verifyJwtAuthorization(requestJwtToken) {
   return accessToken;
 }
 
+// TODO no tests for this
 function validateBulkGranulesRequest(req, res, next) {
   const payload = req.body;
 
-  if (!payload.ids && !payload.query) {
-    return res.boom.badRequest('One of ids or query is required');
+  if (!payload.granules && !payload.query) {
+    return res.boom.badRequest('One of granules or query is required');
   }
 
-  if (payload.ids && !Array.isArray(payload.ids)) {
-    return res.boom.badRequest(`ids should be an array of values, received ${payload.ids}`);
+  if (payload.granules && !Array.isArray(payload.granules)) {
+    return res.boom.badRequest(`granules should be an array of values, received ${payload.granules}`);
   }
 
-  if (!payload.query && payload.ids && payload.ids.length === 0) {
-    return res.boom.badRequest('no values provided for ids');
+  if (!payload.query && payload.granules && payload.granules.length === 0) {
+    return res.boom.badRequest('no values provided for granules');
   }
 
   if (payload.query