Feature/cumulus 2309 granule parallel deletes

README.md

@@ -43,6 +43,7 @@ This is for installation for Cumulus development.  See the [Cumulus deployment i
 - Docker (only required for testing)
 - docker-compose (only required for testing `pip install docker-compose`)
 - Python 3.6+
+- [pipenv](https://pypi.org/project/pipenv/)
 
 Install the correct node version:
 

packages/api/endpoints/granules.js

@@ -10,7 +10,9 @@ const {
   DeletePublishedGranule,
   RecordDoesNotExist,
 } = require('@cumulus/errors');
+const { GranulePgModel, FilePgModel, getKnexClient } = require('@cumulus/db');
 
+const { deleteGranuleAndFiles } = require('../lib/granules');
 const { asyncOperationEndpointErrorHandler } = require('../app/middleware');
 const Search = require('../es/search').Search;
 const indexer = require('../es/indexer');
@@ -140,37 +142,48 @@ async function del(req, res) {
   log.info(`granules.del ${granuleId}`);
 
   const granuleModelClient = new models.Granule();
+  const granulePgModel = new GranulePgModel();
+  const filePgModel = new FilePgModel();
 
-  let granule;
+  const knex = await getKnexClient({ env: process.env });
+
+  let dynamoGranule;
+  let pgGranule;
   try {
-    granule = await granuleModelClient.getRecord({ granuleId });
+    dynamoGranule = await granuleModelClient.getRecord({ granuleId });
+    pgGranule = await granulePgModel.get(knex, { granule_id: granuleId });
   } catch (error) {
     if (error instanceof RecordDoesNotExist) {
       return res.boom.notFound(error);
     }
     throw error;
   }
 
-  if (granule.detail) {
-    return res.boom.badRequest(granule);
+  // DEPRECATED this can be removed
+  if (dynamoGranule.detail) {
+    return res.boom.badRequest(dynamoGranule);
   }
 
-  try {
-    await granuleModelClient.delete(granule);
-  } catch (error) {
-    if (error instanceof DeletePublishedGranule) {
-      return res.boom.badRequest(error.message);
-    }
-    throw error;
+  if (pgGranule.published) {
+    throw new DeletePublishedGranule('You cannot delete a granule that is published to CMR. Remove it from CMR first');
   }
 
+  await deleteGranuleAndFiles(
+    knex,
+    dynamoGranule,
+    pgGranule,
+    filePgModel,
+    granulePgModel,
+    granuleModelClient
+  );
+
   if (inTestMode()) {
     const esClient = await Search.es(process.env.ES_HOST);
     await indexer.deleteRecord({
       esClient,
       id: granuleId,
       type: 'granule',
-      parent: granule.collectionId,
+      parent: dynamoGranule.collectionId,
       index: process.env.ES_INDEX,
       ignore: [404],
     });

packages/api/lib/FileUtils.js

@@ -7,9 +7,10 @@ const isInteger = require('lodash/isInteger');
 const partial = require('lodash/partial');
 const pick = require('lodash/pick');
 const urljoin = require('url-join');
-const { getObjectSize, parseS3Uri } = require('@cumulus/aws-client/S3');
+const { getObjectSize, parseS3Uri, deleteS3Object } = require('@cumulus/aws-client/S3');
 const { removeNilProperties } = require('@cumulus/common/util');
 const Logger = require('@cumulus/logger');
+const { UnparsableFileLocationError } = require('@cumulus/errors');
 const schemas = require('../models/schemas');
 
 const log = new Logger({ sender: 'api/lib/FileUtils' });

packages/api/lib/granules.js

@@ -3,6 +3,8 @@
 const awsClients = require('@cumulus/aws-client/services');
 const isInteger = require('lodash/isInteger');
 const isNil = require('lodash/isNil');
+const { deleteS3Object } = require('@cumulus/aws-client/S3');
+const pMap = require('p-map');
 
 const FileUtils = require('./FileUtils');
 
@@ -62,10 +64,37 @@ function getGranuleProductVolume(granuleFiles = []) {
     .reduce((x, y) => x + y, 0);
 }
 
+// TODO default arguments
+const deleteGranuleAndFiles = async (
+  knex,
+  dynamoGranule,
+  pgGranule,
+  filePgModel,
+  granulePgModel,
+  granuleModelClient
+) => {
+  const files = await knex(filePgModel.tableName)
+    .where({ granule_cumulus_id: pgGranule.cumulus_id });
+
+  await knex.transaction(async (trx) => {
+    await pMap(
+      files,
+      (file) =>
+        knex.transaction(async (fileTrx) => {
+          await filePgModel.delete(fileTrx, { cumulus_id: file.cumulus_id });
+          await deleteS3Object(file.bucket, file.key);
+        })
+    );
+    await granulePgModel.delete(trx, pgGranule);
+    await granuleModelClient.delete(dynamoGranule);
+  });
+};
+
 module.exports = {
   translateGranule,
   getExecutionProcessingTimeInfo,
   getGranuleTimeToArchive,
   getGranuleTimeToPreprocess,
   getGranuleProductVolume,
+  deleteGranuleAndFiles,
 };