src/ceph/qa/tasks/watch_notify_same_primary.py

   1
   2 """
   3 watch_notify_same_primary task
   4 """
   5 from cStringIO import StringIO
   6 import contextlib
   7 import logging
   8
   9 from teuthology.orchestra import run
  10 from teuthology.contextutil import safe_while
  11
  12 log = logging.getLogger(__name__)
  13
  14
  15 @contextlib.contextmanager
  16 def task(ctx, config):
  17     """
  18     Run watch_notify_same_primary
  19
  20     The config should be as follows:
  21
  22     watch_notify_same_primary:
  23         clients: [client list]
  24
  25     The client list should contain 1 client
  26
  27     The test requires 3 osds.
  28
  29     example:
  30
  31     tasks:
  32     - ceph:
  33     - watch_notify_same_primary:
  34         clients: [client.0]
  35     - interactive:
  36     """
  37     log.info('Beginning watch_notify_same_primary...')
  38     assert isinstance(config, dict), \
  39         "please list clients to run on"
  40
  41     clients = config.get('clients', ['client.0'])
  42     assert len(clients) == 1
  43     role = clients[0]
  44     assert isinstance(role, basestring)
  45     PREFIX = 'client.'
  46     assert role.startswith(PREFIX)
  47     (remote,) = ctx.cluster.only(role).remotes.iterkeys()
  48     manager = ctx.managers['ceph']
  49     manager.raw_cluster_cmd('osd', 'set', 'noout')
  50
  51     pool = manager.create_pool_with_unique_name()
  52     def obj(n): return "foo-{num}".format(num=n)
  53     def start_watch(n):
  54         remote.run(
  55             args = [
  56                 "rados",
  57                 "-p", pool,
  58                 "put",
  59                 obj(n),
  60                 "/etc/resolv.conf"],
  61             logger=log.getChild('watch.{id}'.format(id=n)))
  62         proc = remote.run(
  63             args = [
  64                 "rados",
  65                 "-p", pool,
  66                 "watch",
  67                 obj(n)],
  68             stdin=run.PIPE,
  69             stdout=StringIO(),
  70             stderr=StringIO(),
  71             wait=False)
  72         return proc
  73
  74     num = 20
  75
  76     watches = [start_watch(i) for i in range(num)]
  77
  78     # wait for them all to register
  79     for i in range(num):
  80         with safe_while() as proceed:
  81             while proceed():
  82                 proc = remote.run(
  83                     args = [
  84                         "rados",
  85                         "-p", pool,
  86                         "listwatchers",
  87                         obj(i)],
  88                     stdout=StringIO())
  89                 lines = proc.stdout.getvalue()
  90                 num_watchers = lines.count('watcher=')
  91                 log.info('i see %d watchers for %s', num_watchers, obj(i))
  92                 if num_watchers >= 1:
  93                     break
  94
  95     def notify(n, msg):
  96         remote.run(
  97             args = [
  98                 "rados",
  99                 "-p", pool,
 100                 "notify",
 101                 obj(n),
 102                 msg],
 103             logger=log.getChild('notify.{id}'.format(id=n)))
 104
 105     [notify(n, 'notify1') for n in range(len(watches))]
 106
 107     manager.kill_osd(0)
 108     manager.mark_down_osd(0)
 109
 110     [notify(n, 'notify2') for n in range(len(watches))]
 111
 112     try:
 113         yield
 114     finally:
 115         log.info('joining watch_notify_stress')
 116         for watch in watches:
 117             watch.stdin.write("\n")
 118
 119         run.wait(watches)
 120
 121         for watch in watches:
 122             lines = watch.stdout.getvalue().split("\n")
 123             got1 = False
 124             got2 = False
 125             for l in lines:
 126                 if 'notify1' in l:
 127                     got1 = True
 128                 if 'notify2' in l:
 129                     got2 = True
 130             log.info(lines)
 131             assert got1 and got2
 132
 133         manager.revive_osd(0)
 134         manager.remove_pool(pool)