使用python boto对接radosgw测试

对ceph的s3接口测试，除了此前介绍的cosbench之外，还可以手工编写python程序对集群进行功能、性能方面的测试
话不多说，直接上代码：

写在前面

松鼠哥的ceph专业课程上线啦！
面向新手同学，从0实战，全面入门ceph安装部署与运维，有需要的同学赶紧扫码订购吧：

对ceph的s3接口测试，除了此前介绍的cosbench之外，还可以手工编写python程序对集群进行功能、性能方面的测试

下面是代码实例

#!/usr/bin/env python
import boto
import time
import boto.s3.connection
from multiprocessing import Process
from threading import Thread
from multiprocessing import Queue as bigqueue

class Insert_process(Process):
    def __init__(self, maxthread, msg, imgname, endpoint):
        Process.__init__(self)
        self.access_key = 'AAAAAAAAAAAAAAAA' #access key 和secret key
        self.secret_key = 'hhhhhhhhhhhhhhhhhhhhhhhhhhh'
        self.maxthread = maxthread
        self.endpoint = endpoint
        self.msg = msg
        self.imgname = imgname

    def run(self):
        all_thread = []
        for x in xrange(0, self.maxthread):
            tempthread = Insert_thread(access_key=self.access_key,
                                       secret_key=self.secret_key,
                                       endpoint=self.endpoint,
                                       imgname=self.imgname)
            tempthread.daemon = True
            all_thread.append(tempthread)
            tempthread.start()
        gmsg = self.msg.get()
        for y in all_thread:
            y.iamrun = False
        for t in all_thread:
            if t.isAlive():
                t.join()
        return

class Insert_thread(Thread):
    def __init__(self, access_key, secret_key, endpoint, imgname):
        Thread.__init__(self)
        self.access_key = access_key
        self.secret_key = secret_key
        self.iamrun = True
        self.endpoint = endpoint
        self.imgname = imgname

    def run(self):
        #连接到网关
        self.conn = boto.connect_s3(aws_access_key_id = self.access_key,
                                    aws_secret_access_key = self.secret_key,
                                    host = self.endpoint,
                                    port = 7480,
                                    is_secure=False,
                                    calling_format = boto.s3.connection.OrdinaryCallingFormat())
        self.insert_data_bucket = self.conn.get_bucket('my-bucket-2018-04-16') #获得bucket
        while self.iamrun:
            #timestart = time.time()
            self.newobj = self.insert_data_bucket.new_key(str(time.time())+'.jpg') #首先初始化一个文件
            self.newobj.set_contents_from_filename('COSBenchAdaptorDevGuide.pdf')  #然后将本地文件上传到集群中
            #timeend = time.time()
            #print timeend - timestart
        return

def check_switch(switchfile):
    ''' TODO:check the specify file and tell quit or not '''
    with open(switchfile,'r') as switch:
        how = switch.readline()
        if 'false' in how:
            return False
        else:
            return True


if __name__ == '__main__':
        all_proc = []
        imsg = bigqueue()
        imgname = bigqueue()
        maxthreads = 16 #配置每个进程使用多少个线程进行测试
        maxprocess = 8  #配置一共有多少个进程进行测试
        endpoint1 = '172.25.59.204'  #被测节点，并使用默认的7480端口
        endpoint2 = '172.25.59.205'
        for x in xrange(0,maxprocess):
            yo = Insert_process(maxthread=maxthreads, msg=imsg, imgname=imgname, endpoint=endpoint1)
            yo.daemon = True
            all_proc.append(yo)
            yo.start()
        for x in xrange(0,maxprocess):
            yo = Insert_process(maxthread=maxthreads, msg=imsg, imgname=imgname, endpoint=endpoint2)
            yo.daemon = True
            all_proc.append(yo)
            yo.start()
        while True:    #定时去检测当前目录下的switch_hdd文件，是否需要退出测试
            if check_switch('switch_hdd'):
                time.sleep(5)
            else:
                break
        for bbq in xrange(0,maxprocess*2):
                imsg.put('quit')
        for sthread in all_proc:
            sthread.join()

比较粗糙的程序，凑合用着，迟点再补上延迟、并发等功能
使用该程序的时候，脚本当前目录要有switch_hdd 开关文件，需要修改线程、进程数，被测节点ip、access key 和secret key、bucket、上传文件名

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： 奋斗的松鼠
• 本文链接： http://www.strugglesquirrel.com/2018/04/16/使用python-boto对接radosgw测试/
• 版权声明： 本博客所有文章除特别声明外，创作版权均为作者个人所有，未经允许禁止转载！