First pass at processing architecture

sds_data_manager/constructs/batch_compute_resources.py

@@ -0,0 +1,133 @@
+from aws_cdk import RemovalPolicy
+from aws_cdk import aws_batch as batch
+from aws_cdk import aws_ec2 as ec2
+from aws_cdk import aws_ecr as ecr
+from aws_cdk import aws_iam as iam
+from aws_cdk import aws_s3 as s3
+from constructs import Construct
+
+
+class FargateBatchResources(Construct):
+    """Fargate Batch compute environment with named Job Queue, and Job Definition.
+    """
+
+    def __init__(self,
+                 scope: Construct,
+                 construct_id: str,
+                 sds_id: str,
+                 vpc: ec2.Vpc,
+                 processing_step_name: str,
+                 archive_bucket: s3.Bucket,
+                 security_group: classmethod,
+                 batch_max_vcpus=10,
+                 job_vcpus=0.25,
+                 job_memory=512):
+        """Constructor
+
+        Parameters
+        ----------
+        scope : Construct
+        construct_id : str
+        sds_id : str
+            Name suffix for stack
+        vpc : ec2.Vpc
+            VPC into which to launch the compute instance.
+        processing_step_name : str
+            Name of data product being generated in this Batch job.
+        archive_bucket : s3.Bucket
+            S3 bucket.
+        security_group : classmethod
+            Batch processing security group.
+        batch_max_vcpus : int, Optional
+            Maximum number of virtual CPUs per compute instance.
+        job_vcpus : int, Optional
+            Number of virtual CPUs required per Batch job.
+            Dependent on Docker image contents.
+        job_memory : int: Optional
+            Memory required per Batch job in MB. Dependent on Docker image contents.
+        """
+        super().__init__(scope, construct_id)
+
+        self.role = iam.Role(self, f"BatchServiceRole-{sds_id}",
+                             assumed_by=iam.ServicePrincipal('batch.amazonaws.com'),
+                             managed_policies=[
+                                 iam.ManagedPolicy.from_aws_managed_policy_name(
+                                     "service-role/AWSBatchServiceRole")
+                             ]
+                             )
+
+        # Required since our task is hosted on AWS Fargate,
+        # is pulling container images from the ECR, and sending
+        # container logs to CloudWatch.
+        fargate_execution_role = iam.Role(self, f"FargateExecutionRole-{sds_id}",
+                                          assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
+                                          managed_policies=[
+                                              iam.ManagedPolicy.from_aws_managed_policy_name(
+                                                  'service-role/AmazonECSTaskExecutionRolePolicy')
+                                          ])
+
+        # PRIVATE_WITH_NAT allows batch job to pull images from the ECR.
+        self.compute_environment = batch.CfnComputeEnvironment(
+            self, f"FargateBatchComputeEnvironment-{sds_id}",
+            type='MANAGED',
+            service_role=self.role.role_arn,
+            compute_resources=batch.CfnComputeEnvironment.ComputeResourcesProperty(
+                type='FARGATE',
+                maxv_cpus=batch_max_vcpus,
+                subnets=vpc.select_subnets(subnet_type=ec2.SubnetType.PRIVATE_WITH_NAT).subnet_ids,
+                security_group_ids=[security_group.security_group_id]
+            )
+        )
+
+        # The set of compute environments mapped to a job queue
+        # and their order relative to each other
+        compute_environment_order = batch.CfnJobQueue.ComputeEnvironmentOrderProperty(
+            compute_environment=self.compute_environment.ref,
+            order=1)
+
+        # Define registry for storing processing docker images
+        self.container_registry = ecr.Repository(self, f"BatchRepository-{sds_id}",
+                                                 repository_name=f"{processing_step_name.lower()}-repo",
+                                                 image_scan_on_push=True)
+
+        self.container_registry.apply_removal_policy(RemovalPolicy.DESTROY)
+        self.container_registry.grant_pull(fargate_execution_role)
+
+        # Setup job queue
+        self.job_queue_name = f"{processing_step_name}-fargate-batch-job-queue"
+        self.job_queue = batch.CfnJobQueue(self, f"FargateBatchJobQueue-{sds_id}",
+                                           job_queue_name=self.job_queue_name,
+                                           priority=1,
+                                           compute_environment_order=[compute_environment_order])
+
+        # Batch job role, so we can later grant access to the appropriate
+        # S3 buckets and other resources
+        self.batch_job_role = iam.Role(self, f"BatchJobRole-{sds_id}",
+                                       assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
+                                       managed_policies=[
+                                           iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3FullAccess")])
+        archive_bucket.grant_read_write(self.batch_job_role)
+
+        # create job definition
+        self.job_definition_name = f"fargate-batch-job-definition{processing_step_name}"
+        self.job_definition = batch.CfnJobDefinition(
+            self, f"FargateBatchJobDefinition-{sds_id}",
+            job_definition_name=self.job_definition_name,
+            type="CONTAINER",
+            platform_capabilities=['FARGATE'],
+            container_properties={
+                'image': self.container_registry.repository_uri,
+                'resourceRequirements': [
+                    {
+                        'value': str(job_memory),
+                        'type': 'MEMORY'
+                    },
+                    {
+                        'value': str(job_vcpus),
+                        'type': 'VCPU'
+                    }
+                ],
+                'executionRoleArn': fargate_execution_role.role_arn,
+                'jobRoleArn': self.batch_job_role.role_arn,
+            },
+        )

sds_data_manager/constructs/instrument_lambdas.py

@@ -0,0 +1,67 @@
+"""Module containing Constructs for instsrument Lambda functions"""
+from pathlib import Path
+
+from aws_cdk import Duration
+from aws_cdk import aws_lambda as lambda_
+from aws_cdk import aws_lambda_python_alpha as lambda_alpha_
+from aws_cdk import aws_s3 as s3
+from constructs import Construct
+
+
+class InstrumentLambda(Construct):
+    """Generic Construct with customizable runtime code
+    """
+
+    def __init__(self,
+                 scope: Construct,
+                 construct_id: str,
+                 sds_id:str,
+                 processing_step_name: str,
+                 archive_bucket: s3.Bucket,
+                 code_path: str or Path,
+                 instrument_target: str,
+                 instrument_sources: str):
+        """InstrumentLambda Constructor
+
+        Parameters
+        ----------
+        scope : Construct
+        construct_id : str
+        sds_id : str
+            Name suffix for stack
+        processing_step_name : str
+            Processing step name
+        archive_bucket: s3.Bucket
+            S3 bucket
+        code_path : str or Path
+            Path to the Lambda code directory
+        instrument_target : str
+            Target data product
+        instrument_sources : str
+            Data product sources
+        """
+        super().__init__(scope, construct_id)
+
+        # Create Environment Variables
+        lambda_environment = {
+            "PROCESSING_PATH": f"archive-{sds_id}",
+            "INSTRUMENT_SOURCES": instrument_sources,
+            "INSTRUMENT_TARGET": instrument_target,
+            "PROCESSING_NAME": processing_step_name,
+            "OUTPUT_PATH":f"s3://{archive_bucket.bucket_name}/{instrument_target}"
+        }
+
+        self.instrument_lambda = lambda_alpha_.PythonFunction(
+            self,
+            id=f"InstrumentLambda-{processing_step_name}",
+            function_name=f"{processing_step_name}",
+            entry=str(code_path),
+            index=f"instruments/{instrument_target}.py",
+            handler="lambda_handler",
+            runtime=lambda_.Runtime.PYTHON_3_9,
+            timeout=Duration.minutes(10),
+            memory_size=1000,
+            environment=lambda_environment
+        )
+
+        archive_bucket.grant_read_write(self.instrument_lambda)

sds_data_manager/constructs/sdc_step_function.py

@@ -0,0 +1,116 @@
+from aws_cdk import Stack
+from aws_cdk import aws_s3 as s3
+from aws_cdk import aws_stepfunctions as sfn
+from aws_cdk import aws_stepfunctions_tasks as tasks
+from constructs import Construct
+
+from sds_data_manager.constructs.batch_compute_resources import FargateBatchResources
+from sds_data_manager.constructs.instrument_lambdas import InstrumentLambda
+
+
+class SdcStepFunction(Construct):
+    """Step Function Construct
+
+    Creates state machine using processing components.
+    """
+
+    def __init__(self,
+                 scope: Construct,
+                 construct_id: str,
+                 processing_step_name: str,
+                 processing_system: InstrumentLambda,
+                 batch_resources: FargateBatchResources,
+                 instrument_target: str,
+                 archive_bucket: s3.Bucket):
+        """SdcStepFunction Constructor
+
+        Parameters
+        ----------
+        scope : Construct
+        construct_id : str
+        processing_step_name : str
+            The string identifier for the processing step
+        processing_system: BatchProcessingSystem
+            Batch processing system
+        batch_resources: FargateBatchResources
+            Fargate compute environment
+        instrument_target : str
+            Target data product
+        archive_bucket : str
+            S3 bucket
+        """
+        super().__init__(scope, construct_id)
+        print(archive_bucket.bucket_name)
+        # Reformat EventBridge Inputs
+        add_specifics_to_input = sfn.Pass(
+            self, "Reformat EventBridge Inputs",
+            parameters={
+                "TIMEOUT_TIME.$": "$.time",
+            }
+        )
+
+        # Step Functions Tasks to invoke Lambda function
+        instrument_task = tasks.LambdaInvoke(self,
+                                             f"InstrumentLambda-{processing_step_name}",
+                                             lambda_function=processing_system.instrument_lambda,
+                                             payload=sfn.TaskInput.from_object(
+                                                 {"TIMEOUT_TIME.$": "$.TIMEOUT_TIME"}),
+                                             result_path="$.InstrumentOutput",
+                                             result_selector={
+                                                 "STATE.$": "$.Payload.STATE",
+                                                 "JOB_NAME.$": "$.Payload.JOB_NAME",
+                                                 "COMMAND.$": "$.Payload.COMMAND",
+                                                 "OUTPUT_PATH": "$.Payload.OUTPUT_PATH",
+                                                 "INSTRUMENT_TARGET":
+                                                     "$.Payload.INSTRUMENT_TARGET"
+                                             })
+
+        # Batch Job Inputs
+        stack = Stack.of(self)
+        job_definition_arn = \
+            f'arn:aws:batch:{stack.region}:{stack.account}:job-definition/' \
+            f'{batch_resources.job_definition_name}'
+        job_queue_arn = f'arn:aws:batch:{stack.region}:{stack.account}:job-queue/' \
+                        f'{batch_resources.job_queue_name}'
+
+        instrument_target = f"{instrument_target}"
+
+        # Batch Job
+        submit_job = tasks.BatchSubmitJob(
+            self, f"BatchJob-{processing_step_name}",
+            job_name=sfn.JsonPath.string_at("$.InstrumentOutput.JOB_NAME"),
+            job_definition_arn=job_definition_arn,
+            job_queue_arn=job_queue_arn,
+            container_overrides=tasks.BatchContainerOverrides(
+                command=sfn.JsonPath.list_at("$.InstrumentOutput.COMMAND"),
+                environment={
+                    "OUTPUT_PATH": archive_bucket.bucket_name,
+                    "INSTRUMENT_TARGET": instrument_target
+                }
+            ),
+            result_path='$.BatchJobOutput'
+        )
+
+        # Success and Fail Final States
+        fail_state = sfn.Fail(self, "Fail State")
+
+        # Choice State
+        instrument_status = sfn.Choice(self, "Success?")
+        # Go to Batch job
+        created = sfn.Condition.string_equals("$.InstrumentOutput.STATE", "SUCCESS")
+        instrument_status.when(created, submit_job)
+        instrument_status.otherwise(fail_state)
+
+        submit_job.add_catch(fail_state)
+
+        # State sequences
+        add_specifics_to_input.next(
+            instrument_task).next(
+            instrument_status)
+
+        # Define the state machine
+        definition_body = sfn.DefinitionBody.from_chainable(add_specifics_to_input)
+        self.state_machine = sfn.StateMachine(self,
+                                              f"CDKProcessingStepStateMachine-{processing_step_name}",
+                                              definition_body=definition_body,
+                                              state_machine_name=f"{processing_step_name}-step-function")

sds_data_manager/lambda_code/SDSCode/instruments/l1b_Codice.py

@@ -0,0 +1,37 @@
+"""Lambda runtime code that triggers off of arrival of data into S3 bucket.
+"""
+import os
+from datetime import datetime
+
+import boto3
+
+#TODO: ability to access database, EFS, calibration data, etc.
+
+def lambda_handler(event: dict, context):
+    """Handler function"""
+    print(event)
+    print(context)
+
+    now = datetime.now()
+    print("Now time is")
+    print(now)
+
+    # Get the environment variables
+    bucket = os.environ['PROCESSING_PATH']
+    prefix = os.environ['INSTRUMENT_SOURCES']
+
+    # Retrieves objects in the S3 bucket under the given prefix
+    try:
+        s3 = boto3.client('s3')
+        object_list = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)["Contents"]
+        print(object_list)
+    except KeyError:
+        print("No files present.")
+
+    #TODO: this state will change based on availability of data
+    #TODO: we need to think about what needs to be passed into the container
+    return {
+        "STATE": "SUCCESS",
+        "JOB_NAME": os.environ['PROCESSING_NAME'],
+        'COMMAND': ["packet-ingest"]
+    }

sds_data_manager/lambda_code/SDSCode/instruments/l1c_Codice.py

@@ -0,0 +1,37 @@
+"""Lambda runtime code that triggers off of arrival of data into S3 bucket.
+"""
+import os
+from datetime import datetime
+
+import boto3
+
+#TODO: ability to access database, EFS, calibration data, etc.
+
+def lambda_handler(event: dict, context):
+    """Handler function"""
+    print(event)
+    print(context)
+
+    now = datetime.now()
+    print("Now time is")
+    print(now)
+
+    # Get the environment variables
+    bucket = os.environ['PROCESSING_PATH']
+    prefix = os.environ['INSTRUMENT_SOURCES']
+
+    # Retrieves objects in the S3 bucket under the given prefix
+    try:
+        s3 = boto3.client('s3')
+        object_list = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)["Contents"]
+        print(object_list)
+    except KeyError:
+        print("No files present.")
+
+    #TODO: this state will change based on availability of data
+    #TODO: we need to think about what needs to be passed into the container
+    return {
+        "STATE": "SUCCESS",
+        "JOB_NAME": os.environ['PROCESSING_NAME'],
+        'COMMAND': ["packet-ingest"]
+    }