Retry experiment

src/api.v2/controllers/gem2sController.js

@@ -1,33 +1,11 @@
-const { startGem2sPipeline, handleGem2sResponse } = require('../helpers/pipeline/gem2s');
-const { OK, MethodNotAllowedError } = require('../../utils/responses');
+const { runGem2s, handleGem2sResponse } = require('../helpers/pipeline/gem2s');
+const { OK } = require('../../utils/responses');
 const getLogger = require('../../utils/getLogger');
 const parseSNSMessage = require('../../utils/parseSNSMessage');
 const snsTopics = require('../../config/snsTopics');
-const ExperimentParent = require('../model/ExperimentParent');
 
 const logger = getLogger('[Gem2sController] - ');
 
-const runGem2s = async (req, res) => {
-  const { experimentId } = req.params;
-
-  logger.log(`Starting gem2s for experiment ${experimentId}`);
-
-  const { parentExperimentId = null } = await new ExperimentParent()
-    .find({ experiment_id: experimentId })
-    .first();
-  if (parentExperimentId) {
-    throw new MethodNotAllowedError(`Experiment ${experimentId} can't run gem2s`);
-  }
-
-  const newExecution = await startGem2sPipeline(experimentId, req.headers.authorization);
-
-  logger.log(`Started gem2s for experiment ${experimentId} successfully, `);
-  logger.log('New executions data:');
-  logger.log(JSON.stringify(newExecution));
-
-  res.json(OK());
-};
-
 const handleResponse = async (req, res) => {
   let result;
 
@@ -58,7 +36,15 @@ const handleResponse = async (req, res) => {
   res.status(200).send('ok');
 };
 
+const handleGem2sRequest = async (req, res) => {
+  const params = { experimentId: req.params.experimentId };
+
+  await runGem2s(params, req.headers.authorization);
+
+  res.json(OK());
+};
+
 module.exports = {
-  runGem2s,
+  handleGem2sRequest,
   handleResponse,
 };

src/api.v2/controllers/seuratController.js

@@ -1,35 +1,11 @@
-const { startSeuratPipeline, handleSeuratResponse } = require('../helpers/pipeline/seurat');
-const { OK, MethodNotAllowedError } = require('../../utils/responses');
+const { runSeurat, handleSeuratResponse } = require('../helpers/pipeline/seurat');
+const { OK } = require('../../utils/responses');
 const getLogger = require('../../utils/getLogger');
 const parseSNSMessage = require('../../utils/parseSNSMessage');
 const snsTopics = require('../../config/snsTopics');
-const ExperimentParent = require('../model/ExperimentParent');
 
 const logger = getLogger('[SeuratController] - ');
 
-const runSeurat = async (req, res) => {
-  const { experimentId } = req.params;
-
-  logger.log(`Starting seurat for experiment ${experimentId}`);
-
-  const { parentExperimentId = null } = await new ExperimentParent()
-    .find({ experiment_id: experimentId })
-    .first();
-
-  if (parentExperimentId) {
-    throw new MethodNotAllowedError(`Experiment ${experimentId} can't run seurat`);
-  }
-
-  const newExecution = await
-  startSeuratPipeline(experimentId, req.headers.authorization);
-
-  logger.log(`Started seurat for experiment ${experimentId} successfully, `);
-  logger.log('New executions data:');
-  logger.log(JSON.stringify(newExecution));
-
-  res.json(OK());
-};
-
 const handleResponse = async (req, res) => {
   let result;
 
@@ -60,7 +36,16 @@ const handleResponse = async (req, res) => {
   res.status(200).send('ok');
 };
 
+const handleSeuratRequest = async (req, res) => {
+  const params = {
+    experimentId: req.params.experimentId,
+  };
+
+  await runSeurat(params, req.headers.authorization);
+  res.json(OK());
+};
+
 module.exports = {
-  runSeurat,
+  handleSeuratRequest,
   handleResponse,
 };

src/api.v2/controllers/subsetController.js

@@ -1,75 +1,26 @@
-const sqlClient = require('../../sql/sqlClient');
-const getLogger = require('../../utils/getLogger');
-const { GEM2S_PROCESS_NAME } = require('../constants');
-
-const { createSubsetPipeline } = require('../helpers/pipeline/pipelineConstruct');
-const Experiment = require('../model/Experiment');
-const ExperimentExecution = require('../model/ExperimentExecution');
-const ExperimentParent = require('../model/ExperimentParent');
-const UserAccess = require('../model/UserAccess');
-
-const logger = getLogger('[SubsetController] - ');
-
-const runSubset = async (req, res) => {
-  const { experimentId } = req.params;
-
-  logger.log(`Starting subset for experiment ${experimentId}`);
+const { createExperimentToSubset, startSubsetPipeline } = require('../helpers/pipeline/subset');
 
+const handleSubsetRequest = async (req, res) => {
   const {
-    params: { experimentId: parentExperimentId },
+    params: { experimentId },
     body: { name, cellSetKeys },
     user: { sub: userId },
   } = req;
 
-  logger.log(`Creating experiment to subset ${parentExperimentId}`);
-
-  let subsetExperimentId;
-  await sqlClient.get().transaction(async (trx) => {
-    subsetExperimentId = await new Experiment(trx).createCopy(parentExperimentId, name);
-    await new UserAccess(trx).createNewExperimentPermissions(userId, subsetExperimentId);
-
-    await new ExperimentParent(trx).create(
-      { experiment_id: subsetExperimentId, parent_experiment_id: parentExperimentId },
-    );
-  });
-
-  // Samples are not created here, we add them in handleResponse of SubsetSeurat
-
-  logger.log(`Created ${subsetExperimentId}, subsetting experiment ${parentExperimentId} to it`);
+  const subsetExperimentId = await createExperimentToSubset(experimentId, userId, name);
 
-  const {
-    processingConfig: parentProcessingConfig,
-  } = await new Experiment().findById(parentExperimentId).first();
-
-  const { stateMachineArn, executionArn } = await createSubsetPipeline(
-    parentExperimentId,
-    subsetExperimentId,
+  const params = {
+    experimentId,
     name,
     cellSetKeys,
-    parentProcessingConfig,
-    req.headers.authorization,
-  );
-
-  const newExecution = {
-    state_machine_arn: stateMachineArn,
-    execution_arn: executionArn,
+    userId,
+    subsetExperimentId,
   };
 
-  const experimentExecutionClient = new ExperimentExecution();
-
-  await experimentExecutionClient.upsert(
-    {
-      experiment_id: subsetExperimentId,
-      pipeline_type: GEM2S_PROCESS_NAME,
-    },
-    newExecution,
-  );
-
-  logger.log(`Started subset for experiment ${experimentId} successfully, subset experimentId: ${subsetExperimentId}`);
-
+  await startSubsetPipeline(params, req.headers.authorization);
   res.json(subsetExperimentId);
 };
 
 module.exports = {
-  runSubset,
+  handleSubsetRequest,
 };

src/api.v2/helpers/pipeline/__mocks__/gem2s.js

@@ -1,7 +1,9 @@
 const startGem2sPipeline = jest.fn();
 const handleGem2sResponse = jest.fn();
+const runGem2s = jest.fn();
 
 module.exports = {
   startGem2sPipeline,
   handleGem2sResponse,
+  runGem2s,
 };

src/api.v2/helpers/pipeline/gem2s.js

@@ -1,12 +1,13 @@
 const _ = require('lodash');
-const constants = require('../../constants');
+
+const { GEM2S_PROCESS_NAME, QC_PROCESS_NAME } = require('../../constants');
 const getPipelineStatus = require('./getPipelineStatus');
 const { createGem2SPipeline, createQCPipeline } = require('./pipelineConstruct');
 
 const Sample = require('../../model/Sample');
 const Experiment = require('../../model/Experiment');
 const ExperimentExecution = require('../../model/ExperimentExecution');
-
+const ExperimentParent = require('../../model/ExperimentParent');
 const sendNotification = require('./hooks/sendNotification');
 const HookRunner = require('./hooks/HookRunner');
 
@@ -18,6 +19,8 @@ const { getGem2sParams, formatSamples } = require('./shouldGem2sRerun');
 const invalidatePlotsForEvent = require('../../../utils/plotConfigInvalidation/invalidatePlotsForEvent');
 const events = require('../../../utils/plotConfigInvalidation/events');
 
+const { MethodNotAllowedError } = require('../../../utils/responses');
+
 const logger = getLogger('[Gem2sService] - ');
 
 const hookRunner = new HookRunner();
@@ -133,18 +136,18 @@ hookRunner.register('copyS3Objects', [invalidatePlotsForExperiment]);
 hookRunner.registerAll([sendNotification]);
 
 const sendUpdateToSubscribed = async (experimentId, message, io) => {
-  const statusRes = await getPipelineStatus(experimentId, constants.GEM2S_PROCESS_NAME);
+  const statusRes = await getPipelineStatus(experimentId, GEM2S_PROCESS_NAME);
 
   // Concatenate into a proper response.
   const response = {
     ...message,
     status: statusRes,
-    type: constants.GEM2S_PROCESS_NAME,
+    type: GEM2S_PROCESS_NAME,
   };
 
   const { error = null } = message.response || {};
   if (error) {
-    logger.log(`Error in ${constants.GEM2S_PROCESS_NAME} received`);
+    logger.log(`Error in ${GEM2S_PROCESS_NAME} received`);
   }
 
   logger.log('Sending to all clients subscribed to experiment', experimentId);
@@ -189,9 +192,11 @@ const generateGem2sTaskParams = async (experimentId, rawSamples, authJWT) => {
   };
 };
 
-const startGem2sPipeline = async (experimentId, authJWT) => {
+const startGem2sPipeline = async (params, authJWT) => {
   logger.log('Creating GEM2S params...');
 
+  const { experimentId } = params;
+
   const samples = await new Sample().getSamples(experimentId);
 
   const currentGem2SParams = await getGem2sParams(experimentId, samples);
@@ -210,28 +215,43 @@ const startGem2sPipeline = async (experimentId, authJWT) => {
     execution_arn: executionArn,
   };
 
-  const experimentExecutionClient = new ExperimentExecution();
-
-  await experimentExecutionClient.upsert(
-    {
-      experiment_id: experimentId,
-      pipeline_type: 'gem2s',
-    },
+  // Save new execution
+  await new ExperimentExecution().updateExecution(
+    experimentId,
+    GEM2S_PROCESS_NAME,
     newExecution,
+    params,
   );
 
-  await experimentExecutionClient.delete({
+  await new ExperimentExecution().delete({
     experiment_id: experimentId,
-    pipeline_type: 'qc',
+    pipeline_type: QC_PROCESS_NAME,
   });
 
   logger.log('GEM2S params saved.');
 
   return newExecution;
 };
 
-const handleGem2sResponse = async (io, message) => {
+const runGem2s = async (params, authorization) => {
+  const { experimentId } = params;
 
+  logger.log(`Starting gem2s for experiment ${experimentId}`);
+  const { parentExperimentId = null } = await new ExperimentParent()
+    .find({ experiment_id: experimentId })
+    .first();
+  if (parentExperimentId) {
+    throw new MethodNotAllowedError(`Experiment ${experimentId} can't run gem2s`);
+  }
+
+  const newExecution = await startGem2sPipeline(params, authorization);
+
+  logger.log(`Started gem2s for experiment ${experimentId} successfully, `);
+  logger.log('New executions data:');
+  logger.log(JSON.stringify(newExecution));
+};
+
+const handleGem2sResponse = async (io, message) => {
   // Fail hard if there was an error.
   await validateRequest(message, 'GEM2SResponse.v2.yaml');
 
@@ -260,6 +280,7 @@ const handleGem2sResponse = async (io, message) => {
 };
 
 module.exports = {
+  runGem2s,
   startGem2sPipeline,
   handleGem2sResponse,
 };

src/api.v2/helpers/pipeline/pipelineConstruct/index.js

@@ -232,7 +232,12 @@ const createSeuratPipeline = async (experimentId, taskParams, authJWT) => {
 };
 
 const createSubsetPipeline = async (
-  fromExperimentId, toExperimentId, toExperimentName, cellSetKeys, parentProcessingConfig, authJWT,
+  fromExperimentId,
+  toExperimentId,
+  toExperimentName,
+  cellSetKeys,
+  parentProcessingConfig,
+  authJWT,
 ) => {
   const stepsParams = {
     parentExperimentId: fromExperimentId,

src/api.v2/helpers/pipeline/retryExperiment.js

@@ -0,0 +1,47 @@
+const { OK } = require('../../../utils/responses');
+
+const { runGem2s } = require('./gem2s');
+const { runSeurat } = require('./seurat');
+const { startSubsetPipeline } = require('./subset');
+const {
+  GEM2S_PROCESS_NAME, SEURAT_PROCESS_NAME, SUBSET_PROCESS_NAME,
+} = require('../../constants');
+const ExperimentExecution = require('../../model/ExperimentExecution');
+
+const getLogger = require('../../../utils/getLogger');
+
+const logger = getLogger();
+
+const stateMachineToPipeline = {
+  [SUBSET_PROCESS_NAME]: startSubsetPipeline,
+  [GEM2S_PROCESS_NAME]: runGem2s,
+  [SEURAT_PROCESS_NAME]: runSeurat,
+};
+
+const getPipelineType = (stateMachineArn) => Object.keys(stateMachineToPipeline)
+  .find(
+    (processName) => stateMachineArn.includes(processName),
+  );
+
+const retryExperiment = async (req, res) => {
+  const { experimentId } = req.params;
+  logger.log(`Retrying experiment ${experimentId}`);
+  const {
+    retryParams,
+    stateMachineArn,
+  } = await new ExperimentExecution().findOne({ experiment_id: experimentId });
+
+  const pipelineType = getPipelineType(stateMachineArn);
+
+  logger.log(`Previously called with ${JSON.stringify(retryParams)}`);
+  logger.log(`Pipeline type is: ${pipelineType}`);
+
+  // Retry the pipeline
+  await stateMachineToPipeline[pipelineType](retryParams, req.headers.authorization);
+
+  res.json(OK());
+};
+
+module.exports = {
+  retryExperiment,
+};