Updated AWS stuff. start_worker code updated

Author: yadudoc

aws/aws.py

@@ -4,6 +4,7 @@
 import configurator
 import sys
 import random
+import logging
 
 from libcloud.compute.types import Provider
 from libcloud.compute.providers import get_driver
@@ -18,6 +19,8 @@
                NodeState.PENDING    : "PENDING",
                NodeState.UNKNOWN    : "UNKNOWN" }
 
+logging.basicConfig(filename='aws.log', level=logging.INFO)
+
 def list_resources(driver):
 
     resources = driver.list_nodes()
@@ -48,45 +51,55 @@ def get_public_ip(driver, name):
     print "Could not find a resource with the id ", name
     return -1
 
-def create_security_group(driver, configs):
+def aws_create_security_group(driver, configs):
     group_name = configs["SECURITY_GROUP"]
     current    = driver.ex_list_security_groups()
     if group_name in current:
-        print "INFO: Security group: ", group_name, " is already present"
+        logging.info("Security group: %s is already present", group_name)
     else:
-        print "INFO: Creating new security group: ", group_name
+        logging.info("Creating new security group: %s", group_name)
         res = driver.ex_create_security_group(name=group_name,description="Open all ports")
         if not driver.ex_authorize_security_group(group_name, 0, 65000, '0.0.0.0/0'):
-            print "INFO: Authorizing ports for security group failed"
+            logging.info("Authorizing ports for security group failed")
         if not driver.ex_authorize_security_group(group_name, 0, 65000, '0.0.0.0/0', protocol='udp'):
-            print "INFO: Authorizing ports for security group failed"
-        print res
+            logging.info("Authorizing ports for security group failed")
+        logging.debug("Security group: %s", str(res))
 
 def check_keypair(driver, configs):
     if "AWS_KEYPAIR_NAME" in configs and "AWS_KEYPAIR_FILE" in configs:
-        print "AWS_KEYPAIR_NAME : ", configs['AWS_KEYPAIR_NAME']
-        print "AWS_KEYPAIR_FILE : ", configs['AWS_KEYPAIR_FILE']
+        logging.debug("AWS_KEYPAIR_NAME : %s", configs['AWS_KEYPAIR_NAME'])
+        logging.debug("AWS_KEYPAIR_FILE : %s", configs['AWS_KEYPAIR_FILE'])
         all_pairs = driver.list_key_pairs()
         for pair in all_pairs:
             if pair.name == configs['AWS_KEYPAIR_NAME']:
-                print "INFO: KEYPAIR exists, registered"
+                logging.info("KEYPAIR exists, registered")
                 return 0
 
-        print "INFO : KEYPAIR does not exist. Creating keypair"
+        logging.info("KEYPAIR does not exist. Creating keypair")
         key_pair = driver.create_key_pair(name=configs['AWS_KEYPAIR_NAME'])
         f = open(configs['AWS_KEYPAIR_FILE'], 'w')
         f.write(str(key_pair.private_key) + '\n')
         f.close()
         os.chmod(configs['AWS_KEYPAIR_FILE'], 0600)
+        logging.info("KEYPAIR created")
     else:
-        print "AWS_KEYPAIR_NAME and/or AWS_KEYPAIR_FILE missing"
-        print "ERROR: Cannot proceed without AWS_KEYPAIR_NAME and AWS_KEYPAIR_FILE"
+        logging.error("AWS_KEYPAIR_NAME and/or AWS_KEYPAIR_FILE missing")
+        logging.error("Cannot proceed without AWS_KEYPAIR_NAME and AWS_KEYPAIR_FILE")
         exit(-1)
 
 
 def start_headnode(driver, configs):
     userdata   = configurator.getstring("headnode")
 
+    # Check if headnode
+    nodes      = driver.list_nodes()
+    headnode   = False
+    for node in nodes:
+        if node.name == "headnode" and node.state == NodeState.RUNNING:
+            headnode = node
+            print "INFO: Headnode is RUNNING"
+            return 0
+
     size       = NodeSize(id=configs['HEADNODE_MACHINE_TYPE'], name='headnode',
                           ram=None, disk=None, bandwidth=None, price=None, driver=driver)
     image      = NodeImage(id=configs['HEADNODE_IMAGE'], name=None, driver=driver)
@@ -96,7 +109,9 @@ def start_headnode(driver, configs):
                                     ex_keyname=configs['AWS_KEYPAIR_NAME'],
                                     ex_securitygroup=configs['SECURITY_GROUP'],
                                     ex_userdata=userdata )
+    print "INFO: Waiting for headnode bootup ..."
     driver._wait_until_running(node, wait_period=5, timeout=240)
+    print "INFO: Headnode active!"
     if node.public_ips:
         print '-'*51
         print '{0:20} | {1:10} | {2:15}'.format(node.name, NODESTATES[node.state], node.public_ips[0])
@@ -114,10 +129,10 @@ def start_worker(driver, configs, worker_names):
         return -1
 
     # Setup userdata
-    userdata   = configurator.getstring("headnode")
+    userdata   = configurator.getstring("worker")
     userdata   = userdata.replace("SET_HEADNODE_IP", headnode.public_ips[0])
-
-    print userdata
+    logging.info("Headnode connection url : ",headnode.public_ips[0])
+    logging.debug("Worker userdata : %s", userdata)
 
     list_nodes = []
     for worker_name in worker_names:
@@ -131,7 +146,7 @@ def start_worker(driver, configs, worker_names):
                                         ex_securitygroup=configs['SECURITY_GROUP'],
                                         ex_userdata=userdata )
         list_nodes.append(node)
-    print list_nodes
+        logging.info("Worker node started : %s",str(node))
 
 
 def terminate_all_nodes():
@@ -156,7 +171,7 @@ def init():
     #configurator.pretty_configs(configs)
     driver     = get_driver(Provider.EC2_US_WEST_OREGON) # was EC2
     ec2_driver = driver(configs['AWSAccessKeyId'], configs['AWSSecretKey'])
-    create_security_group(ec2_driver, configs)
+    aws_create_security_group(ec2_driver, configs)
     check_keypair(ec2_driver, configs)
     return configs,ec2_driver
 

aws/setup.sh

@@ -8,10 +8,10 @@ check_project ()
     RESULT=$(gcutil listinstances --project=$GCE_PROJECTID 2>&1)
 }
 
-
 start_worker ()
 {
-    echo "Starting worker - TODO"
+    echo "Starting worker(s) $*"
+    ./aws.py start_worker $*
 }
 
 # Is this valid for AWS ?
@@ -22,13 +22,6 @@ check_keys ()
     [[ ! -f ~/.ssh/google_compute_engine.pub  ]] && echo "Google public key missing"  && return
 }
 
-
-stop_workers()
-{
-    echo "Stopping all instances"
-}
-
-
 stop_n_workers()
 {
     COUNT=1
@@ -43,27 +36,25 @@ stop_n_workers()
 # This script ensures that only the specified number of workers are active
 start_n_workers ()
 {
+    echo "In start_n_workers"
     COUNT=$1
     CURRENT=1
-    out=$(gcutil --project=$GCE_PROJECTID listinstances | grep "swift-worker")
+    out=$(list_resources | grep "swift-worker")
     if [[ "$?" == 0 ]]
     then
         echo "Current workers"
         echo "${out[*]}"
-        CURRENT=$(gcutil --project=$GCE_PROJECTID listinstances | grep "swift-worker" | wc -l)
+        CURRENT=$(list_resources | grep "swift-worker" | wc -l)
         echo "Count : " $CURRENT
         echo "New workers needed : $(($COUNT - $CURRENT))"
     fi
 
     for i in $(seq $CURRENT 1 $COUNT)
     do
-        start_worker $i &> $LOG &
+        start_worker swift-worker-$i &> $LOG &
     done
     wait
-    gcutil --project=$GCE_PROJECTID listinstances
-    echo "Updating WORKER_HOSTS"
-    EXTERNAL_IPS=$(gcutil --project=$GCE_PROJECTID listinstances | grep worker | awk '{print $10}')
-    WORKER_NAMES=$(gcutil --project=$GCE_PROJECTID listinstances | grep worker | awk '{print $2}')
+    list_resources
 }
 
 start_n_more ()
@@ -125,6 +116,28 @@ dissolve()
     ./aws.py dissolve
 }
 
+start_headnode()
+{
+    ./aws.py start_headnode
+}
+
+start_workers()
+{
+    WORKERS_REQUESTED=$AWS_WORKER_COUNT
+    CURRENT_COUNT=$(list_resources | grep "swift-worker" | wc -l)
+    echo "Current workers   : $CURRENT_COUNT"
+    echo "Workers requested : $WORKERS_REQUESTED"
+    WORKERS_REQUIRED=$(($WORKERS_REQUESTED - $CURRENT_COUNT))
+    if [[ $WORKERS_REQUIRED -gt 0 ]]
+    then
+        #printf("swift-worker-%03d", {$CURRENT_COUNT..$(($CURRENT_COUNT+$COUNT_NEEDED))})
+        END=$(($CURRENT_COUNT+$WORKERS_REQUIRED-1))
+        start_worker $(printf "swift-worker-%03d " $(seq $CURRENT_COUNT 1 $END))
+    else
+        echo "No additional workers needed"
+    fi
+}
+
 connect()
 {
     source configs
@@ -137,8 +150,12 @@ connect()
     ssh -A -o StrictHostKeyChecking=no -l $AWS_USERNAME -i $AWS_KEYPAIR_FILE $IP
 }
 
+
 init()
 {
     source configs
+    start_headnode
+    start_workers
+    list_resources
 }
 init

swift-localhost-tutorial/app/simulate

@@ -0,0 +1 @@
+simulate.sh

swift-localhost-tutorial/app/simulate.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python
+import sys
+import os
+import socket
+import time
+import math
+import random
+import getpass
+
+def parse():
+    from optparse import OptionParser
+    parser=OptionParser()
+    parser.add_option("-b", "--bias", type="int", dest="bias", default=0,
+                      help="offset bias: add this integer to all results");
+
+    parser.add_option("-B", "--biasfile", type="string", dest="biasfile",
+                      default="none", help="file of integer biases to add to results" );
+
+    parser.add_option("-l", "--log", type="string", dest="log", default="yes",
+                      help="generate a log in stderr if not nul");
+
+    parser.add_option("-n", "--nvalues", type="int", dest="nvalues",default=1,
+                      help="print this many values per simulation" );
+
+    parser.add_option("-s", "--seed", type="int", dest="initseed", default=0,
+                      help="use this integer [0..32767] as a seed");
+
+    parser.add_option("-S", "--seedfile", type="string", dest="seedfile",
+                      default="none", help="use this file (containing integer seeds [0..32767]) one per line" );
+
+    parser.add_option("-t", "--timesteps", type="int", dest="timesteps",
+                      default=0, help='number of simulated "timesteps" in seconds (determines runtime)' );
+
+    parser.add_option("-r", "--range", type="int", dest="range", default=100,
+                      help="range (limit) of generated results" );
+
+    parser.add_option("-w", "--width", type="int", dest="width", default=8,
+                      help="Width ?" );
+
+    parser.add_option("-x", "--scale", type="int", dest="scale", default=1,
+                      help="scale the results by this integer" );
+    # Not implemented yet
+    parser.add_option("-p", "--paramfile", type="string", dest="paramfile", default="none",
+                      help="Not implemented yet" );
+    return (parser.parse_args());
+
+def log():
+    import datetime
+    print >> sys.stderr, "Called as: ", str(sys.argv)
+    print >> sys.stderr, "Start time: ", datetime.datetime.now()
+    print >> sys.stderr, "Running on node: ", socket.gethostname()
+    print >> sys.stderr, "Running as user: ", getpass.getuser()
+    print >> sys.stderr, "\nEnvironment:\n\n"
+    print >> sys.stderr, os.environ
+
+def printparams(options):
+    print 'bias =',options.bias
+    print 'biasfile = ',options.biasfile
+    print 'log = ', options.log
+    print 'nvalues = ',options.nvalues
+    print 'seed = ', options.initseed
+    print 'seedfile = ', options.seedfile
+    print 'timesteps = ', options.timesteps
+    print 'range = ', options.range
+    print 'width = ', options.width
+    print 'scale = ', options.scale
+    print 'paramfile = ', options.paramfile
+
+def simulate(options):
+    time.sleep(options.timesteps)
+    bias=[];
+    if (options.biasfile != "none"):
+        try:
+            with open(options.biasfile) as biasfile:
+                lines = biasfile.read().splitlines()
+                for line in lines:
+                    bias.append(int(line))
+        except IOError:
+            print "Error accessing content from file: ", options.biasfile 
+    bias_count = len(bias)
+
+    for i in range(options.nvalues):
+        value = (random.random() +
+        random.random()*math.pow(2,16) +
+        random.random()*math.pow(2,32) +
+        random.random()*math.pow(2,48))
+        value=( (int(value)%options.range) * options.scale + options.bias)
+        if ( i < bias_count ):
+            value = value + bias[i]
+        elif ( bias_count > 0 ):
+            value = value + bias[bias_count-1]
+
+        print '{num:{fill}{width}}'.format(num=value, fill=' ', width=options.width)
+
+
+def seed(options):
+    if (options.initseed != 0 ):
+        random.seed(options.initseed)
+    if (options.seedfile != "none"):
+        try:
+            with open(options.seedfile) as seedfile:
+                lines=seedfile.read().splitlines()
+                seed = 0
+                for line in lines:
+                    seed = seed + int(line)
+            random.seed(seed)
+        except IOError:
+            print "Could not open file: ", options.seedfile
+
+
+if __name__ == "__main__":
+    (options, args) = parse()
+    printparams(options)
+    seed(options)
+    simulate(options)
+    log()
+
+
+
+
+
+
+
+
+