ceph_load_gen.py

#!/usr/bin/env python
import rados
import time
import random
import threading
import numpy
import string
import sys



setup_latencies = []
write_latencies = []

class que_length(object):
    """docstring for que_length"""
    def __init__(self):
        self.len = 0


def write_object(connection=None, ioctx=None, obj=None, obj_name=None, queue_length=None, it=None):
    queue_length.len+=1
    close = False
    start = time.time()
    try:
        if connection is None:
            connection = rados.Rados(conffile="/etc/ceph/ceph.conf")
            connection.connect()
        if ioctx is None:
            close = True
            ioctx = connection.open_ioctx("cdvr_ec")
            ioctx.require_ioctx_open()
        setup_latencies.append(time.time()-start)
        status = ioctx.aio_write(obj_name, obj)
        status.wait_for_complete()
        if close:
            ioctx.close()
        write_latencies.append(time.time()-start)
    except Exception as e:
        print e
        queue_length.len-=1
        return True
    queue_length.len-=1
    return True

def que_status(queue=None, sleep_time=0.01, depth=4):
    while queue.len >= depth:
        # print "queue sleep", queue.len,  depth
        time.sleep(sleep_time)
    return

def write_queue( connection=None, ioctx=None, obj=None, obj_name=None, que_length=None):
    que_status(queue=que_length, depth=10)
    write_object( connection=connection, ioctx=ioctx, obj=obj, obj_name=obj_name, queue_length=que_length)



def main():
    n=0
    Data = ''.join([random.choice(string.ascii_letters) for j in
                    range(4000000)])

    connections = []
    contexts = []
    for i in range(1):
        connection = rados.Rados(conffile="/etc/ceph/ceph.conf")
        connection.connect()
	time.sleep(1)
        connections.append(connection)
        for j in range(10):
            contexts.append(connection.open_ioctx("cdvr_ec"))

    que = que_length()
    threads=[]
    connection=None
    ioctx=None
    #connection = rados.Rados(conffile="/etc/ceph/ceph2.conf")
    #connection.connect()
    #ioctx = connection.open_ioctx("cdvr_ec")
    while n < 5000:
        que_status(queue=que, depth=20)
        name = ''.join([random.choice(string.ascii_letters) for
                        j in range(40)])
        t=threading.Thread(group=None,target=write_object,
                args=(1,
                    contexts[n%10],Data, name, que, n) )
        n+=1
        t.start()
        threads.append(t)
    an = numpy.array(write_latencies)
    if len(write_latencies) > 0:
        for x in xrange(0,101,1):
            print "Percentile {x} = {y}".format(x=x, y=numpy.percentile(an,x))
        print "Max latancies = {max}, Min = {min}, mean = {mean}".format(max=an.max(), min=an.min(), mean=an.mean())
    an = numpy.array(setup_latencies)
    if len(setup_latencies) > 0:
        for x in xrange(0,101,1):
            print "Percentile {x} = {y}".format(x=x, y=numpy.percentile(an,x))
        print "Max latancies = {max}, Min = {min}, mean = {mean}".format(max=an.max(), min=an.min(), mean=an.mean())


if __name__ == '__main__':
    main()