ceph/qa/tasks/cephfs/mount.py

   1 from contextlib import contextmanager
   2 import json
   3 import logging
   4 import datetime
   5 import time
   6 from textwrap import dedent
   7 import os
   8 from StringIO import StringIO
   9 from teuthology.orchestra import run
  10 from teuthology.orchestra.run import CommandFailedError, ConnectionLostError
  11
  12 log = logging.getLogger(__name__)
  13
  14
  15 class CephFSMount(object):
  16     def __init__(self, test_dir, client_id, client_remote):
  17         """
  18         :param test_dir: Global teuthology test dir
  19         :param client_id: Client ID, the 'foo' in client.foo
  20         :param client_remote: Remote instance for the host where client will run
  21         """
  22
  23         self.test_dir = test_dir
  24         self.client_id = client_id
  25         self.client_remote = client_remote
  26         self.mountpoint_dir_name = 'mnt.{id}'.format(id=self.client_id)
  27
  28         self.test_files = ['a', 'b', 'c']
  29
  30         self.background_procs = []
  31
  32     @property
  33     def mountpoint(self):
  34         return os.path.join(
  35             self.test_dir, '{dir_name}'.format(dir_name=self.mountpoint_dir_name))
  36
  37     def is_mounted(self):
  38         raise NotImplementedError()
  39
  40     def mount(self, mount_path=None, mount_fs_name=None):
  41         raise NotImplementedError()
  42
  43     def umount(self):
  44         raise NotImplementedError()
  45
  46     def umount_wait(self, force=False, require_clean=False):
  47         """
  48
  49         :param force: Expect that the mount will not shutdown cleanly: kill
  50                       it hard.
  51         :param require_clean: Wait for the Ceph client associated with the
  52                               mount (e.g. ceph-fuse) to terminate, and
  53                               raise if it doesn't do so cleanly.
  54         :return:
  55         """
  56         raise NotImplementedError()
  57
  58     def kill_cleanup(self):
  59         raise NotImplementedError()
  60
  61     def kill(self):
  62         raise NotImplementedError()
  63
  64     def cleanup(self):
  65         raise NotImplementedError()
  66
  67     def wait_until_mounted(self):
  68         raise NotImplementedError()
  69
  70     def get_keyring_path(self):
  71         return '/etc/ceph/ceph.client.{id}.keyring'.format(id=self.client_id)
  72
  73     @property
  74     def config_path(self):
  75         """
  76         Path to ceph.conf: override this if you're not a normal systemwide ceph install
  77         :return: stringv
  78         """
  79         return "/etc/ceph/ceph.conf"
  80
  81     @contextmanager
  82     def mounted(self):
  83         """
  84         A context manager, from an initially unmounted state, to mount
  85         this, yield, and then unmount and clean up.
  86         """
  87         self.mount()
  88         self.wait_until_mounted()
  89         try:
  90             yield
  91         finally:
  92             self.umount_wait()
  93
  94     def create_files(self):
  95         assert(self.is_mounted())
  96
  97         for suffix in self.test_files:
  98             log.info("Creating file {0}".format(suffix))
  99             self.client_remote.run(args=[
 100                 'sudo', 'touch', os.path.join(self.mountpoint, suffix)
 101             ])
 102
 103     def check_files(self):
 104         assert(self.is_mounted())
 105
 106         for suffix in self.test_files:
 107             log.info("Checking file {0}".format(suffix))
 108             r = self.client_remote.run(args=[
 109                 'sudo', 'ls', os.path.join(self.mountpoint, suffix)
 110             ], check_status=False)
 111             if r.exitstatus != 0:
 112                 raise RuntimeError("Expected file {0} not found".format(suffix))
 113
 114     def create_destroy(self):
 115         assert(self.is_mounted())
 116
 117         filename = "{0} {1}".format(datetime.datetime.now(), self.client_id)
 118         log.debug("Creating test file {0}".format(filename))
 119         self.client_remote.run(args=[
 120             'sudo', 'touch', os.path.join(self.mountpoint, filename)
 121         ])
 122         log.debug("Deleting test file {0}".format(filename))
 123         self.client_remote.run(args=[
 124             'sudo', 'rm', '-f', os.path.join(self.mountpoint, filename)
 125         ])
 126
 127     def _run_python(self, pyscript, py_version='python'):
 128         return self.client_remote.run(
 129                args=['sudo', 'adjust-ulimits', 'daemon-helper', 'kill',
 130                      py_version, '-c', pyscript], wait=False, stdin=run.PIPE,
 131                stdout=StringIO())
 132
 133     def run_python(self, pyscript, py_version='python'):
 134         p = self._run_python(pyscript, py_version)
 135         p.wait()
 136         return p.stdout.getvalue().strip()
 137
 138     def run_shell(self, args, wait=True):
 139         args = ["cd", self.mountpoint, run.Raw('&&'), "sudo"] + args
 140         return self.client_remote.run(args=args, stdout=StringIO(),
 141                                       stderr=StringIO(), wait=wait)
 142
 143     def open_no_data(self, basename):
 144         """
 145         A pure metadata operation
 146         """
 147         assert(self.is_mounted())
 148
 149         path = os.path.join(self.mountpoint, basename)
 150
 151         p = self._run_python(dedent(
 152             """
 153             f = open("{path}", 'w')
 154             """.format(path=path)
 155         ))
 156         p.wait()
 157
 158     def open_background(self, basename="background_file"):
 159         """
 160         Open a file for writing, then block such that the client
 161         will hold a capability.
 162
 163         Don't return until the remote process has got as far as opening
 164         the file, then return the RemoteProcess instance.
 165         """
 166         assert(self.is_mounted())
 167
 168         path = os.path.join(self.mountpoint, basename)
 169
 170         pyscript = dedent("""
 171             import time
 172
 173             f = open("{path}", 'w')
 174             f.write('content')
 175             f.flush()
 176             f.write('content2')
 177             while True:
 178                 time.sleep(1)
 179             """).format(path=path)
 180
 181         rproc = self._run_python(pyscript)
 182         self.background_procs.append(rproc)
 183
 184         # This wait would not be sufficient if the file had already
 185         # existed, but it's simple and in practice users of open_background
 186         # are not using it on existing files.
 187         self.wait_for_visible(basename)
 188
 189         return rproc
 190
 191     def wait_for_visible(self, basename="background_file", timeout=30):
 192         i = 0
 193         while i < timeout:
 194             r = self.client_remote.run(args=[
 195                 'sudo', 'ls', os.path.join(self.mountpoint, basename)
 196             ], check_status=False)
 197             if r.exitstatus == 0:
 198                 log.debug("File {0} became visible from {1} after {2}s".format(
 199                     basename, self.client_id, i))
 200                 return
 201             else:
 202                 time.sleep(1)
 203                 i += 1
 204
 205         raise RuntimeError("Timed out after {0}s waiting for {1} to become visible from {2}".format(
 206             i, basename, self.client_id))
 207
 208     def lock_background(self, basename="background_file", do_flock=True):
 209         """
 210         Open and lock a files for writing, hold the lock in a background process
 211         """
 212         assert(self.is_mounted())
 213
 214         path = os.path.join(self.mountpoint, basename)
 215
 216         script_builder = """
 217             import time
 218             import fcntl
 219             import struct"""
 220         if do_flock:
 221             script_builder += """
 222             f1 = open("{path}-1", 'w')
 223             fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)"""
 224         script_builder += """
 225             f2 = open("{path}-2", 'w')
 226             lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 227             fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 228             while True:
 229                 time.sleep(1)
 230             """
 231
 232         pyscript = dedent(script_builder).format(path=path)
 233
 234         log.info("lock_background file {0}".format(basename))
 235         rproc = self._run_python(pyscript)
 236         self.background_procs.append(rproc)
 237         return rproc
 238
 239     def lock_and_release(self, basename="background_file"):
 240         assert(self.is_mounted())
 241
 242         path = os.path.join(self.mountpoint, basename)
 243
 244         script = """
 245             import time
 246             import fcntl
 247             import struct
 248             f1 = open("{path}-1", 'w')
 249             fcntl.flock(f1, fcntl.LOCK_EX)
 250             f2 = open("{path}-2", 'w')
 251             lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 252             fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 253             """
 254         pyscript = dedent(script).format(path=path)
 255
 256         log.info("lock_and_release file {0}".format(basename))
 257         return self._run_python(pyscript)
 258
 259     def check_filelock(self, basename="background_file", do_flock=True):
 260         assert(self.is_mounted())
 261
 262         path = os.path.join(self.mountpoint, basename)
 263
 264         script_builder = """
 265             import fcntl
 266             import errno
 267             import struct"""
 268         if do_flock:
 269             script_builder += """
 270             f1 = open("{path}-1", 'r')
 271             try:
 272                 fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)
 273             except IOError, e:
 274                 if e.errno == errno.EAGAIN:
 275                     pass
 276             else:
 277                 raise RuntimeError("flock on file {path}-1 not found")"""
 278         script_builder += """
 279             f2 = open("{path}-2", 'r')
 280             try:
 281                 lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 282                 fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 283             except IOError, e:
 284                 if e.errno == errno.EAGAIN:
 285                     pass
 286             else:
 287                 raise RuntimeError("posix lock on file {path}-2 not found")
 288             """
 289         pyscript = dedent(script_builder).format(path=path)
 290
 291         log.info("check lock on file {0}".format(basename))
 292         self.client_remote.run(args=[
 293             'sudo', 'python', '-c', pyscript
 294         ])
 295
 296     def write_background(self, basename="background_file", loop=False):
 297         """
 298         Open a file for writing, complete as soon as you can
 299         :param basename:
 300         :return:
 301         """
 302         assert(self.is_mounted())
 303
 304         path = os.path.join(self.mountpoint, basename)
 305
 306         pyscript = dedent("""
 307             import os
 308             import time
 309
 310             fd = os.open("{path}", os.O_RDWR | os.O_CREAT, 0644)
 311             try:
 312                 while True:
 313                     os.write(fd, 'content')
 314                     time.sleep(1)
 315                     if not {loop}:
 316                         break
 317             except IOError, e:
 318                 pass
 319             os.close(fd)
 320             """).format(path=path, loop=str(loop))
 321
 322         rproc = self._run_python(pyscript)
 323         self.background_procs.append(rproc)
 324         return rproc
 325
 326     def write_n_mb(self, filename, n_mb, seek=0, wait=True):
 327         """
 328         Write the requested number of megabytes to a file
 329         """
 330         assert(self.is_mounted())
 331
 332         return self.run_shell(["dd", "if=/dev/urandom", "of={0}".format(filename),
 333                                "bs=1M", "conv=fdatasync",
 334                                "count={0}".format(n_mb),
 335                                "seek={0}".format(seek)
 336                                ], wait=wait)
 337
 338     def write_test_pattern(self, filename, size):
 339         log.info("Writing {0} bytes to {1}".format(size, filename))
 340         return self.run_python(dedent("""
 341             import zlib
 342             path = "{path}"
 343             f = open(path, 'w')
 344             for i in range(0, {size}):
 345                 val = zlib.crc32("%s" % i) & 7
 346                 f.write(chr(val))
 347             f.close()
 348         """.format(
 349             path=os.path.join(self.mountpoint, filename),
 350             size=size
 351         )))
 352
 353     def validate_test_pattern(self, filename, size):
 354         log.info("Validating {0} bytes from {1}".format(size, filename))
 355         return self.run_python(dedent("""
 356             import zlib
 357             path = "{path}"
 358             f = open(path, 'r')
 359             bytes = f.read()
 360             f.close()
 361             if len(bytes) != {size}:
 362                 raise RuntimeError("Bad length {{0}} vs. expected {{1}}".format(
 363                     len(bytes), {size}
 364                 ))
 365             for i, b in enumerate(bytes):
 366                 val = zlib.crc32("%s" % i) & 7
 367                 if b != chr(val):
 368                     raise RuntimeError("Bad data at offset {{0}}".format(i))
 369         """.format(
 370             path=os.path.join(self.mountpoint, filename),
 371             size=size
 372         )))
 373
 374     def open_n_background(self, fs_path, count):
 375         """
 376         Open N files for writing, hold them open in a background process
 377
 378         :param fs_path: Path relative to CephFS root, e.g. "foo/bar"
 379         :return: a RemoteProcess
 380         """
 381         assert(self.is_mounted())
 382
 383         abs_path = os.path.join(self.mountpoint, fs_path)
 384
 385         pyscript = dedent("""
 386             import sys
 387             import time
 388             import os
 389
 390             n = {count}
 391             abs_path = "{abs_path}"
 392
 393             if not os.path.exists(os.path.dirname(abs_path)):
 394                 os.makedirs(os.path.dirname(abs_path))
 395
 396             handles = []
 397             for i in range(0, n):
 398                 fname = "{{0}}_{{1}}".format(abs_path, i)
 399                 handles.append(open(fname, 'w'))
 400
 401             while True:
 402                 time.sleep(1)
 403             """).format(abs_path=abs_path, count=count)
 404
 405         rproc = self._run_python(pyscript)
 406         self.background_procs.append(rproc)
 407         return rproc
 408
 409     def create_n_files(self, fs_path, count, sync=False):
 410         assert(self.is_mounted())
 411
 412         abs_path = os.path.join(self.mountpoint, fs_path)
 413
 414         pyscript = dedent("""
 415             import sys
 416             import time
 417             import os
 418
 419             n = {count}
 420             abs_path = "{abs_path}"
 421
 422             if not os.path.exists(os.path.dirname(abs_path)):
 423                 os.makedirs(os.path.dirname(abs_path))
 424
 425             for i in range(0, n):
 426                 fname = "{{0}}_{{1}}".format(abs_path, i)
 427                 h = open(fname, 'w')
 428                 h.write('content')
 429                 if {sync}:
 430                     h.flush()
 431                     os.fsync(h.fileno())
 432                 h.close()
 433             """).format(abs_path=abs_path, count=count, sync=str(sync))
 434
 435         self.run_python(pyscript)
 436
 437     def teardown(self):
 438         for p in self.background_procs:
 439             log.info("Terminating background process")
 440             self._kill_background(p)
 441
 442         self.background_procs = []
 443
 444     def _kill_background(self, p):
 445         if p.stdin:
 446             p.stdin.close()
 447             try:
 448                 p.wait()
 449             except (CommandFailedError, ConnectionLostError):
 450                 pass
 451
 452     def kill_background(self, p):
 453         """
 454         For a process that was returned by one of the _background member functions,
 455         kill it hard.
 456         """
 457         self._kill_background(p)
 458         self.background_procs.remove(p)
 459
 460     def get_global_id(self):
 461         raise NotImplementedError()
 462
 463     def get_osd_epoch(self):
 464         raise NotImplementedError()
 465
 466     def stat(self, fs_path, wait=True):
 467         """
 468         stat a file, and return the result as a dictionary like this:
 469         {
 470           "st_ctime": 1414161137.0,
 471           "st_mtime": 1414161137.0,
 472           "st_nlink": 33,
 473           "st_gid": 0,
 474           "st_dev": 16777218,
 475           "st_size": 1190,
 476           "st_ino": 2,
 477           "st_uid": 0,
 478           "st_mode": 16877,
 479           "st_atime": 1431520593.0
 480         }
 481
 482         Raises exception on absent file.
 483         """
 484         abs_path = os.path.join(self.mountpoint, fs_path)
 485
 486         pyscript = dedent("""
 487             import os
 488             import stat
 489             import json
 490             import sys
 491
 492             try:
 493                 s = os.stat("{path}")
 494             except OSError as e:
 495                 sys.exit(e.errno)
 496
 497             attrs = ["st_mode", "st_ino", "st_dev", "st_nlink", "st_uid", "st_gid", "st_size", "st_atime", "st_mtime", "st_ctime"]
 498             print json.dumps(
 499                 dict([(a, getattr(s, a)) for a in attrs]),
 500                 indent=2)
 501             """).format(path=abs_path)
 502         proc = self._run_python(pyscript)
 503         if wait:
 504             proc.wait()
 505             return json.loads(proc.stdout.getvalue().strip())
 506         else:
 507             return proc
 508
 509     def touch(self, fs_path):
 510         """
 511         Create a dentry if it doesn't already exist.  This python
 512         implementation exists because the usual command line tool doesn't
 513         pass through error codes like EIO.
 514
 515         :param fs_path:
 516         :return:
 517         """
 518         abs_path = os.path.join(self.mountpoint, fs_path)
 519         pyscript = dedent("""
 520             import sys
 521             import errno
 522
 523             try:
 524                 f = open("{path}", "w")
 525                 f.close()
 526             except IOError as e:
 527                 sys.exit(errno.EIO)
 528             """).format(path=abs_path)
 529         proc = self._run_python(pyscript)
 530         proc.wait()
 531
 532     def path_to_ino(self, fs_path, follow_symlinks=True):
 533         abs_path = os.path.join(self.mountpoint, fs_path)
 534
 535         if follow_symlinks:
 536             pyscript = dedent("""
 537                 import os
 538                 import stat
 539
 540                 print os.stat("{path}").st_ino
 541                 """).format(path=abs_path)
 542         else:
 543             pyscript = dedent("""
 544                 import os
 545                 import stat
 546
 547                 print os.lstat("{path}").st_ino
 548                 """).format(path=abs_path)
 549
 550         proc = self._run_python(pyscript)
 551         proc.wait()
 552         return int(proc.stdout.getvalue().strip())
 553
 554     def path_to_nlink(self, fs_path):
 555         abs_path = os.path.join(self.mountpoint, fs_path)
 556
 557         pyscript = dedent("""
 558             import os
 559             import stat
 560
 561             print os.stat("{path}").st_nlink
 562             """).format(path=abs_path)
 563
 564         proc = self._run_python(pyscript)
 565         proc.wait()
 566         return int(proc.stdout.getvalue().strip())
 567
 568     def ls(self, path=None):
 569         """
 570         Wrap ls: return a list of strings
 571         """
 572         cmd = ["ls"]
 573         if path:
 574             cmd.append(path)
 575
 576         ls_text = self.run_shell(cmd).stdout.getvalue().strip()
 577
 578         if ls_text:
 579             return ls_text.split("\n")
 580         else:
 581             # Special case because otherwise split on empty string
 582             # gives you [''] instead of []
 583             return []
 584
 585     def setfattr(self, path, key, val):
 586         """
 587         Wrap setfattr.
 588
 589         :param path: relative to mount point
 590         :param key: xattr name
 591         :param val: xattr value
 592         :return: None
 593         """
 594         self.run_shell(["setfattr", "-n", key, "-v", val, path])
 595
 596     def getfattr(self, path, attr):
 597         """
 598         Wrap getfattr: return the values of a named xattr on one file, or
 599         None if the attribute is not found.
 600
 601         :return: a string
 602         """
 603         p = self.run_shell(["getfattr", "--only-values", "-n", attr, path], wait=False)
 604         try:
 605             p.wait()
 606         except CommandFailedError as e:
 607             if e.exitstatus == 1 and "No such attribute" in p.stderr.getvalue():
 608                 return None
 609             else:
 610                 raise
 611
 612         return p.stdout.getvalue()
 613
 614     def df(self):
 615         """
 616         Wrap df: return a dict of usage fields in bytes
 617         """
 618
 619         p = self.run_shell(["df", "-B1", "."])
 620         lines = p.stdout.getvalue().strip().split("\n")
 621         fs, total, used, avail = lines[1].split()[:4]
 622         log.warn(lines)
 623
 624         return {
 625             "total": int(total),
 626             "used": int(used),
 627             "available": int(avail)
 628         }