paralle_curl.py

#!/usr/bin/python
import sys
import os
import time
import os.path
import threading
import socket

verbose=os.getenv('VERBOSE')

curlcmdbase="curl -i "

fs='test'
user='%s:tester'%fs
password='testing'

#default is no encryption
url_prefix=os.getenv('URL_PREFIX')
if not url_prefix: url_prefix='http'

# if we're using plain http: default to 8080 like openswift
# otherwise don't specify at all, this defaults to 443
http_portenv=os.getenv('HTTP_PORT')
if (not http_portenv) and url_prefix == 'http' : http_portenv='61005'
http_port = int(http_portenv)
http_portstr=''
if http_port: http_portstr=':%d'%http_port

# don't force them to use a certificate unless we are using https
cert = os.getenv('CERT')
if (not cert) and url_prefix == 'https': cert='cert.crt'
cacert_option=''
if cert: cacert_option = ' --cacert %s '%cert
# parse command line arguments

if len(sys.argv) < 6:
  print 'usage: parallel_curl.sh op threads files size-KB client1,...,clientM server1,...,serverN'
  sys.exit(1)

op=sys.argv[1]
threads=int(sys.argv[2])
files=int(sys.argv[3])
sz=int(sys.argv[4])
client_list=sys.argv[5].split(',')
server_list=sys.argv[6].split(',')

svrcount=len(server_list)
clntcount=len(client_list)
me=socket.gethostname().split('.')[0]

if verbose:
  print 'doing %s %d threads %d objects/thread %d KB/object'%(op, threads, files, sz)
  print 'url_prefix: ' + url_prefix
  print 'http_portstr: ' + http_portstr
  print 'cert: ' + cacert_option

# parallel requests are done with this class
count3 = 0
class curl_thread(threading.Thread):
  def __init__(self, client, thread_id, cmd):
    threading.Thread.__init__(self)
    self.cmd = cmd
    self.client = client
    self.thread_id = thread_id

  def run(self):
    global count3
    for fnum in range(0, files):
      next_cmd = 'ssh -nx -o StrictHostKeyChecking=no %s "'%self.client + self.cmd%fnum
      next_cmd += ' | strings " >> /tmp/curl-thread.log'#%self.thread_id
      if verbose: print 'next command for thread %d: %s'%(self.thread_id, next_cmd)
      os.system(next_cmd)
      count3 =count3 +1

# get authorization tokens from file or from server if necessary

tkfile=os.path.join(os.getenv('HOME'), 'tk.tmp')
auth_tokens = []
if os.access(tkfile, os.R_OK):
  f = open(tkfile, "r")
  auth_tokens = [ t.strip() for t in f.readlines() ]
  f.close()
if len(auth_tokens) < min(threads,svrcount):  # if we couldn't find any authorization tokens from last run
  for k in range(0, min(threads,svrcount)):
    svr = server_list[k]
    cmd = curlcmdbase + cacert_option
    cmd += " -v -H 'X-Storage-User: %s' -H 'X-Storage-Pass: %s' %s://%s:%s/auth/v1.0 2>&1 | "
    cmd += "awk '/X-Auth-Token/{ print $3 }' >> %s"
    cmd = cmd%(user, password, url_prefix, svr, http_port, tkfile)
    if verbose: print cmd
    os.system(cmd)
  f = open(tkfile, "r")
  auth_tokens = [ t.strip() for t in f.readlines() ]
  f.close()
if len(auth_tokens) == 0:
  raise Exception('failed to obtain any authorization tokens')
if verbose: print 'authorization tokens = %s'%auth_tokens
# for put operation we need a file on each client to put
# generate one and make sure it's on all clients

if op == 'put':
  current_sz = -1
  bfn = '/tmp/big.dd'
  if os.access(bfn, os.R_OK):
    current_sz = os.path.getsize(bfn)
  if sz * 1024 != current_sz:
    cmd = 'rm -fv %s ; dd if=/dev/zero of=%s bs=1k count=%d && ls -l %s && gzip -c < %s > %s.gz'%\
         (bfn, bfn, sz, bfn, bfn, bfn)
    if verbose: print cmd
    os.system(cmd)
    assert(os.path.getsize(bfn) == sz * 1024)
    for j in range(0, min(threads,clntcount)):
      clnt = client_list[j]
      cmd=' rsync -av %s.gz %s:%s.gz'%(bfn, clnt, bfn)
      if verbose: print cmd
      os.system(cmd)
      cmd = 'ssh -nx %s "gunzip -c < %s.gz > %s && ls -l %s"'%(clnt, bfn, bfn, bfn)
      if verbose: print cmd
      os.system(cmd)
os.system('rm -f /tmp/curl-thread*.log')
start_time=time.time()

# generate curl command for each thread and start it

curl_thread_list = []
count = 0
for thr in range(0, threads):
  clnt = client_list[thr%clntcount] # which client to use
  svr = server_list[thr%svrcount] # which server to use
  tk = auth_tokens[thr%svrcount]  # which authorization token to use
  curlcmd = curlcmdbase + cacert_option + "-H 'X-Auth-Token: %s ' "%tk
  if op == "put":
    # container must exist before we put objects in it

    container_put = 'ssh -nx ' + clnt + " \"" + curlcmd + " -X PUT %s://%s%s/v1/AUTH_test/%s-%d\""%\
                (url_prefix, server_list[0], http_portstr, clnt, thr)
    count =count +1;
    print "count \n", count
    time.sleep(1)
    if verbose: print "\n",container_put
    os.system(container_put)
    curlcmd += "-X PUT -T /tmp/big.dd "
  elif op == "get":
    curlcmd += "-X GET "
  elif op == "delete":
    curlcmd += "-X DELETE "
  else:
    raise Exception("invalid op " + op)
  curlcmd += " %s://%s%s/v1/AUTH_%s/%s-%d/f%s"%(url_prefix, svr, http_portstr, fs, clnt, thr, '%d')
  curl_thread_list.append(curl_thread(clnt, thr, curlcmd))
time.sleep(1)
count2 = 0
for thr in range(0, threads):
  count2 = count2+1
  curl_thread_list[thr].start()
for thr in range(0, threads):
  curl_thread_list[thr].join(100)

# read log files from HTTP activity
# and detect errors
errors=0
#for thr in range(0, threads):
fn = '/tmp/curl-thread.log'#%thr
if verbose: print '*******reading %s'%fn
logf = open(fn, "r")
lines = logf.readlines()
logf.close()
for ln in lines:
    if ln.__contains__('<title>'):
        code = ln.split('>')[1].split()[0]
        rc = int(code)
        if rc < 200 or rc >= 300:
                print 'ERROR: '+ code
                errors += 1
end_time = time.time()
if errors > 0:
  print 'WARNING: %d errors found'%errors
print 'doing %s %d threads %d objects/thread %d KB/object'%(op, threads, files, sz)
print 'clients: %s'%client_list
print 'servers: %s'%server_list
elapsed_time = end_time - start_time
print 'elapsed time = %7.2f sec'%elapsed_time
throughput = files * threads / elapsed_time
print 'throughput = %9.2f objs/sec'%throughput
if op != 'delete':
  transfer_rate = sz * throughput / 1024.0
  print 'transfer rate = %9.2f MB/s'%transfer_rate
print "count thread===", count2
print "\ncount 3========", count3