#!/usr/bin/python import sys import tempfile import time from execo_g5k import * from execo import * OUTDIR="/home/dponsard/DFS_EXECO/RESULT" def usage(): print "Usage %s job site dfs_type" % sys.argv[0] sys.exit(1) def start_benchmark(client_nodes,bench_size=4096): prepare=Remote("sync && echo 3 > /proc/sys/vm/drop_caches && echo 3 > /proc/sys/vm/drop_caches", client_nodes, connexion_params={'user': 'root'}) bench=Remote("TMP_DIR=`mktemp -d /data/BENCH.XXXX` ;dd if=/dev/zero of=${TMP_DIR}/write.dd bs=1M count=%d; sync" % bench_size, client_nodes, connexion_params={'user': 'root'}) prepare.run() bench.run() cumul_speed = 0 for p in bench.processes(): cumul_speed += bench_size/float(p.end_date()-p.start_date()) print "Benchmark speed is "+str(cumul_speed) print "Benchmark output" for p in bench.processes(): print p.stdout() return cumul_speed def create_dfs5k_conf(dfs_type,site,job,master_node,client_nodes,data_nodes): try: f=tempfile.NamedTemporaryFile(prefix="dfs5k_%s_%s_%s-" % (dfs_type,site,job), delete=False) except Exception as e: print "Failed to open /tmp/dfs5k.conf" print e sys.exit(1) if dfs_type == "ceph" or dfs_type=="lustre": f.write("""name: DFS5k options: dataDir: /tmp master: root@%s clients: """ % master_node.address) for client_node in client_nodes: f.write(" root@%s\n" % client_node.address) f.write("dataNodes:\n") for data_node in data_nodes: f.write(" root@%s\n" % data_node.address) elif dfs_type == "glusterfs": f.write("""name: DFS5k options: stripe %d master: root@%s clients: """ % (len(data_nodes),master_node.address)) for client_node in client_nodes: f.write(" root@%s\n" % client_node.address) f.write("dataNodes:\n") for data_node in data_nodes: f.write(" root@%s:/tmp\n" % data_node.address) else: print "Error: Unsupported dfs type %s" % dfs_type sys.exit(1) f.close() return f def start_localfs_bench(site,job,node,bench_size=4096): prepare=SshProcess("sync && echo 3 > /proc/sys/vm/drop_caches && echo 3 > /proc/sys/vm/drop_caches", node, connexion_params={'user': 'root'}) bench=SshProcess("dd if=/dev/zero of=/tmp/write.dd bs=1M count=%d && sync" % bench_size, node, connexion_params={'user': 'root'}) prepare.run() start=time.time() bench.run() end=time.time() print "Benchmark output" print bench.stdout() print "Cleanning /tmp" clean=SshProcess("rm -rf /tmp/write.dd",node, connexion_params={'user': 'root'}) clean.run() speed = bench_size/float(end-start) print "Benchmark speed is "+str(speed) return speed def clean_nodes_before(dfs_type,site,dfs_conf,nodes): print "Cleaning /data" Remote("rm -rf /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run() print "Umounting DFS5k" print "Umounting DFS5k" SshProcess("dfs5k -d -a umount -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run() if dfs_type != "lustre": print "UnDeploying DFS5k" SshProcess("dfs5k -d -a undeploy -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run() else: print "UnDeploying DFS5k" p=SshProcess("dfs5k -d -a undeploy -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site), close_stdin=False).start() p._process.stdin.write("y\n") p._process.stdin.flush() p.wait() print "Cleaning /tmp and /data" Remote("rm -rf /tmp/* /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run() return def clean_nodes_after(dfs_type,site,dfs_conf,nodes): print "Cleaning /data" Remote("rm -rf /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run() print "Umounting DFS5k" SshProcess("dfs5k -d -a umount -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run() print "UnDeploying DFS5k" SshProcess("dfs5k -d -a undeploy -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run() print "Cleaning /tmp and /data" Remote("rm -rf /tmp/* /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run() return def start_dfs_bench(dfs_type,site,job,master_node,client_nodes,data_nodes): f = create_dfs5k_conf(dfs_type,site,job,master_node,client_nodes,data_nodes) print "Copying dfs5k configuration file" Process("scp %s %s:/tmp/" % (f.name,site)).run() print "Copying SSH key to master node" SshProcess("scp ~/.ssh/id_rsa root@%s:.ssh/" % master_node.address, Host(site)).run() clean_nodes_before(dfs_type, site, f, data_nodes+client_nodes+[master_node]) print "Deploying DFS5k" SshProcess("dfs5k -d -a deploy -s %s -f %s" % (dfs_type,f.name) , Host(site)).run() print "Mounting DFS5k" print "Deploying DFS5k" SshProcess("dfs5k -d -a mount -s %s -f %s" % (dfs_type,f.name) , Host(site)).run() print "Performing benchmark %s" % dfs_type print "Master node is %s" % master_node print "## %d Data nodes are %s" % (len(data_nodes), data_nodes) print "## %d Client nodes are %s" % (len(client_nodes), client_nodes) speed = start_benchmark(client_nodes) clean_nodes_after(dfs_type, site, f, data_nodes+client_nodes+[master_node]) return speed def prepare_nodes(job,site,dfs_type): nodes = [] try: if get_oar_job_info(job,site)["state"]!="Running": print "Cannot deploy on job %s @ %s, job not started" % (job,site) else: nodes += get_oar_job_nodes(job,site) except Exception as e: print "Cannot get info on job %s @ %s" % (job,site) print e try: print "Deploying on %s" % nodes if dfs_type=="lustre": print "Deploying on %s environment centos-x64-lustre for %s" % (nodes, dfs_type) (ok_nodes, failed_nodes) = deploy(Deployment(nodes,env_name="centos-x64-lustre"), node_connexion_params={'user': 'root'}) else: print "Deploying on %s environment squeeze-x64-nfs for %s" % (nodes, dfs_type) (ok_nodes, failed_nodes) = deploy(Deployment(nodes,env_name="squeeze-x64-nfs")) print "Deployed on %s " % ok_nodes except Exception as e: print "Failed to deploy" print e raise e sys.exit(1) else: #print "Installing iozone" #Remote("apt-get update && apt-get install -y iozone3",ok_nodes, connexion_params={'user': 'root'}).run() return list(ok_nodes) if len(sys.argv) < 4: usage() job = int(sys.argv[1]) site = sys.argv[2] dfs_type= sys.argv[3] f_result = open(OUTDIR+"/results_%s_%s_%s.txt" % (job,site,dfs_type),"w") nodes=prepare_nodes(job,site,dfs_type) #nodes=["A","B","C","D","E","F","G","H","I","J","K","L","M","N","O","P"] print nodes master_node=nodes[0] #Write first line of the table line="X" for i in range(2,len(nodes)-1): line+=" %d" % i line+="\n" f_result.write(line) print "#### START LOCALFS BENCH #####" speed = start_localfs_bench(site,job,master_node) line="0" for i in range(2,len(nodes)-1): line+=" %d" % int(speed) line+="\n" f_result.write(line) for n_clients in [1,2,4,6,8,10,12,14,16,18,20]: f_result.write("%d" % n_clients) available_nodes = nodes[1:] client_nodes=available_nodes[:n_clients] available_nodes=available_nodes[n_clients:] for n_datanodes in range(2,len(available_nodes)+1): data_nodes=available_nodes[:n_datanodes] print "\n\n" print "#### START DFS BENCH #####" print master_node,client_nodes,data_nodes print "#### ############### #####" speed = start_dfs_bench(dfs_type,site,job,master_node,client_nodes,data_nodes) f_result.write(" %d" % int(speed)) f_result.flush() f_result.write("\n") f_result.flush() f_result.write("\n\n\n") f_result.write(str(nodes)) f_result.close()