Merge pull request #38 from awslabs/jjbain-bring-your-own-arn

Jjbain bring your own arn

Author: ravij3

backend/backend/handlers/pipelines/createPipeline.py

@@ -65,6 +65,17 @@ def upload_Pipeline(body):
     print("Setting Time Stamp")
     dtNow = datetime.datetime.utcnow().strftime('%B %d %Y - %H:%M:%S')
 
+    userResource = {
+        'isProvided': False,
+        'resourceId': ''
+    }
+    if body['containerUri'] != None:
+        userResource['isProvided'] = True 
+        userResource['resourceId'] = body['containerUri'] 
+    elif body['lambdaName'] != None:
+        userResource['isProvided'] = True 
+        userResource['resourceId'] = body['lambdaName']
+
     item = {
         'databaseId': body['databaseId'],
         'pipelineId':body['pipelineId'],
@@ -73,13 +84,17 @@ def upload_Pipeline(body):
         'description': body['description'],
         'dateCreated': json.dumps(dtNow),
         'pipelineType':body['pipelineType'],
+        'userProvidedResource': json.dumps(userResource),
         'enabled':False #Not doing anything with this yet
     }
     table.put_item(
         Item=item,
         ConditionExpression='attribute_not_exists(databaseId) and attribute_not_exists(pipelineId)'
     )
-        
+    #If a lambda function name or ECR container URI was provided by the user, creation is not necessary
+    if userResource['isProvided'] == True:
+        return json.dumps({"message": 'Succeeded'})
+    
     print("Running CFT")
     if body['pipelineType']=='SageMaker':
         createSagemakerPipeline(body)
@@ -186,7 +201,7 @@ def lambda_handler(event, context):
     }
     print(event['body'])
     if isinstance(event['body'], str):
-        event['body'] = json.loads(event['body'])    
+        event['body'] = json.loads(event['body']) 
     try:
         reqs=['databaseId','pipelineId','description','assetType','pipelineType','outputType']
         for r in reqs:

backend/backend/handlers/pipelines/pipelineService.py

@@ -97,12 +97,21 @@ def delete_pipeline(databaseId, pipelineId):
     if "#deleted" in databaseId:
         return response
     item = get_pipeline(databaseId, pipelineId)
+    
     if item:
         print("Deleting pipeline: ", item)
-        if item['pipelineType'] == 'SageMaker':
-            delete_stack(item['pipelineId'])
-        else: #Lambda Pipeline
-            delete_lambda(item['pipelineId'])
+        try:
+            userResource = json.loads(item['userProvidedResource'])
+            if userResource['isProvided'] == False:
+                if item['pipelineType'] == 'SageMaker':
+                    delete_stack(item['pipelineId'])
+                else: #Lambda Pipeline
+                    delete_lambda(item['pipelineId'])
+        except KeyError: #For pipelines created before user provided resources were implemented
+            if item['pipelineType'] == 'SageMaker':
+                delete_stack(item['pipelineId'])
+            else: #Lambda Pipeline
+                delete_lambda(item['pipelineId'])
 
         item['databaseId'] = databaseId + "#deleted"
         table.put_item(

backend/backend/handlers/workflows/createWorkflow.py

@@ -268,8 +268,18 @@ def create_step_function(pipelines, databaseId, workflowId):
 
 
 def create_sagemaker_step(databaseId, region, role, account_id, job_names, instance_type, i, pipeline, input_s3_uri, output_s3_uri):
-    image_uri = account_id+'.dkr.ecr.'+region + \
-        '.amazonaws.com/'+pipeline['name']
+
+    try: 
+        userResource = json.loads(pipeline['userProvidedResource'])
+        if userResource['isProvided'] == False:    
+            image_uri = account_id+'.dkr.ecr.'+region + \
+                '.amazonaws.com/'+pipeline['name']
+        else:
+            image_uri = userResource['resourceId']
+    except KeyError: #For pipelines created before user provided resources were implemented
+        image_uri = account_id+'.dkr.ecr.'+region + \
+                '.amazonaws.com/'+pipeline['name']
+
     processor = Processor(
         role=role,
         image_uri=image_uri,
@@ -307,6 +317,15 @@ def create_sagemaker_step(databaseId, region, role, account_id, job_names, insta
 
 
 def create_lambda_step(pipeline, input_s3_uri, output_s3_uri):
+    try:
+        userResource = json.loads(pipeline['userProvidedResource'])
+        if userResource['isProvided'] == False:
+            functionName = pipeline['name']
+        else:
+            functionName = userResource['resourceId']
+    except KeyError: #For pipelines created before user provided resources were implemented
+        functionName = pipeline['name']
+
     lambda_payload = {
         "body": {
             "inputPath.$": input_s3_uri,
@@ -317,6 +336,6 @@ def create_lambda_step(pipeline, input_s3_uri, output_s3_uri):
         state_id="{}-{}".format(pipeline['name'], uuid.uuid1().hex),
         parameters={
             # replace with the name of your function
-            "FunctionName": pipeline['name'],
+            "FunctionName": functionName,
             "Payload": lambda_payload
         })

infra/lib/storage-builder.ts

@@ -11,6 +11,7 @@ import { Duration } from "aws-cdk-lib";
 import { BlockPublicAccess } from "aws-cdk-lib/aws-s3";
 import { Construct } from "constructs";
 import { requireTLSAddToResourcePolicy } from "./security";
+import { NagSuppressions } from "cdk-nag";
 
 export interface storageResources {
     s3: {
@@ -33,7 +34,6 @@ export interface storageResources {
 export function storageResourcesBuilder(scope: Construct, staging_bucket?: string): storageResources {
     const accessLogsBucket = new s3.Bucket(scope, "AccessLogsBucket", {
         encryption: s3.BucketEncryption.S3_MANAGED,
-        serverAccessLogsPrefix: "access-log-bucket-logs/",
         versioned: true,
         blockPublicAccess: BlockPublicAccess.BLOCK_ALL,
     });
@@ -45,6 +45,10 @@ export function storageResourcesBuilder(scope: Construct, staging_bucket?: strin
 
     requireTLSAddToResourcePolicy(accessLogsBucket);
 
+    NagSuppressions.addResourceSuppressions(accessLogsBucket, [
+        {id: 'AwsSolutions-S1', reason: 'This is an access logs bucket, we do not want access logs to be reporting to itself, causing a loop.'}
+    ]);
+
     const assetBucket = new s3.Bucket(scope, "AssetBucket", {
         cors: [
             {

web/src/components/createupdate/CreateUpdateElement.js

@@ -125,6 +125,7 @@ export default function CreateUpdateElement(props) {
     if (open) {
       const newFormErrors = Object.assign({}, formErrors);
       const formPropNames = Object.keys(formValues);
+      console.log(formPropNames)
       for (let i = 0; i < formPropNames.length; i++) {
         const formPropName = formPropNames[i];
         //ignore values without validation on entity
@@ -173,6 +174,9 @@ export default function CreateUpdateElement(props) {
             } else {
               acc[cur] = formValues[cur];
             }
+            if (acc[cur] == undefined || acc[cur] == null){ 
+              acc[cur] = null;
+            }
             return acc;
           },
           {}
@@ -269,13 +273,22 @@ export default function CreateUpdateElement(props) {
                 elementDefinition,
                 options,
                 defaultOption,
+                appearsWhen,
               } = controlDefinition;
               const { FormElement, elementProps } = elementDefinition;
               if (!formValues[id] && defaultOption) {
                 handleUpdateFormValues(id, defaultOption);
               }
+
+              //find the form field that isn't toggled based on pipeline type, then clear it.
+              const optionalFieldHidden = appearsWhen && formValues[appearsWhen[0]] && appearsWhen[1] != formValues[appearsWhen[0]]["label"]
+              if(formValues[id] && optionalFieldHidden){
+                console.log(id+ ": "+ formValues[id])
+                formValues[id] = ""
+              }
+
               return (
-                <div key={i}>
+                optionalFieldHidden || <div key={i}>
                   <FormField
                     constraintText={constraintText}
                     label={label}

web/src/components/createupdate/CreateUpdateWorkflow.js

@@ -65,6 +65,7 @@ export default function CreateUpdateWorkflow(props) {
   useEffect(() => {
     const getData = async () => {
       const items = await fetchDatabaseWorkflows({databaseId: databaseId});
+      console.log(items)
       if (items !== false && Array.isArray(items)) {
         setReload(false);
         const currentItem = items.find(
@@ -78,7 +79,8 @@ export default function CreateUpdateWorkflow(props) {
             return {
               value: item.name,
               type: item.pipelineType, 
-              outputType: item.outputType
+              outputType: item.outputType,
+              userProvidedResource: item.userProvidedResource
             };
           }
         );
@@ -142,7 +144,7 @@ export default function CreateUpdateWorkflow(props) {
       setActiveTab("pipelines");
     } else {
       const functions = workflowPipelines.map((item) => {
-        return { name: item.value, pipelineType: item.type, outputType: item.outputType };
+        return { name: item.value, pipelineType: item.type, outputType: item.outputType, userProvidedResource: item.userProvidedResource };
       });
       const config = {
         body: {

web/src/components/createupdate/entity-types/EntityPropTypes.js

@@ -47,6 +47,15 @@ export const validateFileType = (value) => {
   return result[0] === String(value);
 };
 
+export const validateContainerUri = value => {
+  const regex = /^[0-9]{12}.dkr.ecr.+?amazonaws.com\/.+/g;
+  const result = String(value).match(regex);
+  if (result === null) {
+    return false;
+  }
+  return result[0] === String(value);
+}
+
 export const fileTypePropType = function (props, propName) {
   //check if prop is supplied by option element
   let testValue = props[propName];
@@ -233,6 +242,37 @@ export const typedObjectArrayPropType = function (type, props, propName) {
 
   return null;
 };
+
+export const containerUriPropType = function(props, propName){
+  //check if prop is supplied by option element
+  let testValue = props[propName];
+  if (testValue && testValue.hasOwnProperty("value"))
+    testValue = testValue.value;
+  console.log(testValue);
+  if (!validateContainerUri(testValue)) {
+    if(!verifyIsNotEmpty(testValue) || testValue == null){
+      return null;
+    }
+    return new Error(
+      `Invalid value for ${propName}. Enter a valid Amazon ECR image Uri ACCOUNT_NUMBER.dkr.ecr.REGION.amazonaws.com/IMAGE_NAME`
+    );
+  }
+  return null;
+}
+
+export const validateLambdaName = function(props, propName){
+  //check if prop is supplied by option element
+  let testValue = props[propName];
+  if (testValue && testValue.hasOwnProperty("value"))
+    testValue = testValue.value;
+  
+  if(!verifyIsNotEmpty(testValue)){
+    return null;
+  }
+
+  return stringMaxLength.bind(null, 64)
+}
+
 export const EntityPropTypes = {
   ENTITY_ID: entityIdPropType,
   ENTITY_ID_ARRAY: entityIdArrayPropType,
@@ -242,6 +282,8 @@ export const EntityPropTypes = {
   STRING_64: stringMaxLength.bind(null, 64),
   STRING_128: stringMaxLength.bind(null, 128),
   STRING_256: stringMaxLength.bind(null, 256),
+  CONTAINER_URI: containerUriPropType,
+  LAMBDA_NAME: validateLambdaName,
   BOOL: boolPropType,
   OBJECT: objectPropType,
   TYPED_OBJECT: typedObjectPropType,

web/src/components/createupdate/entity-types/PipelineEntity.js

@@ -11,13 +11,17 @@ export default function PipelineEntity(props) {
         databaseId,
         description,
         pipelineType,
+        externalContainerUri,
+        externalLambdaName,
         assetType,
         outputType,
     } = props;
     this.pipelineId = pipelineId;
     this.databaseId = databaseId;
     this.description = description;
     this.pipelineType = pipelineType;
+    this.externalContainerUri = externalContainerUri;
+    this.externalLambdaName = externalLambdaName;
     this.assetType = assetType;
     this.outputType = outputType;
 }
@@ -29,4 +33,6 @@ PipelineEntity.propTypes = {
     pipelineType: EntityPropTypes.STRING_32,
     assetType: EntityPropTypes.FILE_TYPE,
     outputType: EntityPropTypes.FILE_TYPE,
+    containerUri: EntityPropTypes.CONTAINER_URI,
+    lambdaName: EntityPropTypes.LAMBDA_NAME,
 };

web/src/components/createupdate/form-definitions/PipelineFormDefinition.js

@@ -55,6 +55,7 @@ export const PipelineFormDefinition = new FormDefinition({
         formElement: Input,
         elementProps: { autoFocus: true },
       }),
+      required: true,
     }),
     new ControlDefinition({
       label: "Database Name",
@@ -64,6 +65,7 @@ export const PipelineFormDefinition = new FormDefinition({
         formElement: DatabaseSelector,
         elementProps: {},
       }),
+      required: true,
     }),
     new ControlDefinition({
       label: "Pipeline Type",
@@ -77,6 +79,29 @@ export const PipelineFormDefinition = new FormDefinition({
           disabled: false,
         },
       }),
+      required: true,
+    }),
+    new ControlDefinition({
+      label: "Container Uri (Optional)",
+      id: "containerUri",
+      constraintText: "ACCOUNT_NUMBER.dkr.ecr.REGION.amazonaws.com/IMAGE_NAME. If you do not provide an image stored in Amazon ECR, an Amazon Sagemaker notebook instance will be provisioned on your behalf with steps to deploy one.",
+      elementDefinition: new ElementDefinition({
+        formElement: Input,
+        elementProps: { autoFocus: false },
+      }),
+      appearsWhen: ["pipelineType", "SageMaker"],
+      required: false,
+    }),
+    new ControlDefinition({
+      label: "Lambda Function Name (Optional)",
+      id: "lambdaName",
+      constraintText: "If no name is provided a template lambda function will be deployed on your behalf",
+      elementDefinition: new ElementDefinition({
+        formElement: Input,
+        elementProps: { autoFocus: false },
+      }),
+      appearsWhen: ["pipelineType", "Lambda"],
+      required: false,
     }),
     new ControlDefinition({
       label: "Description",
@@ -86,6 +111,7 @@ export const PipelineFormDefinition = new FormDefinition({
         formElement: Textarea,
         elementProps: { rows: 4 },
       }),
+      required: true,
     }),
     new ControlDefinition({
       label: "Input Filetype",
@@ -96,6 +122,7 @@ export const PipelineFormDefinition = new FormDefinition({
         formElement: Select,
         elementProps: {},
       }),
+      required: true,
     }),
     new ControlDefinition({
       label: "Output Filetype",
@@ -106,6 +133,7 @@ export const PipelineFormDefinition = new FormDefinition({
         formElement: Select,
         elementProps: {},
       }),
+      required: true,
     }),
   ],
 });

web/src/components/createupdate/form-definitions/types/ControlDefinition.js

@@ -15,13 +15,15 @@ export default function ControlDefinition(props) {
     elementDefinition,
     options,
     defaultOption,
+    appearsWhen,
   } = props;
   this.id = id;
   this.label = label;
   this.constraintText = constraintText;
   this.elementDefinition = elementDefinition;
   this.options = options;
   this.defaultOption = defaultOption;
+  this.appearsWhen = appearsWhen;
 }
 
 ControlDefinition.propTypes = {
@@ -31,4 +33,5 @@ ControlDefinition.propTypes = {
   elementDefinition: PropTypes.instanceOf(ElementDefinition),
   options: PropTypes.arrayOf(OptionDefinition),
   defaultOption: PropTypes.instanceOf(OptionDefinition),
+  appearsWhen: PropTypes.array,
 };