<commit_before>#!/usr/bin/python
import json
import sys
import time
import os
import logging
import botocore.session
import boto3
from .utils import (
    printlog,
    does_key_exist,
    create_jobid
)
from .vars import (
    AWS_REGION,
    S3_ACCESS_ARN,
    TIBANNA_REPO_NAME,
    TIBANNA_REPO_BRANCH,
    AMI_ID_WDL,
    AMI_ID_SHELL,
    AMI_ID_CWL_V1,
    AMI_ID_CWL_DRAFT3
)
from .exceptions import (
    EC2LaunchException,
    EC2InstanceLimitException,
    EC2InstanceLimitWaitException
)
from .nnested_array import flatten, run_on_nested_arrays1
from Benchmark import run as B

logger = logging.getLogger()
logger.setLevel(logging.INFO)

NONSPOT_EC2_PARAM_LIST = ['TagSpecifications', 'InstanceInitiatedShutdownBehavior',
                          'MaxCount', 'MinCount', 'DisableApiTermination']


def get_start_time():
    return time.strftime("%Y%m%d-%H:%M:%S-%Z")


def create_json_filename(jobid, json_dir):
    return json_dir + '/' + jobid + '.run.json'


def launch_and_get_instance_id(launch_args, jobid, spot_instance=None, spot_duration=None,
                               behavior_on_capacity_limit='fail'):
    try:  # capturing stdout from the launch command
        os.environ['AWS_DEFAULT_REGION'] = AWS_REGION
        ec2 = boto3.client('ec2')
    except Exception as e:
        raise Exception("Failed to create a client for EC2")

    if spot_instance:
        spot_options = {'SpotInstanceType': 'one-time',
                        'InstanceInterruptionBehavior': 'terminate'}
        if spot_duration:
            spot_options['BlockDurationMinutes'] = spot_duration
        launch_args.update({'InstanceMarketOptions': {'MarketType': 'spot',
                                                      'SpotOptions': spot_options}})
    try:
        res = 0
        res = ec2.run_instances(**launch_args)
    except Exception as e:
        if 'InsufficientInstanceCapacity' in str(e) or 'InstanceLimitExceeded' in str(e):
            if behavior_on_capacity_limit == 'fail':
                errmsg = "Instance limit exception - use 'behavior_on_capacity_limit' option"
                errmsg += "to change the behavior to wait_and_retry, or retry_without_spot. %s" % str(e)
                raise EC2InstanceLimitException(errmsg)
            elif behavior_on_capacity_limit == 'wait_and_retry':
                errmsg = "Instance limit exception - wait and retry later: %s" % str(e)
                raise EC2InstanceLimitWaitException(errmsg)
            elif behavior_on_capacity_limit == 'retry_without_spot':
                if not spot_instance:
                    errmsg = "'behavior_on_capacity_limit': 'retry_without_spot' works only with"
                    errmsg += "'spot_instance' : true. %s" % str(e)
                    raise Exception(errmsg)
                del(launch_args['InstanceMarketOptions'])
                try:
                    res = ec2.run_instances(**launch_args)
                    printlog("trying without spot : %s" % str(res))
                except Exception as e2:
                    errmsg = "Instance limit exception without spot instance %s" % str(e2)
                    raise EC2InstanceLimitException(errmsg)
        else:
            raise Exception("failed to launch instance for job {jobid}: {log}. %s"
                            .format(jobid=jobid, log=res) % e)

    try:
        instance_id = res['Instances'][0]['InstanceId']
    except Exception as e:
        raise Exception("failed to retrieve instance ID for job {jobid}".format(jobid=jobid))

    return instance_id


def read_config(CONFIG_FILE, CONFIG_KEYS):

    # 1. read .workflow.config.json file and get some variables
    with open(CONFIG_FILE, 'r') as f:
        cfg = json.load(f)

    # checking all the necessary keys exist
    for k in CONFIG_KEYS:
        if k not in cfg:
            sys.exit("The config file doesn't have key {}".format(k))

    return cfg


def create_json_dict(input_dict):
    # a is the final_args dictionary. json_dir is the output directory for the json file

    # create jobid here
    if 'jobid' in input_dict and input_dict.get('jobid'):
        jobid = input_dict.get('jobid')
    else:
        jobid = create_jobid()

    # start time
    start_time = get_start_time()

    a = input_dict.get('args')
    log_bucket = input_dict.get('config').get('log_bucket')

    # pre is a dictionary to be printed as a pre-run json file.
    pre = {'config': input_dict.get('config').copy()}  # copy only config since arg is redundant with 'Job'
    pre.update({'Job': {'JOBID': jobid,
                        'App': {
                                 'App_name': a.get('app_name', ''),
                                 'App_version': a.get('app_version', ''),
                                 'language': a.get('language', ''),
                                 'cwl_url': a.get('cwl_directory_url', ''),
                                 'main_cwl': a.get('cwl_main_filename', ''),
                                 'other_cwl_files': ','.join(a.get('cwl_child_filenames', [])),
                                 'wdl_url': a.get('wdl_directory_url', ''),
                                 'main_wdl': a.get('wdl_main_filename', ''),
                                 'other_wdl_files': ','.join(a.get('wdl_child_filenames', [])),
                                 'command': a.get('command', '')
                        },
                        'Input': {
                                 'Input_files_data': {},    # fill in later (below)
                                 'Secondary_files_data': {},   # fill in later (below)
                                 'Input_parameters': a.get('input_parameters', {}),
                                 'Env': a.get('input_env', {})
                        },
                        'Output': {
                                 'output_bucket_directory': a['output_S3_bucket'],
                                 'output_target': a['output_target'],
                                 'alt_cond_output_argnames': a.get('alt_cond_output_argnames', []),<commit_msg>continaer image support for run_task_awsem<commit_after>#!/usr/bin/python
import json
import sys
import time
import os
import logging
import botocore.session
import boto3
from .utils import (
    printlog,
    does_key_exist,
    create_jobid
)
from .vars import (
    AWS_REGION,
    S3_ACCESS_ARN,
    TIBANNA_REPO_NAME,
    TIBANNA_REPO_BRANCH,
    AMI_ID_WDL,
    AMI_ID_SHELL,
    AMI_ID_CWL_V1,
    AMI_ID_CWL_DRAFT3
)
from .exceptions import (
    EC2LaunchException,
    EC2InstanceLimitException,
    EC2InstanceLimitWaitException
)
from .nnested_array import flatten, run_on_nested_arrays1
from Benchmark import run as B

logger = logging.getLogger()
logger.setLevel(logging.INFO)

NONSPOT_EC2_PARAM_LIST = ['TagSpecifications', 'InstanceInitiatedShutdownBehavior',
                          'MaxCount', 'MinCount', 'DisableApiTermination']


def get_start_time():
    return time.strftime("%Y%m%d-%H:%M:%S-%Z")


def create_json_filename(jobid, json_dir):
    return json_dir + '/' + jobid + '.run.json'


def launch_and_get_instance_id(launch_args, jobid, spot_instance=None, spot_duration=None,
                               behavior_on_capacity_limit='fail'):
    try:  # capturing stdout from the launch command
        os.environ['AWS_DEFAULT_REGION'] = AWS_REGION
        ec2 = boto3.client('ec2')
    except Exception as e:
        raise Exception("Failed to create a client for EC2")

    if spot_instance:
        spot_options = {'SpotInstanceType': 'one-time',
                        'InstanceInterruptionBehavior': 'terminate'}
        if spot_duration:
            spot_options['BlockDurationMinutes'] = spot_duration
        launch_args.update({'InstanceMarketOptions': {'MarketType': 'spot',
                                                      'SpotOptions': spot_options}})
    try:
        res = 0
        res = ec2.run_instances(**launch_args)
    except Exception as e:
        if 'InsufficientInstanceCapacity' in str(e) or 'InstanceLimitExceeded' in str(e):
            if behavior_on_capacity_limit == 'fail':
                errmsg = "Instance limit exception - use 'behavior_on_capacity_limit' option"
                errmsg += "to change the behavior to wait_and_retry, or retry_without_spot. %s" % str(e)
                raise EC2InstanceLimitException(errmsg)
            elif behavior_on_capacity_limit == 'wait_and_retry':
                errmsg = "Instance limit exception - wait and retry later: %s" % str(e)
                raise EC2InstanceLimitWaitException(errmsg)
            elif behavior_on_capacity_limit == 'retry_without_spot':
                if not spot_instance:
                    errmsg = "'behavior_on_capacity_limit': 'retry_without_spot' works only with"
                    errmsg += "'spot_instance' : true. %s" % str(e)
                    raise Exception(errmsg)
                del(launch_args['InstanceMarketOptions'])
                try:
                    res = ec2.run_instances(**launch_args)
                    printlog("trying without spot : %s" % str(res))
                except Exception as e2:
                    errmsg = "Instance limit exception without spot instance %s" % str(e2)
                    raise EC2InstanceLimitException(errmsg)
        else:
            raise Exception("failed to launch instance for job {jobid}: {log}. %s"
                            .format(jobid=jobid, log=res) % e)

    try:
        instance_id = res['Instances'][0]['InstanceId']
    except Exception as e:
        raise Exception("failed to retrieve instance ID for job {jobid}".format(jobid=jobid))

    return instance_id


def read_config(CONFIG_FILE, CONFIG_KEYS):

    # 1. read .workflow.config.json file and get some variables
    with open(CONFIG_FILE, 'r') as f:
        cfg = json.load(f)

    # checking all the necessary keys exist
    for k in CONFIG_KEYS:
        if k not in cfg:
            sys.exit("The config file doesn't have key {}".format(k))

    return cfg


def create_json_dict(input_dict):
    # a is the final_args dictionary. json_dir is the output directory for the json file

    # create jobid here
    if 'jobid' in input_dict and input_dict.get('jobid'):
        jobid = input_dict.get('jobid')
    else:
        jobid = create_jobid()

    # start time
    start_time = get_start_time()

    a = input_dict.get('args')
    log_bucket = input_dict.get('config').get('log_bucket')

    # pre is a dictionary to be printed as a pre-run json file.
    pre = {'config': input_dict.get('config').copy()}  # copy only config since arg is redundant with 'Job'
    pre.update({'Job': {'JOBID': jobid,
                        'App': {
                                 'App_name': a.get('app_name', ''),
                                 'App_version': a.get('app_version', ''),
                                 'language': a.get('language', ''),
                                 'cwl_url': a.get('cwl_directory_url', ''),
                                 'main_cwl': a.get('cwl_main_filename', ''),
                                 'other_cwl_files': ','.join(a.get('cwl_child_filenames', [])),
                                 'wdl_url': a.get('wdl_directory_url', ''),
                                 'main_wdl': a.get('wdl_main_filename', ''),
                                 'other_wdl_files': ','.join(a.get('wdl_child_filenames', [])),
                                 'command': a.get('command', ''),
                                 'container_image': a.get('container_image', '')
                        },
                        'Input': {
                                 'Input_files_data': {},    # fill in later (below)
                                 'Secondary_files_data': {},   # fill in later (below)
                                 'Input_parameters': a.get('input_parameters', {}),
                                 'Env': a.get('input_env', {})
                        },
                        'Output': {
                                 'output_bucket_directory': a['output_S3_bucket'],
                                 'output_target': a['output_target'],
                                 'alt_cond_output_argnames': a.get('alt_cond_output_argnames', []),
{'fraction_non_alphanumeric': 0.1062076841504201, 'fraction_numerical': 0.0039154906599233215, 'mean_word_length': 1.961352657004831, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 39, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6113010', 'n_tokens_mistral': 3556, 'n_tokens_neox': 3449, 'n_words': 794}