various fixes

awsf3-docker/run.sh

@@ -27,8 +27,9 @@ done
 
 export RUN_JSON_FILE_NAME=$JOBID.run.json
 export POSTRUN_JSON_FILE_NAME=$JOBID.postrun.json
-export EBS_DIR=/data1  ## WARNING: also hardcoded in aws_decode_run_json.py
-export LOCAL_OUTDIR=$EBS_DIR/out  
+export EBS_DIR=/data1
+export MOUNT_DIR_PREFIX=/mnt
+export LOCAL_OUTDIR=$EBS_DIR/out
 export LOCAL_INPUT_DIR=$EBS_DIR/input  ## WARNING: also hardcoded in aws_decode_run_json.py
 export LOCAL_WF_TMPDIR=$EBS_DIR/tmp
 export MD5FILE=$JOBID.md5sum.txt
@@ -71,7 +72,7 @@ fi
 
 # EBS_DIR cannot be directly mounted to docker container since it's already a mount point for EBS,
 # so mount /mnt/data1/ instead and create a symlink.
-ln -s /mnt/$EBS_DIR $EBS_DIR
+ln -s $MOUNT_DIR_PREFIX/$EBS_DIR $EBS_DIR
 
 # Transferring profile info
 ln -s /home/ubuntu/.aws /root/.aws
@@ -152,12 +153,12 @@ send_log
 ### download data & reference files from s3
 exl echo
 exl echo "## Downloading data & reference files from S3"
-exl date 
+exl date
 exl mkdir -p $LOCAL_INPUT_DIR
 exl cat $DOWNLOAD_COMMAND_FILE
-exle source $DOWNLOAD_COMMAND_FILE 
+exle source $DOWNLOAD_COMMAND_FILE
 exl date
-send_log 
+send_log
 
 
 ### mount input buckets
@@ -198,7 +199,7 @@ exl echo "## $(docker info | grep 'Total Memory')"
 exl echo
 send_log
 cwd0=$(pwd)
-cd $LOCAL_WFDIR  
+cd $LOCAL_WFDIR
 mkdir -p $LOCAL_WF_TMPDIR
 if [[ $LANGUAGE == 'wdl_v1' || $LANGUAGE == 'wdl' ]]
 then
@@ -227,13 +228,15 @@ else
     exl echo "Error: CWL draft3 is no longer supported. Please switch to v1"
     handle_error 1
   fi
-  exlj cwltool --enable-dev --non-strict --no-read-only --no-match-user --outdir $LOCAL_OUTDIR --tmp-outdir-prefix $LOCAL_WF_TMPDIR --tmpdir-prefix $LOCAL_WF_TMPDIR $PRESERVED_ENV_OPTION $SINGULARITY_OPTION $MAIN_CWL $cwd0/$INPUT_YML_FILE
+  # cwltool cannot recognize symlinks and end up copying output from tmp directory intead of moving.
+  # To prevent this, use the original directory name here.
+  exlj cwltool --enable-dev --non-strict --no-read-only --no-match-user --outdir $MOUNT_DIR_PREFIX$LOCAL_OUTDIR --tmp-outdir-prefix $MOUNT_DIR_PREFIX$LOCAL_WF_TMPDIR --tmpdir-prefix $MOUNT_DIR_PREFIX$LOCAL_WF_TMPDIR $PRESERVED_ENV_OPTION $SINGULARITY_OPTION $MAIN_CWL $cwd0/$INPUT_YML_FILE
   handle_error $?
 fi
 cd $cwd0
 exl echo
 exl echo "Finished running the command/workflow"
-send_log 
+send_log
 
 ### copy output files to s3
 exl echo
@@ -282,14 +285,14 @@ fi
 exl awsf3 update_postrun_json_upload_output -i $POSTRUN_JSON_FILE_NAME $LOGJSON_OPTION -m $LOCAL_OUTDIR/$MD5FILE -o $POSTRUN_JSON_FILE_NAME -L $LANGUAGE
 exl awsf3 upload_postrun_json -i $POSTRUN_JSON_FILE_NAME
 send_log
- 
+
 ### updating status
 exl echo
 exl echo "## Checking the job status (0 means success)"
 ## if STATUS is 21,0,0,1 JOB_STATUS is 21,0,0,1. If STATUS is 0,0,0,0,0,0, JOB_STATUS is 0.
 if [ $(echo $STATUS| sed 's/0//g' | sed 's/,//g') ]; then export JOB_STATUS=$STATUS ; else export JOB_STATUS=0; fi
 exl echo "JOB_STATUS=$JOB_STATUS"
-# This env variable (JOB_STATUS) will be read by aws_update_run_json.py and the result will go into $POSTRUN_JSON_FILE_NAME. 
+# This env variable (JOB_STATUS) will be read by aws_update_run_json.py and the result will go into $POSTRUN_JSON_FILE_NAME.
 
 # update & upload postrun json
 exl echo

tibanna/__main__.py

@@ -157,7 +157,6 @@ def args(self):
                  {'flag': ["-d", "--shutdown-min"],
                   'help': "use a specified shutdown mininutes for the rerun"},
                  {'flag': ["-b", "--ebs-size"],
-                  'type': int,
                   'help': "use a specified ebs size for the rerun (GB)"},
                  {'flag': ["-T", "--ebs-type"],
                   'help': "use a specified ebs type for the rerun (gp2 vs io1)"},
@@ -203,7 +202,6 @@ def args(self):
                  {'flag': ["-d", "--shutdown-min"],
                   'help': "use a specified shutdown mininutes for the rerun"},
                  {'flag': ["-b", "--ebs-size"],
-                  'type': int,
                   'help': "use a specified ebs size for the rerun (GB)"},
                  {'flag': ["-T", "--ebs-type"],
                   'help': "use a specified ebs type for the rerun (gp2 vs io1)"},

tibanna/_version.py

@@ -1,4 +1,4 @@
 """Version information."""
 
 # The following line *must* be the last in the module, exactly as formatted:
-__version__ = "1.0.4"
+__version__ = "1.0.5"

tibanna/core.py

@@ -677,7 +677,10 @@ def rerun(self, exec_arn=None, job_id=None, sfn=None,
             if shutdown_min:
                 override_config['shutdown_min'] = shutdown_min
             if ebs_size:
-                override_config['ebs_size'] = int(ebs_size)
+                if ebs_size.endswith('x'):
+                    override_config['ebs_size'] = ebs_size
+                else:
+                    override_config['ebs_size'] = int(ebs_size)
             if overwrite_input_extra:
                 override_config['overwrite_input_extra'] = overwrite_input_extra
             if key_name:
@@ -1221,4 +1224,3 @@ def create_ami(self, build_from_scratch=True, source_image_to_copy_from=None, so
                 args.update({'base_region': source_image_region})
 
         return AMI(**args).create_ami_for_tibanna(make_public=make_public)
-

tibanna/iam_utils.py

@@ -7,6 +7,7 @@
     AWS_ACCOUNT_NUMBER,
     AWS_REGION,
     LAMBDA_TYPE,
+    SFN_TYPE,
     RUN_TASK_LAMBDA_NAME,
     CHECK_TASK_LAMBDA_NAME
 )
@@ -20,6 +21,7 @@ class IAM(object):
     account_id = AWS_ACCOUNT_NUMBER
     region = AWS_REGION
     lambda_type = LAMBDA_TYPE  # lambda_type : '' for unicorn, 'pony' for pony, 'zebra' for zebra
+    sfn_type = SFN_TYPE  # sfn type : 'unicorn' for unicorn, 'pony' for pony, 'zebra' for zebra
     run_task_lambda_name = RUN_TASK_LAMBDA_NAME
     check_task_lambda_name = CHECK_TASK_LAMBDA_NAME
 
@@ -48,6 +50,7 @@ def __init__(self, user_group_tag, bucket_names='', no_randomize=True):
             random_tag = str(int(random.random() * 10000))
             self.user_group_name = self.user_group_tag + '_' + random_tag
         self.tibanna_policy_prefix = self.prefix + self.user_group_name
+        self.tibanna_sfn_name = self.prefix + self.sfn_type + '_' + self.user_group_name
 
         # bucket names
         self.bucket_names = bucket_names
@@ -63,7 +66,8 @@ def iam_group_name(self):
     @property
     def policy_types(self):
         return ['bucket', 'termination', 'list', 'cloudwatch', 'passrole', 'lambdainvoke',
-                'desc_stepfunction', 'cloudwatch_metric', 'cw_dashboard', 'dynamodb', 'ec2_desc']
+                'cloudwatch_metric', 'cw_dashboard', 'dynamodb', 'ec2_desc',
+                'executions']
 
     def policy_arn(self, policy_type):
         return 'arn:aws:iam::' + self.account_id + ':policy/' + self.policy_name(policy_type)
@@ -75,11 +79,11 @@ def policy_suffix(self, policy_type):
                     'cloudwatch': 'cloudwatchlogs',
                     'passrole': 'iam_passrole_s3',
                     'lambdainvoke': 'lambdainvoke',
-                    'desc_stepfunction': 'desc_sts',
                     'cloudwatch_metric': 'cw_metric',
                     'cw_dashboard': 'cw_dashboard',
                     'dynamodb': 'dynamodb',
-                    'ec2_desc': 'ec2_desc'}
+                    'ec2_desc': 'ec2_desc',
+                    'executions': 'executions'}
         if policy_type not in suffices:
             raise Exception("policy %s must be one of %s." % (policy_type, str(self.policy_types)))
         return suffices[policy_type]
@@ -94,11 +98,11 @@ def policy_definition(self, policy_type):
                        'cloudwatch': self.policy_cloudwatchlogs,
                        'passrole': self.policy_iam_passrole_s3,
                        'lambdainvoke': self.policy_lambdainvoke,
-                       'desc_stepfunction': self.policy_desc_stepfunction,
                        'cloudwatch_metric': self.policy_cloudwatch_metric,
                        'cw_dashboard': self.policy_cw_dashboard,
                        'dynamodb': self.policy_dynamodb,
-                       'ec2_desc': self.policy_ec2_desc_policy}
+                       'ec2_desc': self.policy_ec2_desc_policy,
+                       'executions': self.policy_executions}
         if policy_type not in definitions:
             raise Exception("policy %s must be one of %s." % (policy_type, str(self.policy_types)))
         return definitions[policy_type]
@@ -128,7 +132,7 @@ def role_service(self, role_type):
 
     def policy_arn_list_for_role(self, role_type):
         run_task_custom_policy_types = ['list', 'cloudwatch', 'passrole', 'bucket', 'dynamodb',
-                                        'desc_stepfunction', 'cw_dashboard']
+                                        'executions', 'cw_dashboard']
         check_task_custom_policy_types = ['cloudwatch_metric', 'cloudwatch', 'bucket', 'ec2_desc',
                                           'termination', 'dynamodb']
         arnlist = {'ec2': [self.policy_arn(_) for _ in ['bucket', 'cloudwatch_metric']] +
@@ -268,18 +272,32 @@ def policy_lambdainvoke(self):
         return policy
 
     @property
-    def policy_desc_stepfunction(self):
-        execution_arn_prefix = 'arn:aws:states:' + self.region + ':' + self.account_id + ':execution:'
-        resource = execution_arn_prefix + self.tibanna_policy_prefix + ':*'
+    def policy_executions(self):
+        execution_arn_prefix = 'arn:aws:states:' + self.region + ':' + self.account_id + ':stateMachine:'
+        sfn_arn_prefix = 'arn:aws:states:' + self.region + ':' + self.account_id + ':execution:'
+        resources = [execution_arn_prefix + self.tibanna_sfn_name,
+                    sfn_arn_prefix + self.tibanna_sfn_name + ':*']
         policy = {
             "Version": "2012-10-17",
             "Statement": [
                 {
                     "Effect": "Allow",
                     "Action": [
-                        "states:DescribeExecution"
+                        "states:StartExecution",
+                        "states:StopExecution",
+                        "states:ListExecutions",
+                        "states:DescribeExecution",
+                        "states:GetExecutionHistory",
+                        "states:DescribeStateMachineForExecution",
+                        "states:DescribeStateMachine"
                     ],
-                    "Resource": resource
+                    "Resource": resources
+                },
+                {
+                    "Sid": "VisualEditor1",
+                    "Effect": "Allow",
+                    "Action": "states:ListStateMachines",
+                    "Resource": "*"
                 }
             ]
         }
@@ -421,16 +439,6 @@ def create_user_group(self, verbose=False):
                 # deleting a group would require users to be detached from the group.
                 self.detach_policies_from_group()
         group = self.iam.Group(self.iam_group_name)
-        response = group.attach_policy(
-            PolicyArn='arn:aws:iam::aws:policy/AWSStepFunctionsFullAccess'
-        )
-        if verbose:
-            logger.debug("response from IAM attach_policy :" + str(response))
-        response = group.attach_policy(
-            PolicyArn='arn:aws:iam::aws:policy/AWSStepFunctionsConsoleFullAccess'
-        )
-        if verbose:
-            logger.debug("response from IAM attach_policy :" + str(response))
         response = group.attach_policy(
             PolicyArn='arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
         )
@@ -441,7 +449,8 @@ def create_user_group(self, verbose=False):
         )
         if verbose:
             logger.debug("response from IAM attach_policy :" + str(response))
-        custom_policy_types = ['bucket', 'ec2_desc', 'cloudwatch_metric', 'dynamodb', 'termination']
+        custom_policy_types = ['bucket', 'ec2_desc', 'cloudwatch_metric', 'dynamodb',
+                               'termination', 'executions']
         for pn in [self.policy_name(pt) for pt in custom_policy_types]:
             response = group.attach_policy(
                 PolicyArn='arn:aws:iam::' + self.account_id + ':policy/' + pn