Merge branch 'master' of https://github.com/simple-rules/harmony-benchmark into ricl-aws

# Conflicts:
#	aws-experiment-launch/create_and_deploy.py

aws-experiment-launch/create_and_deploy.py

@@ -4,6 +4,8 @@ import sys
 import json
 import time
 import datetime
+from threading import Thread
+from Queue import Queue
 import base64
 
 REGION_NAME = 'region_name'
@@ -24,7 +26,7 @@ time_stamp = time.time()
 CURRENT_SESSION = datetime.datetime.fromtimestamp(
     time_stamp).strftime('%H-%M-%S-%Y-%m-%d')
 PLACEMENT_GROUP = "PLACEMENT-" + CURRENT_SESSION
-NODE_VALUE = "NODE-" + CURRENT_SESSION
+NODE_NAME_SUFFIX = "NODE-" + CURRENT_SESSION
 
 """
 TODO:
@@ -39,13 +41,7 @@ Build (argparse,functions) support for
 
 """
 
-
+### CREATE INSTANCES ###
-def get_instance_ids(response):
-    instance_ids = []
-    for reservation in response["Reservations"]:
-        for instance in reservation["Instances"]:
-            instance_ids.append(instance["InstanceId"])
-    return instance_ids
 
 
 def run_one_region_instances(config, region_number, number_of_instances, isOnDemand=True):
@@ -57,100 +53,22 @@ def run_one_region_instances(config, region_number, number_of_instances, isOnDem
     session = boto3.Session(region_name=region_name)
     ec2_client = session.client('ec2')
     if isOnDemand:
-        response, placement = create_instances(
+        response = create_instances(
             config, ec2_client, region_number, int(number_of_instances))
     else:
-        response, placement = request_spots(
+        response = request_spots(
             config, ec2_client, region_number, int(number_of_instances))
-    print(placement)
+    return session
-    return session, placement
-
-
-def run_one_region_codedeploy(region_number, placement_group, commitId):
-    #todo: explore the use ec2 resource and not client. e.g. create_instances --  Might make for better code.
-    """
-    for getting instance ids:---
-    ec2 = boto3.resource('ec2', region_name=region_name])
-    result = ec2.instances.filter(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}])
-    for instance in result:
-        instances.append(instance.id)
-    
-    for getting public ips : --
-        ec2 = boto3.resource('ec2')
-        instance
-    """
-    region_name = config[region_number][REGION_NAME]
-    session = boto3.Session(region_name=region_name)
-    ec2_client = session.client('ec2')
-    response = ec2_client.describe_instances(
-        Filters=[
-            {
-                'Name': 'placement-group-name',
-                'Values': [
-                        placement_group
-                ]
-            }
-        ]
-    )
-    instance_ids = get_instance_ids(response)
-
-    print("Waiting for all instances to start running")
-    waiter = ec2_client.get_waiter('instance_running')
-    waiter.wait(InstanceIds=instance_ids)
-
-    print("Waiting for all instances to be status ok")
-    waiter = ec2_client.get_waiter('instance_status_ok')
-    waiter.wait(InstanceIds=instance_ids)
-
-    print("Waiting for system to be status ok")
-    waiter = ec2_client.get_waiter('system_status_ok')
-    waiter.wait(InstanceIds=instance_ids)
-
-    codedeploy = session.client('codedeploy')
-    application_name = APPLICATION_NAME
-    deployment_group = APPLICATION_NAME + "-" + str(commitId)
-    repo = REPO
-    response = get_application(codedeploy, application_name)
-    response = get_deployment_group(
-        codedeploy, application_name, deployment_group)
-    deploy(codedeploy, application_name,
-           deployment_group, repo, commitId, wait=True)
-    return response
-
-
-def get_availability_zones(ec2_client):
-    response = ec2_client.describe_availability_zones()
-    all_zones = []
-    if response.get('AvailabilityZones', None):
-       region_info = response.get('AvailabilityZones')
-       for info in region_info:
-           if info['State'] == 'available':
-               all_zones.append(info['ZoneName'])
-    return all_zones
-
-
-def get_one_availability_zone(ec2_client):
-    all_zones = get_availability_zones(ec2_client)
-    if len(all_zones) > 0:
-        return all_zones[0]
-    else:
-        print("No availability zone for this region")
-        sys.exit()
 
 
 def create_instances(config, ec2_client, region_number, number_of_instances):
-    placement_group = region_number + "-" + PLACEMENT_GROUP
+    NODE_NAME = region_number + "-" + NODE_NAME_SUFFIX
-    response = ec2_client.create_placement_group(
-        GroupName=placement_group,
-        Strategy='spread'
-    )
     response = ec2_client.run_instances(
         MinCount=number_of_instances,
         MaxCount=number_of_instances,
         ImageId=config[region_number][REGION_AMI],
         Placement={
             'AvailabilityZone': get_one_availability_zone(ec2_client),
-            'GroupName': placement_group
         },
         SecurityGroups=[config[region_number][REGION_SECURITY_GROUP]],
         IamInstanceProfile={
@@ -165,19 +83,13 @@ def create_instances(config, ec2_client, region_number, number_of_instances):
                 'Tags': [
                     {
                         'Key': 'Name',
-                        'Value': 'Node'
+                        'Value': NODE_NAME
                     },
                 ]
             },
-        ],
+        ]
-        # InstanceMarketOptions={
-        #     'MarketType': 'spot',
-        #     'SpotOptions': {
-        #         'BlockDurationMinutes': 60,
-        #     }
-        # }
     )
-    return response, placement_group
+    return response
 
 
 def request_spots(config, ec2_client, region_number, number_of_instances):
@@ -205,10 +117,81 @@ def request_spots(config, ec2_client, region_number, number_of_instances):
             }
         }
     )
-    return response, placement_group
+    return response
+
+
+def get_availability_zones(ec2_client):
+    response = ec2_client.describe_availability_zones()
+    all_zones = []
+    if response.get('AvailabilityZones', None):
+       region_info = response.get('AvailabilityZones')
+       for info in region_info:
+           if info['State'] == 'available':
+               all_zones.append(info['ZoneName'])
+    return all_zones
+
+
+def get_one_availability_zone(ec2_client):
+    all_zones = get_availability_zones(ec2_client)
+    if len(all_zones) > 0:
+        return all_zones[0]
+    else:
+        print("No availability zone for this region")
+        sys.exit()
+
+#### CODEDEPLOY ###
+
+
+def run_one_region_codedeploy(region_number, commitId):
+    #todo: explore the use ec2 resource and not client. e.g. create_instances --  Might make for better code.
+    """
+    for getting instance ids:---
+    ec2 = boto3.resource('ec2', region_name=region_name])
+    result = ec2.instances.filter(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}])
+    for instance in result:
+        instances.append(instance.id)
+    
+    for getting public ips : --
+        ec2 = boto3.resource('ec2')
+        instance
+    """
+    region_name = config[region_number][REGION_NAME]
+    NODE_NAME = region_number + "-" + NODE_NAME_SUFFIX
+    session = boto3.Session(region_name=region_name)
+    ec2_client = session.client('ec2')
+    filters = [
+        {
+            'Name': 'tag:Name',
+            'Values': [NODE_NAME]
+        }
+    ]
+
+    print("Waiting for all instances to start running")
+    waiter = ec2_client.get_waiter('instance_running')
+    waiter.wait(Filters=filters)
+
+    print("Waiting for all instances to be status ok")
+    waiter = ec2_client.get_waiter('instance_status_ok')
+    waiter.wait(Filters=filters)
+
+    print("Waiting for system to be status ok")
+    waiter = ec2_client.get_waiter('system_status_ok')
+    waiter.wait(Filters=filters)
+
+    codedeploy = session.client('codedeploy')
+    application_name = APPLICATION_NAME
+    deployment_group = APPLICATION_NAME + "-" + str(commitId)
+    repo = REPO
+    response = get_application(codedeploy, application_name)
+    response = get_deployment_group(
+        codedeploy, application_name, deployment_group)
+    depId = deploy(codedeploy, application_name,
+                   deployment_group, repo, commitId)
+    return region_number, depId
 
 
 def get_deployment_group(codedeploy, application_name, deployment_group):
+    NODE_NAME = region_number + "-" + NODE_NAME_SUFFIX
     response = codedeploy.list_deployment_groups(
         applicationName=application_name
     )
@@ -229,7 +212,7 @@ def get_deployment_group(codedeploy, application_name, deployment_group):
                     [
                         {
                             'Key': 'Name',
-                            'Value': 'Node',
+                            'Value': NODE_NAME,
                             'Type': 'KEY_AND_VALUE'
                         },
                     ],
@@ -239,14 +222,6 @@ def get_deployment_group(codedeploy, application_name, deployment_group):
         return response
 
 
-def get_commitId(commitId):
-    if commitId is None:
-        commitId = run("git rev-list --max-count=1 HEAD",
-                       hide=True).stdout.strip()
-        print("Got newest commitId as " + commitId)
-    return commitId
-
-
 def get_application(codedeploy, application_name):
     response = codedeploy.list_applications()
     if application_name in response['applications']:
@@ -259,7 +234,7 @@ def get_application(codedeploy, application_name):
     return response
 
 
-def deploy(codedeploy, application_name, deployment_group, repo, commitId, wait=True):
+def deploy(codedeploy, application_name, deployment_group, repo, commitId):
     """Deploy new code at specified revision to instance.
 
     arguments:
@@ -283,11 +258,8 @@ def deploy(codedeploy, application_name, deployment_group, repo, commitId, wait=
     )
     depId = res["deploymentId"]
     print("Deployment ID: " + depId)
-
     # The deployment is launched at this point, so exit unless asked to wait
     # until it finishes
-    if not wait:
-        return
     info = {'status': 'Created'}
     start = time.time()
     while info['status'] not in ('Succeeded', 'Failed', 'Stopped',) and (time.time() - start < 300.0):
@@ -296,9 +268,41 @@ def deploy(codedeploy, application_name, deployment_group, repo, commitId, wait=
         time.sleep(15)
     if info['status'] == 'Succeeded':
         print("\nDeploy Succeeded")
+        return depId
     else:
         print("\nDeploy Failed")
         print(info)
+        return depId
+
+
+def run_one_region_codedeploy_wrapper(region_number, commitId, queue):
+    region_number, depId = run_one_region_codedeploy(region_number, commitId)
+    queue.put((region_number, depId))
+
+
+def launch_code_deploy(region_list, commitId):
+    queue = Queue()
+    jobs = []
+    for i in range(len(region_list)):
+        region_number = region_list[i]
+        my_thread = Thread(target=run_one_region_codedeploy_wrapper, args=(
+            region_number, commitId, queue))
+        my_thread.start()
+        jobs.append(my_thread)
+    for my_thread in jobs:
+        my_thread.join()
+    results = [queue.get() for job in jobs]
+    return results
+
+##### UTILS ####
+
+
+def get_instance_ids(describe_instances_response):
+    instance_ids = []
+    for reservation in describe_instances_response["Reservations"]:
+        for instance in reservation["Instances"]:
+            instance_ids.append(instance["InstanceId"])
+    return instance_ids
 
 
 def read_configuration_file(filename):
@@ -316,6 +320,16 @@ def read_configuration_file(filename):
     return config
 
 
+def get_commitId(commitId):
+    if commitId is None:
+        commitId = run("git rev-list --max-count=1 HEAD",
+                       hide=True).stdout.strip()
+        print("Got newest commitId as " + commitId)
+    return commitId
+
+##### UTILS ####
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         description='This script helps you start instances across multiple regions')
@@ -334,16 +348,10 @@ if __name__ == "__main__":
     assert len(region_list) == len(instances_list), "number of regions: %d != number of instances per region: %d" % (
         len(region_list), len(intances_list))
     commitId = args.commitId
-    placement_groups = []
-
     for i in range(len(region_list)):
         region_number = region_list[i]
         number_of_instances = instances_list[i]
-        session, placement_group = run_one_region_instances(
+        session = run_one_region_instances(
-            config, region_number, number_of_instances)
+            config, region_number, number_of_instances, False)
-        placement_groups.append(placement_group)
+    results = launch_code_deploy(region_list, commitId)
-
+    print(results)
-    for i in range(len(region_list)):
-        region_number = region_list[i]
-        placement_group = placement_groups[i]
-        run_one_region_codedeploy(region_number, placement_group, commitId)