From 6f3d269f828e519c61913393d6e7ef504100520d Mon Sep 17 00:00:00 2001 From: Richard Liu Date: Thu, 5 Jul 2018 00:19:44 -0700 Subject: [PATCH 1/3] added spot_fleet. --- aws-experiment-launch/create_and_deploy.py | 72 ++++++++++++++++++++-- 1 file changed, 66 insertions(+), 6 deletions(-) diff --git a/aws-experiment-launch/create_and_deploy.py b/aws-experiment-launch/create_and_deploy.py index f9e1d000c..c03d195a7 100644 --- a/aws-experiment-launch/create_and_deploy.py +++ b/aws-experiment-launch/create_and_deploy.py @@ -8,11 +8,18 @@ from threading import Thread from Queue import Queue import base64 + +class InstanceType: + ON_DEMAND = 1 + SPOT_INSTANCE = 2 + SPOT_FLEET = 3 + + REGION_NAME = 'region_name' REGION_KEY = 'region_key' REGION_SECURITY_GROUP = 'region_security_group' REGION_HUMAN_NAME = 'region_human_name' -INSTANCE_TYPE = 't2.micro' +INSTANCE_TYPE = 't2.small' REGION_AMI = 'region_ami' # USER_DATA = 'user-data.sh' # UserData must be base64 encoded for spot instances. @@ -44,7 +51,7 @@ Build (argparse,functions) support for ### CREATE INSTANCES ### -def run_one_region_instances(config, region_number, number_of_instances, isOnDemand=True): +def run_one_region_instances(config, region_number, number_of_instances, instance_type=InstanceType.ON_DEMAND): #todo: explore the use ec2 resource and not client. e.g. create_instances -- Might make for better code. """ e.g. ec2.create_instances @@ -52,11 +59,14 @@ def run_one_region_instances(config, region_number, number_of_instances, isOnDem region_name = config[region_number][REGION_NAME] session = boto3.Session(region_name=region_name) ec2_client = session.client('ec2') - if isOnDemand: + if instance_type == InstanceType.ON_DEMAND: response = create_instances( config, ec2_client, region_number, int(number_of_instances)) + elif instance_type == InstanceType.SPOT_INSTANCE: + response = request_spot_instances( + config, ec2_client, region_number, int(number_of_instances)) else: - response = request_spots( + response = request_spot_fleet( config, ec2_client, region_number, int(number_of_instances)) return session @@ -102,7 +112,8 @@ def create_instances(config, ec2_client, region_number, number_of_instances): return response -def request_spots(config, ec2_client, region_number, number_of_instances): +def request_spot_instances(config, ec2_client, region_number, number_of_instances): + NODE_NAME = region_number + "-" + NODE_NAME_SUFFIX response = ec2_client.request_spot_instances( # DryRun=True, BlockDurationMinutes=60, @@ -124,6 +135,55 @@ def request_spots(config, ec2_client, region_number, number_of_instances): return response +def request_spot_fleet(config, ec2_client, region_number, number_of_instances): + NODE_NAME = region_number + "-" + NODE_NAME_SUFFIX + # https://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.Client.request_spot_fleet + response = ec2_client.request_spot_fleet( + # DryRun=True, + SpotFleetRequestConfig={ + # https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-fleet.html#spot-fleet-allocation-strategy + 'AllocationStrategy': 'diversified', + # 'IamFleetRole': IAM_INSTANCE_PROFILE, // TODO@ricl, create fleet role. + 'LaunchSpecifications': [ + { + 'SecurityGroups': [ + { + 'GroupName': config[region_number][REGION_SECURITY_GROUP] + } + ], + 'IamInstanceProfile': { + 'Name': IAM_INSTANCE_PROFILE + }, + 'ImageId': config[region_number][REGION_AMI], + 'InstanceType': INSTANCE_TYPE, + 'KeyName': config[region_number][REGION_KEY], + 'Placement': { + 'AvailabilityZone': get_one_availability_zone(ec2_client) + }, + 'UserData': USER_DATA, + # 'WeightedCapacity': 123.0, + 'TagSpecifications': [ + { + 'ResourceType': 'instance', + 'Tags': [ + { + 'Key': 'Name', + 'Value': NODE_NAME + }, + ] + } + ] + }, + ], + # 'SpotPrice': 'string', # The maximum price per unit hour that you are willing to pay for a Spot Instance. The default is the On-Demand price. + 'TargetCapacity': 1, + 'OnDemandTargetCapacity': 0, + 'Type': 'maintain', + } + ) + return response + + def get_availability_zones(ec2_client): response = ec2_client.describe_availability_zones() all_zones = [] @@ -356,6 +416,6 @@ if __name__ == "__main__": region_number = region_list[i] number_of_instances = instances_list[i] session = run_one_region_instances( - config, region_number, number_of_instances) + config, region_number, number_of_instances, InstanceType.SPOT_FLEET) results = launch_code_deploy(region_list, commitId) print(results) From fedbef7fad786cac588b218f650ffbe7e0ee2a1e Mon Sep 17 00:00:00 2001 From: Richard Liu Date: Thu, 5 Jul 2018 08:32:33 -0700 Subject: [PATCH 2/3] update soldier to support client. --- .../experiment/commander/main.go | 8 ++- .../experiment/soldier/main.go | 70 ++++++++++++++++--- 2 files changed, 67 insertions(+), 11 deletions(-) diff --git a/aws-experiment-launch/experiment/commander/main.go b/aws-experiment-launch/experiment/commander/main.go index eb78dbaad..4efc93ed2 100644 --- a/aws-experiment-launch/experiment/commander/main.go +++ b/aws-experiment-launch/experiment/commander/main.go @@ -10,10 +10,13 @@ import ( ) const ( - message = "init http://localhost:8080/config.txt" StopCharacter = "\r\n\r\n" ) +var ( + configFile *string +) + func SocketClient(addr string) { conn, err := net.Dial("tcp", addr) @@ -23,6 +26,7 @@ func SocketClient(addr string) { log.Fatalln(err) } + message := "init http://localhost:8080/" + *configFile conn.Write([]byte(message)) // conn.Write([]byte(StopCharacter)) log.Printf("Send: %s", message) @@ -33,7 +37,7 @@ func SocketClient(addr string) { } func main() { - configFile := flag.String("config_file", "config.txt", "file containing all ip addresses") + configFile = flag.String("config_file", "test.txt", "file containing all ip addresses") flag.Parse() configs := readConfigFile(*configFile) diff --git a/aws-experiment-launch/experiment/soldier/main.go b/aws-experiment-launch/experiment/soldier/main.go index 14b98229c..3de62c759 100644 --- a/aws-experiment-launch/experiment/soldier/main.go +++ b/aws-experiment-launch/experiment/soldier/main.go @@ -99,9 +99,10 @@ func handleCommand(command string, w *bufio.Writer) { func handleInitCommand(args []string, w *bufio.Writer) { log.Println("Init command", args) // create local config file + localConfig = "node_config_" + *port + ".txt" out, err := os.Create(localConfig) if err != nil { - log.Fatal("Failed to create local file") + log.Fatal("Failed to create local file", err) } defer out.Close() @@ -127,28 +128,55 @@ func handleInitCommand(args []string, w *bufio.Writer) { log.Println("Successfully downloaded config") log.Println(string(content)) - runInstance() + run() w.Write([]byte("Successfully init-ed")) w.Flush() } -func runInstance() { +func createLogFolder() string { t := time.Now().Format("20060102-150405") logFolder := "../tmp_log/log-" + t err := os.MkdirAll(logFolder, os.ModePerm) if err != nil { log.Fatal("Failed to create log folder") } - cmdName := "./benchmark" - cmdArgs := []string{"-ip", *ip, "-port", *port, "-config_file", localConfig, "-log_folder", logFolder} - log.Println(cmdName, cmdArgs) - cmdOut, err := exec.Command(cmdName, cmdArgs...).Output() + return logFolder +} + +func runCmd(name string, args []string) { + log.Println(name, args) + out, err := exec.Command(name, args...).Output() if err != nil { log.Fatal("Failed to run command: ", err) } - log.Println(string(cmdOut)) + log.Println(string(out)) +} + +func run() { + config := readConfigFile(localConfig) + + myConfig := getMyConfig(*ip, *port, &config) + + log.Println(myConfig) + if myConfig[2] == "client" { + runClient() + } else { + runInstance() + } +} + +func runInstance() { + log.Println("running instance") + logFolder := createLogFolder() + runCmd("./benchmark", []string{"-ip", *ip, "-port", *port, "-config_file", localConfig, "-log_folder", logFolder}) +} + +func runClient() { + log.Println("running client") + logFolder := createLogFolder() + runCmd("./txgen", []string{"-config_file", localConfig, "-log_folder", logFolder}) } func isTransportOver(data string) (over bool) { @@ -156,11 +184,35 @@ func isTransportOver(data string) (over bool) { return } +func readConfigFile(configFile string) [][]string { + file, _ := os.Open(configFile) + fscanner := bufio.NewScanner(file) + + result := [][]string{} + for fscanner.Scan() { + p := strings.Split(fscanner.Text(), " ") + result = append(result, p) + } + return result +} + +func getMyConfig(myIP string, myPort string, config *[][]string) []string { + for _, node := range *config { + ip, port := node[0], node[1] + if ip == myIP && port == myPort { + return node + } + } + return nil +} + +// go build -o bin/soldier aws-experiment-launch/experiment/soldier/main.go +// cd bin/ +// ./soldier --port=xxxx func main() { ip = flag.String("ip", "127.0.0.1", "IP of the node.") port = flag.String("port", "3000", "port of the node.") flag.Parse() - localConfig = "node_config_" + *port + ".txt" SocketServer() } From d5f24489f1580eb71a92eb19e4796f6e1480b7a6 Mon Sep 17 00:00:00 2001 From: Richard Liu Date: Thu, 5 Jul 2018 09:17:29 -0700 Subject: [PATCH 3/3] update soldier to be non-command-blocking. --- .../experiment/soldier/main.go | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/aws-experiment-launch/experiment/soldier/main.go b/aws-experiment-launch/experiment/soldier/main.go index 3de62c759..01d596032 100644 --- a/aws-experiment-launch/experiment/soldier/main.go +++ b/aws-experiment-launch/experiment/soldier/main.go @@ -24,13 +24,13 @@ var ( localConfig string ) -func SocketServer() { +func socketServer() { listen, err := net.Listen("tcp4", ":"+*port) - defer listen.Close() if err != nil { log.Fatalf("Socket listen port %s failed,%s", *port, err) os.Exit(1) } + defer listen.Close() log.Printf("Begin listen for command on port: %s", *port) for { @@ -61,13 +61,15 @@ ILOOP: case io.EOF: break ILOOP case nil: - log.Println("Receive:", data) + log.Println("Received command", data) if isTransportOver(data) { log.Println("Tranport Over!") break ILOOP } - go handleCommand(data, w) + handleCommand(data, w) + + log.Println("Waiting for new command...") default: log.Fatalf("Receive data failed:%s", err) @@ -141,17 +143,18 @@ func createLogFolder() string { if err != nil { log.Fatal("Failed to create log folder") } + log.Println("Created log folder", logFolder) return logFolder } func runCmd(name string, args []string) { log.Println(name, args) - out, err := exec.Command(name, args...).Output() + err := exec.Command(name, args...).Start() if err != nil { log.Fatal("Failed to run command: ", err) } - log.Println(string(out)) + log.Println("Command running") } func run() { @@ -214,5 +217,5 @@ func main() { port = flag.String("port", "3000", "port of the node.") flag.Parse() - SocketServer() + socketServer() }