Application and deployment for fastq sync microservice

Author: alexiswl

config/config.ts

@@ -72,6 +72,10 @@ import {
   getFastqUnarchivingManagerStackProps,
   getFastqUnarchivingManagerTableStackProps,
 } from './stacks/fastqUnarchivingManager';
+import {
+  getFastqSyncManagerStackProps,
+  getFastqSyncManagerTableStackProps,
+} from './stacks/fastqSyncManager';
 
 interface EnvironmentConfig {
   name: string;
@@ -115,6 +119,7 @@ export const getEnvironmentConfig = (stage: AppStage): EnvironmentConfig | null
       accessKeySecretStackProps: getAccessKeySecretStackProps(stage),
       fastqManagerTableStackProps: getFastqManagerTableStackProps(stage),
       fastqUnarchivingManagerTableStackProps: getFastqUnarchivingManagerTableStackProps(),
+      fastqSyncManagerTableStackProps: getFastqSyncManagerTableStackProps(),
     },
     statelessConfig: {
       metadataManagerStackProps: getMetadataManagerStackProps(stage),
@@ -147,6 +152,7 @@ export const getEnvironmentConfig = (stage: AppStage): EnvironmentConfig | null
       pgDDProps: getPgDDProps(stage),
       fastqManagerStackProps: getFastqManagerStackProps(stage),
       fastqUnarchivingManagerStackProps: getFastqUnarchivingManagerStackProps(stage),
+      fastqSyncManagerStackProps: getFastqSyncManagerStackProps(),
     },
   };
 

config/constants.ts

@@ -967,3 +967,8 @@ export const fastqUnarchivingEventDetailType = {
   updateJob: 'FastqUnarchivingJobUpdated',
 };
 export const fastqUnarchivingManagerEventSource = 'orcabus.fastqunarchivingmanager';
+
+/*
+Fastq sync service
+*/
+export const fastqSyncEventDetailType = 'fastqSync';

config/stacks/fastqSyncManager.ts

@@ -0,0 +1,82 @@
+import {
+  /* Events */
+  eventBusName,
+  fastqManagerEventDetails,
+  fastqManagerEventSource,
+  fastqSyncEventDetailType,
+  fastqUnarchivingEventDetailType,
+  fastqUnarchivingManagerEventSource,
+  hostedZoneNameParameterPath,
+  jwtSecretName,
+} from '../constants';
+
+import { FastqSyncManagerStackConfig } from '../../lib/workload/stateless/stacks/fastq-sync/deploy/interfaces';
+import { FastqSyncTableConfig } from '../../lib/workload/stateful/stacks/fastq-sync-dynamodb/deploy/stack';
+
+/*
+export interface FastqSyncManagerStackConfig {
+  /*
+  Orcabus token and zone name for external lambda functions
+  */
+//orcabusTokenSecretsManagerPath: string;
+//hostedZoneNameSsmParameterPath: string;
+
+/*
+  Data tables
+  */
+//fastqSyncDynamodbTableName: string;
+/*
+  Event bus stuff
+  */
+//eventBusName: string;
+// eventTriggers: FastqSyncEventTriggers
+//}
+//*/
+
+const fastqSyncDynamodbTableName = 'fastqSyncTokenTable';
+
+// Stateful
+export const getFastqSyncManagerTableStackProps = (): FastqSyncTableConfig => {
+  return {
+    /* DynamoDB table for fastq list rows */
+    dynamodbTableName: fastqSyncDynamodbTableName,
+  };
+};
+
+// Stateless
+export const getFastqSyncManagerStackProps = (): FastqSyncManagerStackConfig => {
+  return {
+    /*
+    Table stuff
+    */
+    fastqSyncDynamodbTableName: fastqSyncDynamodbTableName,
+
+    /*
+    Events stuff
+    */
+    eventBusName: eventBusName,
+    eventTriggers: {
+      fastqSetUpdated: {
+        eventSource: fastqManagerEventSource,
+        eventDetailType: fastqManagerEventDetails.updateFastqSet,
+      },
+      fastqListRowUpdated: {
+        eventSource: fastqManagerEventSource,
+        eventDetailType: fastqManagerEventDetails.updateFastqListRow,
+      },
+      fastqUnarchiving: {
+        eventSource: fastqUnarchivingManagerEventSource,
+        eventDetailType: fastqUnarchivingEventDetailType.updateJob,
+      },
+      fastqSync: {
+        eventDetailType: fastqSyncEventDetailType,
+      },
+    },
+
+    /*
+    Orcabus token and zone name for external lambda functions
+    */
+    orcabusTokenSecretsManagerPath: jwtSecretName,
+    hostedZoneNameSsmParameterPath: hostedZoneNameParameterPath,
+  };
+};

lib/workload/stateful/stacks/fastq-sync-dynamodb/Readme.md

@@ -0,0 +1,6 @@
+# Fastq Sync dynamodb
+
+Simple database with dynamodb to store task tokens and link them to fastq set ids (And vice versa)
+
+A task token can only map to one fastq set id, but a fastq set id can map to multiple task tokens.
+

lib/workload/stateful/stacks/fastq-sync-dynamodb/deploy/stack.ts

@@ -0,0 +1,22 @@
+import * as cdk from 'aws-cdk-lib';
+import { Construct } from 'constructs';
+import { DynamodbPartitionedPipelineConstruct } from '../../../../components/dynamodb-partitioned-table';
+
+export interface FastqSyncTableConfig {
+  dynamodbTableName: string;
+}
+
+export type FastqSyncManagerTableStackProps = FastqSyncTableConfig & cdk.StackProps;
+
+export class FastqSyncManagerTable extends cdk.Stack {
+  constructor(scope: Construct, id: string, props: FastqSyncManagerTableStackProps) {
+    super(scope, id, props);
+
+    /*
+    Initialise dynamodb table, where id_type is the primary sort key
+    */
+    const dynamodbTable = new DynamodbPartitionedPipelineConstruct(this, 'fastq_sync_table', {
+      tableName: props.dynamodbTableName,
+    });
+  }
+}

lib/workload/stateful/statefulStackCollectionClass.ts

@@ -74,6 +74,10 @@ import {
   FastqUnarchivingManagerTable,
   FastqUnarchivingManagerTableStackProps,
 } from './stacks/fastq-unarchiving-dynamodb/deploy';
+import {
+  FastqSyncManagerTable,
+  FastqSyncManagerTableStackProps,
+} from './stacks/fastq-sync-dynamodb/deploy/stack';
 
 export interface StatefulStackCollectionProps {
   dataBucketStackProps: DataBucketStackProps;
@@ -98,6 +102,7 @@ export interface StatefulStackCollectionProps {
   accessKeySecretStackProps: AccessKeySecretStackProps;
   fastqManagerTableStackProps: FastqManagerTableStackProps;
   fastqUnarchivingManagerTableStackProps: FastqUnarchivingManagerTableStackProps;
+  fastqSyncManagerTableStackProps: FastqSyncManagerTableStackProps;
 }
 
 export class StatefulStackCollection {
@@ -125,6 +130,7 @@ export class StatefulStackCollection {
   readonly accessKeySecretStack: Stack;
   readonly fastqManagerTableStack: Stack;
   readonly fastqUnarchivingManagerTableStack: Stack;
+  readonly fastqSyncManagerTableStack: Stack;
 
   constructor(
     scope: Construct,
@@ -292,6 +298,15 @@ export class StatefulStackCollection {
         ...statefulConfiguration.fastqUnarchivingManagerTableStackProps,
       }
     );
+
+    this.fastqSyncManagerTableStack = new FastqSyncManagerTable(
+      scope,
+      'FastqSyncManagerTableStack',
+      {
+        ...this.createTemplateProps(env, 'FastqSyncManagerTableStack'),
+        ...statefulConfiguration.fastqSyncManagerTableStackProps,
+      }
+    );
   }
 
   /**

lib/workload/stateless/stacks/fastq-sync/Readme.md

@@ -1,15 +1,15 @@
 # Fastq Sync Service
 
-The fastq sync service is a simple service that allows step functions with task tokens to 'hang' 
-until the requirements of either a fastq list row or fastq set have been met. 
+The fastq sync service is a simple service that allows step functions with fastq set ids as inputs to 'hang' 
+until the requirements of the fastq set have been met. 
 
 This is useful for workflow-glue services that have fastq set ids but need to wait for either
 
 1. The fastq set readsets to be created
 2. The fastq set to have been qc'd AND have a fingerprint file and compression information
 3. This is also useful for data sharing services that require the fastqs to be unarchived before they can be shared
 
-The step function will then hang at that step until the task token has been 'unlocked'
+The step function will then hang at that step until the task token has been 'unlocked' by the fastq sync service.
 
 ## Registering task tokens
 
@@ -25,34 +25,38 @@ Workflow glue services can use the fastq sync service by generating the followin
     "fastqSetId": "fqs.123456",
     // Then one or more of the following
     // Requirements can be left out if not needed
-    // Do all fastq list rows in the set contain readsets?
-    "hasReadsets": true,  
-    // Do all fastq list rows in the set contain an ntsm uri?
-    "hasFingerprints": true,  
-    // Do all fastq list rows in the set contain compression information?
-    // Useful if the fastq list rows are in ora format. 
-    // Some pipelines require the gzip file size in bytes in order 
-    // to stream the gzip file from ora back into s3 
-    "hasCompressionInformation": true,  
-    // Do all fastq list rows in the set contain qc information?
-    // We don't use this for anything yet but we may use this in the future
-    // to ensure that a fastq set has met the ideal coverage levels
-    "hasQc": true,
+    "requirements": {
+      // Do all fastq list rows in the set contain readsets?
+      "hasActiveReadSet": true,  
+      // Do all fastq list rows in the set contain an ntsm uri?
+      "hasFingerprint": true,  
+      // Do all fastq list rows in the set contain compression information?
+      // Useful if the fastq list rows are in ora format. 
+      // Some pipelines require the gzip file size in bytes in order 
+      // to stream the gzip file from ora back into s3 
+      "hasFileCompressionInformation": true,  
+      // Do all fastq list rows in the set contain qc information?
+      // We don't use this for anything yet but we may use this in the future
+      // to ensure that a fastq set has met the ideal coverage levels
+      "hasQc": true,
+    }
   }
 }
 ```
 
 The fastq sync service will also trigger the qc, fingerprint or compression information services if they do not exist. 
 
-If any of the fastq list rows are in archive, the fastq sync service will also trigger the archive service to unarchive these fastq list rows.
+If any of the fastq list rows are in archive, the fastq sync service will also trigger the fastq unarchiving service to thaw out these fastq list rows.
+And place them into the 'byob' bucket.  
+
 
 ## Unlocking task tokens
 
-The fastq sync service will listen for the following events:
+The fastq sync service will then also listen for the following event typed:
 
 1. FastqListRowUpdated (from the fastq management service)
 2. FastqSetUpdated (from the fastq management service)
-3. UnarchivingCompleted (from the fastq unarchiving service)
-
-The fastq sync service will then check against the requirements of the fastq set or fastq list row for each task token and if so, unlock the task token.
+3. UnarchivingJobUpdated (from the fastq unarchiving service, where the status is 'SUCCEEDED')
 
+Everytime one of the events is triggered, the fastq sync service will check if the fastq list row or fastq set has met the requirements.
+If all requirements are met for the fastq set, the fastq sync service will unlock the task token.