2011-06-13 23:36:21 +00:00
|
|
|
import random
|
|
|
|
import time
|
|
|
|
import re
|
|
|
|
import gevent
|
|
|
|
from orchestra import run
|
|
|
|
|
2011-08-25 22:18:42 +00:00
|
|
|
CLEANINT=60
|
|
|
|
DELAY=5
|
|
|
|
|
2011-06-13 23:36:21 +00:00
|
|
|
class Thrasher(gevent.Greenlet):
|
2011-08-25 22:18:42 +00:00
|
|
|
def __init__(self, manager, config, logger=None):
|
2011-06-13 23:36:21 +00:00
|
|
|
self.ceph_manager = manager
|
|
|
|
self.ceph_manager.wait_till_clean()
|
|
|
|
osd_status = self.ceph_manager.get_osd_status()
|
|
|
|
self.in_osds = osd_status['in']
|
|
|
|
self.out_osds = osd_status['out']
|
|
|
|
self.stopping = False
|
|
|
|
self.logger = logger
|
2011-08-25 22:18:42 +00:00
|
|
|
self.config = config
|
2011-06-13 23:36:21 +00:00
|
|
|
if self.logger != None:
|
|
|
|
self.log = lambda x: self.logger.info(x)
|
|
|
|
else:
|
|
|
|
def tmp(x):
|
|
|
|
print x
|
|
|
|
self.log = tmp
|
2011-08-25 22:18:42 +00:00
|
|
|
if self.config is None:
|
|
|
|
self.config = dict()
|
2011-06-13 23:36:21 +00:00
|
|
|
gevent.Greenlet.__init__(self, self.do_thrash)
|
|
|
|
self.start()
|
|
|
|
|
|
|
|
def remove_osd(self):
|
2011-08-25 22:19:30 +00:00
|
|
|
chance_down = 0
|
|
|
|
if self.config.get("chanceDown"):
|
|
|
|
if isinstance(self.config["chanceDown"], int):
|
|
|
|
chance_down = float(self.config["chanceDown"])/100
|
|
|
|
else:
|
|
|
|
chance_down = self.config["chanceDown"]
|
2011-06-13 23:36:21 +00:00
|
|
|
osd = random.choice(self.in_osds)
|
2011-08-25 22:19:30 +00:00
|
|
|
if random.uniform(0,1) < chance_down:
|
|
|
|
self.log("Marking osd {id_} down".format(id_=osd))
|
|
|
|
self.ceph_manager.mark_down_osd(osd)
|
|
|
|
else:
|
|
|
|
self.log("Removing osd %s"%(str(osd),))
|
|
|
|
self.ceph_manager.mark_out_osd(osd)
|
|
|
|
self.in_osds.remove(osd)
|
|
|
|
self.out_osds.append(osd)
|
|
|
|
|
2011-06-13 23:36:21 +00:00
|
|
|
|
|
|
|
def add_osd(self):
|
|
|
|
osd = random.choice(self.out_osds)
|
|
|
|
self.log("Adding osd %s"%(str(osd),))
|
|
|
|
self.out_osds.remove(osd)
|
|
|
|
self.in_osds.append(osd)
|
|
|
|
self.ceph_manager.mark_in_osd(osd)
|
|
|
|
|
|
|
|
def all_up(self):
|
|
|
|
while len(self.out_osds) > 0:
|
|
|
|
self.add_osd()
|
|
|
|
|
|
|
|
def do_join(self):
|
|
|
|
self.stopping = True
|
|
|
|
self.get()
|
|
|
|
|
|
|
|
def do_thrash(self):
|
2011-08-25 22:18:42 +00:00
|
|
|
cleanint = CLEANINT
|
|
|
|
delay = DELAY
|
|
|
|
minin = 2
|
|
|
|
minout = 0
|
|
|
|
if self.config.get("cleanInterval"):
|
|
|
|
cleanint = self.config["cleanInterval"]
|
|
|
|
if self.config.get("opDelay"):
|
|
|
|
delay = self.config["opDelay"]
|
|
|
|
if self.config.get("minIn"):
|
|
|
|
minin = self.config["minIn"]
|
|
|
|
if self.config.get("minOut"):
|
|
|
|
minout = self.config["minOut"]
|
2011-06-13 23:36:21 +00:00
|
|
|
self.log("starting do_thrash")
|
|
|
|
while not self.stopping:
|
|
|
|
self.log(" ".join([str(x) for x in ["in_osds: ", self.in_osds, " out_osds: ", self.out_osds]]))
|
2011-08-25 22:18:42 +00:00
|
|
|
if random.uniform(0,1) < (float(delay)/cleanint):
|
2011-08-25 21:38:34 +00:00
|
|
|
self.ceph_manager.wait_till_clean()
|
2011-08-25 22:18:42 +00:00
|
|
|
if (len(self.out_osds) == minout):
|
2011-06-13 23:36:21 +00:00
|
|
|
self.remove_osd()
|
2011-08-25 22:18:42 +00:00
|
|
|
elif (len(self.in_osds) <= minin):
|
2011-07-12 01:00:03 +00:00
|
|
|
self.add_osd()
|
2011-06-13 23:36:21 +00:00
|
|
|
else:
|
2011-07-12 01:00:03 +00:00
|
|
|
x = random.choice([self.remove_osd, self.add_osd])
|
|
|
|
x()
|
2011-08-25 22:18:42 +00:00
|
|
|
time.sleep(delay)
|
2011-06-13 23:36:21 +00:00
|
|
|
|
|
|
|
class CephManager:
|
|
|
|
def __init__(self, controller, logger=None):
|
|
|
|
self.controller = controller
|
|
|
|
if (logger):
|
|
|
|
self.log = lambda x: logger.info(x)
|
|
|
|
else:
|
|
|
|
def tmp(x):
|
|
|
|
print x
|
|
|
|
self.log = tmp
|
|
|
|
|
|
|
|
def raw_cluster_cmd(self, suffix):
|
|
|
|
proc = self.controller.run(
|
|
|
|
args=[
|
|
|
|
"/bin/sh", "-c",
|
|
|
|
" ".join([
|
|
|
|
"LD_LIBRARY_PRELOAD=/tmp/cephtest/binary/usr/local/lib",
|
|
|
|
"/tmp/cephtest/binary/usr/local/bin/ceph -k /tmp/cephtest/ceph.keyring -c "+\
|
|
|
|
"/tmp/cephtest/ceph.conf " + suffix
|
|
|
|
])
|
|
|
|
],
|
|
|
|
stdout=run.PIPE,
|
|
|
|
wait=False
|
|
|
|
)
|
2011-07-12 01:00:03 +00:00
|
|
|
|
2011-06-13 23:36:21 +00:00
|
|
|
out = ""
|
|
|
|
tmp = proc.stdout.read(1)
|
|
|
|
while tmp:
|
|
|
|
out += tmp
|
|
|
|
tmp = proc.stdout.read(1)
|
|
|
|
return out
|
|
|
|
|
|
|
|
def raw_cluster_status(self):
|
|
|
|
return self.raw_cluster_cmd("-s")
|
2011-07-12 01:00:03 +00:00
|
|
|
|
2011-06-13 23:36:21 +00:00
|
|
|
def raw_osd_status(self):
|
|
|
|
return self.raw_cluster_cmd("osd dump -o -")
|
|
|
|
|
|
|
|
def raw_pg_status(self):
|
|
|
|
return self.controller.do_ssh("pg dump -o -")
|
|
|
|
|
|
|
|
def get_osd_status(self):
|
|
|
|
osd_lines = filter(
|
|
|
|
lambda x: x[:3] == 'osd' and (("up" in x) or ("down" in x)),
|
|
|
|
self.raw_osd_status().split('\n'))
|
|
|
|
self.log(osd_lines)
|
|
|
|
in_osds = [int(i[3:].split()[0]) for i in filter(
|
|
|
|
lambda x: " in " in x,
|
|
|
|
osd_lines)]
|
|
|
|
out_osds = [int(i[3:].split()[0]) for i in filter(
|
|
|
|
lambda x: " out " in x,
|
|
|
|
osd_lines)]
|
|
|
|
up_osds = [int(i[3:].split()[0]) for i in filter(
|
|
|
|
lambda x: " up " in x,
|
|
|
|
osd_lines)]
|
|
|
|
down_osds = [int(i[3:].split()[0]) for i in filter(
|
|
|
|
lambda x: " down " in x,
|
|
|
|
osd_lines)]
|
2011-07-12 01:00:03 +00:00
|
|
|
return { 'in' : in_osds, 'out' : out_osds, 'up' : up_osds,
|
2011-06-13 23:36:21 +00:00
|
|
|
'down' : down_osds, 'raw' : osd_lines }
|
|
|
|
|
|
|
|
def get_num_pgs(self):
|
|
|
|
status = self.raw_cluster_status()
|
|
|
|
return int(re.search(
|
|
|
|
"\d* pgs:",
|
|
|
|
status).group(0).split()[0])
|
|
|
|
|
|
|
|
def get_num_active_clean(self):
|
|
|
|
status = self.raw_cluster_status()
|
|
|
|
self.log(status)
|
|
|
|
match = re.search(
|
|
|
|
"\d* active.clean",
|
|
|
|
status)
|
|
|
|
if match == None:
|
|
|
|
return 0
|
|
|
|
else:
|
|
|
|
return int(match.group(0).split()[0])
|
|
|
|
|
|
|
|
def is_clean(self):
|
|
|
|
return self.get_num_active_clean() == self.get_num_pgs()
|
|
|
|
|
|
|
|
def wait_till_clean(self):
|
|
|
|
self.log("waiting till clean")
|
|
|
|
while not self.is_clean():
|
|
|
|
time.sleep(3)
|
|
|
|
self.log("clean!")
|
|
|
|
|
|
|
|
def mark_out_osd(self, osd):
|
|
|
|
self.raw_cluster_cmd("osd out %s"%(str(osd,)))
|
|
|
|
|
2011-08-25 22:19:30 +00:00
|
|
|
def mark_down_osd(self, osd):
|
|
|
|
self.raw_cluster_cmd("osd down %s"%(str(osd,)))
|
|
|
|
|
2011-06-13 23:36:21 +00:00
|
|
|
def mark_in_osd(self, osd):
|
|
|
|
self.raw_cluster_cmd("osd in %s"%(str(osd,)))
|