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):
 128         return self.client_remote.run(args=[
 129             'sudo', 'adjust-ulimits', 'daemon-helper', 'kill', 'python', '-c', pyscript
 130         ], wait=False, stdin=run.PIPE, stdout=StringIO())
 131
 132     def run_python(self, pyscript):
 133         p = self._run_python(pyscript)
 134         p.wait()
 135         return p.stdout.getvalue().strip()
 136
 137     def run_shell(self, args, wait=True):
 138         args = ["cd", self.mountpoint, run.Raw('&&'), "sudo"] + args
 139         return self.client_remote.run(args=args, stdout=StringIO(),
 140                                       stderr=StringIO(), wait=wait)
 141
 142     def open_no_data(self, basename):
 143         """
 144         A pure metadata operation
 145         """
 146         assert(self.is_mounted())
 147
 148         path = os.path.join(self.mountpoint, basename)
 149
 150         p = self._run_python(dedent(
 151             """
 152             f = open("{path}", 'w')
 153             """.format(path=path)
 154         ))
 155         p.wait()
 156
 157     def open_background(self, basename="background_file"):
 158         """
 159         Open a file for writing, then block such that the client
 160         will hold a capability.
 161
 162         Don't return until the remote process has got as far as opening
 163         the file, then return the RemoteProcess instance.
 164         """
 165         assert(self.is_mounted())
 166
 167         path = os.path.join(self.mountpoint, basename)
 168
 169         pyscript = dedent("""
 170             import time
 171
 172             f = open("{path}", 'w')
 173             f.write('content')
 174             f.flush()
 175             f.write('content2')
 176             while True:
 177                 time.sleep(1)
 178             """).format(path=path)
 179
 180         rproc = self._run_python(pyscript)
 181         self.background_procs.append(rproc)
 182
 183         # This wait would not be sufficient if the file had already
 184         # existed, but it's simple and in practice users of open_background
 185         # are not using it on existing files.
 186         self.wait_for_visible(basename)
 187
 188         return rproc
 189
 190     def wait_for_visible(self, basename="background_file", timeout=30):
 191         i = 0
 192         while i < timeout:
 193             r = self.client_remote.run(args=[
 194                 'sudo', 'ls', os.path.join(self.mountpoint, basename)
 195             ], check_status=False)
 196             if r.exitstatus == 0:
 197                 log.debug("File {0} became visible from {1} after {2}s".format(
 198                     basename, self.client_id, i))
 199                 return
 200             else:
 201                 time.sleep(1)
 202                 i += 1
 203
 204         raise RuntimeError("Timed out after {0}s waiting for {1} to become visible from {2}".format(
 205             i, basename, self.client_id))
 206
 207     def lock_background(self, basename="background_file", do_flock=True):
 208         """
 209         Open and lock a files for writing, hold the lock in a background process
 210         """
 211         assert(self.is_mounted())
 212
 213         path = os.path.join(self.mountpoint, basename)
 214
 215         script_builder = """
 216             import time
 217             import fcntl
 218             import struct"""
 219         if do_flock:
 220             script_builder += """
 221             f1 = open("{path}-1", 'w')
 222             fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)"""
 223         script_builder += """
 224             f2 = open("{path}-2", 'w')
 225             lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 226             fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 227             while True:
 228                 time.sleep(1)
 229             """
 230
 231         pyscript = dedent(script_builder).format(path=path)
 232
 233         log.info("lock file {0}".format(basename))
 234         rproc = self._run_python(pyscript)
 235         self.background_procs.append(rproc)
 236         return rproc
 237
 238     def check_filelock(self, basename="background_file", do_flock=True):
 239         assert(self.is_mounted())
 240
 241         path = os.path.join(self.mountpoint, basename)
 242
 243         script_builder = """
 244             import fcntl
 245             import errno
 246             import struct"""
 247         if do_flock:
 248             script_builder += """
 249             f1 = open("{path}-1", 'r')
 250             try:
 251                 fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)
 252             except IOError, e:
 253                 if e.errno == errno.EAGAIN:
 254                     pass
 255             else:
 256                 raise RuntimeError("flock on file {path}-1 not found")"""
 257         script_builder += """
 258             f2 = open("{path}-2", 'r')
 259             try:
 260                 lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 261                 fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 262             except IOError, e:
 263                 if e.errno == errno.EAGAIN:
 264                     pass
 265             else:
 266                 raise RuntimeError("posix lock on file {path}-2 not found")
 267             """
 268         pyscript = dedent(script_builder).format(path=path)
 269
 270         log.info("check lock on file {0}".format(basename))
 271         self.client_remote.run(args=[
 272             'sudo', 'python', '-c', pyscript
 273         ])
 274
 275     def write_background(self, basename="background_file", loop=False):
 276         """
 277         Open a file for writing, complete as soon as you can
 278         :param basename:
 279         :return:
 280         """
 281         assert(self.is_mounted())
 282
 283         path = os.path.join(self.mountpoint, basename)
 284
 285         pyscript = dedent("""
 286             import os
 287             import time
 288
 289             fd = os.open("{path}", os.O_RDWR | os.O_CREAT, 0644)
 290             try:
 291                 while True:
 292                     os.write(fd, 'content')
 293                     time.sleep(1)
 294                     if not {loop}:
 295                         break
 296             except IOError, e:
 297                 pass
 298             os.close(fd)
 299             """).format(path=path, loop=str(loop))
 300
 301         rproc = self._run_python(pyscript)
 302         self.background_procs.append(rproc)
 303         return rproc
 304
 305     def write_n_mb(self, filename, n_mb, seek=0, wait=True):
 306         """
 307         Write the requested number of megabytes to a file
 308         """
 309         assert(self.is_mounted())
 310
 311         return self.run_shell(["dd", "if=/dev/urandom", "of={0}".format(filename),
 312                                "bs=1M", "conv=fdatasync",
 313                                "count={0}".format(n_mb),
 314                                "seek={0}".format(seek)
 315                                ], wait=wait)
 316
 317     def write_test_pattern(self, filename, size):
 318         log.info("Writing {0} bytes to {1}".format(size, filename))
 319         return self.run_python(dedent("""
 320             import zlib
 321             path = "{path}"
 322             f = open(path, 'w')
 323             for i in range(0, {size}):
 324                 val = zlib.crc32("%s" % i) & 7
 325                 f.write(chr(val))
 326             f.close()
 327         """.format(
 328             path=os.path.join(self.mountpoint, filename),
 329             size=size
 330         )))
 331
 332     def validate_test_pattern(self, filename, size):
 333         log.info("Validating {0} bytes from {1}".format(size, filename))
 334         return self.run_python(dedent("""
 335             import zlib
 336             path = "{path}"
 337             f = open(path, 'r')
 338             bytes = f.read()
 339             f.close()
 340             if len(bytes) != {size}:
 341                 raise RuntimeError("Bad length {{0}} vs. expected {{1}}".format(
 342                     len(bytes), {size}
 343                 ))
 344             for i, b in enumerate(bytes):
 345                 val = zlib.crc32("%s" % i) & 7
 346                 if b != chr(val):
 347                     raise RuntimeError("Bad data at offset {{0}}".format(i))
 348         """.format(
 349             path=os.path.join(self.mountpoint, filename),
 350             size=size
 351         )))
 352
 353     def open_n_background(self, fs_path, count):
 354         """
 355         Open N files for writing, hold them open in a background process
 356
 357         :param fs_path: Path relative to CephFS root, e.g. "foo/bar"
 358         :return: a RemoteProcess
 359         """
 360         assert(self.is_mounted())
 361
 362         abs_path = os.path.join(self.mountpoint, fs_path)
 363
 364         pyscript = dedent("""
 365             import sys
 366             import time
 367             import os
 368
 369             n = {count}
 370             abs_path = "{abs_path}"
 371
 372             if not os.path.exists(os.path.dirname(abs_path)):
 373                 os.makedirs(os.path.dirname(abs_path))
 374
 375             handles = []
 376             for i in range(0, n):
 377                 fname = "{{0}}_{{1}}".format(abs_path, i)
 378                 handles.append(open(fname, 'w'))
 379
 380             while True:
 381                 time.sleep(1)
 382             """).format(abs_path=abs_path, count=count)
 383
 384         rproc = self._run_python(pyscript)
 385         self.background_procs.append(rproc)
 386         return rproc
 387
 388     def create_n_files(self, fs_path, count, sync=False):
 389         assert(self.is_mounted())
 390
 391         abs_path = os.path.join(self.mountpoint, fs_path)
 392
 393         pyscript = dedent("""
 394             import sys
 395             import time
 396             import os
 397
 398             n = {count}
 399             abs_path = "{abs_path}"
 400
 401             if not os.path.exists(os.path.dirname(abs_path)):
 402                 os.makedirs(os.path.dirname(abs_path))
 403
 404             for i in range(0, n):
 405                 fname = "{{0}}_{{1}}".format(abs_path, i)
 406                 h = open(fname, 'w')
 407                 h.write('content')
 408                 if {sync}:
 409                     h.flush()
 410                     os.fsync(h.fileno())
 411                 h.close()
 412             """).format(abs_path=abs_path, count=count, sync=str(sync))
 413
 414         self.run_python(pyscript)
 415
 416     def teardown(self):
 417         for p in self.background_procs:
 418             log.info("Terminating background process")
 419             self._kill_background(p)
 420
 421         self.background_procs = []
 422
 423     def _kill_background(self, p):
 424         if p.stdin:
 425             p.stdin.close()
 426             try:
 427                 p.wait()
 428             except (CommandFailedError, ConnectionLostError):
 429                 pass
 430
 431     def kill_background(self, p):
 432         """
 433         For a process that was returned by one of the _background member functions,
 434         kill it hard.
 435         """
 436         self._kill_background(p)
 437         self.background_procs.remove(p)
 438
 439     def spam_dir_background(self, path):
 440         """
 441         Create directory `path` and do lots of metadata operations
 442         in it until further notice.
 443         """
 444         assert(self.is_mounted())
 445         abs_path = os.path.join(self.mountpoint, path)
 446
 447         pyscript = dedent("""
 448             import sys
 449             import time
 450             import os
 451
 452             abs_path = "{abs_path}"
 453
 454             if not os.path.exists(abs_path):
 455                 os.makedirs(abs_path)
 456
 457             n = 0
 458             while True:
 459                 file_path = os.path.join(abs_path, "tmp%d" % n)
 460                 f = open(file_path, 'w')
 461                 f.close()
 462                 n = n + 1
 463             """).format(abs_path=abs_path)
 464
 465         rproc = self._run_python(pyscript)
 466         self.background_procs.append(rproc)
 467         return rproc
 468
 469     def get_global_id(self):
 470         raise NotImplementedError()
 471
 472     def get_osd_epoch(self):
 473         raise NotImplementedError()
 474
 475     def stat(self, fs_path, wait=True):
 476         """
 477         stat a file, and return the result as a dictionary like this:
 478         {
 479           "st_ctime": 1414161137.0,
 480           "st_mtime": 1414161137.0,
 481           "st_nlink": 33,
 482           "st_gid": 0,
 483           "st_dev": 16777218,
 484           "st_size": 1190,
 485           "st_ino": 2,
 486           "st_uid": 0,
 487           "st_mode": 16877,
 488           "st_atime": 1431520593.0
 489         }
 490
 491         Raises exception on absent file.
 492         """
 493         abs_path = os.path.join(self.mountpoint, fs_path)
 494
 495         pyscript = dedent("""
 496             import os
 497             import stat
 498             import json
 499             import sys
 500
 501             try:
 502                 s = os.stat("{path}")
 503             except OSError as e:
 504                 sys.exit(e.errno)
 505
 506             attrs = ["st_mode", "st_ino", "st_dev", "st_nlink", "st_uid", "st_gid", "st_size", "st_atime", "st_mtime", "st_ctime"]
 507             print json.dumps(
 508                 dict([(a, getattr(s, a)) for a in attrs]),
 509                 indent=2)
 510             """).format(path=abs_path)
 511         proc = self._run_python(pyscript)
 512         if wait:
 513             proc.wait()
 514             return json.loads(proc.stdout.getvalue().strip())
 515         else:
 516             return proc
 517
 518     def touch(self, fs_path):
 519         """
 520         Create a dentry if it doesn't already exist.  This python
 521         implementation exists because the usual command line tool doesn't
 522         pass through error codes like EIO.
 523
 524         :param fs_path:
 525         :return:
 526         """
 527         abs_path = os.path.join(self.mountpoint, fs_path)
 528         pyscript = dedent("""
 529             import sys
 530             import errno
 531
 532             try:
 533                 f = open("{path}", "w")
 534                 f.close()
 535             except IOError as e:
 536                 sys.exit(errno.EIO)
 537             """).format(path=abs_path)
 538         proc = self._run_python(pyscript)
 539         proc.wait()
 540
 541     def path_to_ino(self, fs_path, follow_symlinks=True):
 542         abs_path = os.path.join(self.mountpoint, fs_path)
 543
 544         if follow_symlinks:
 545             pyscript = dedent("""
 546                 import os
 547                 import stat
 548
 549                 print os.stat("{path}").st_ino
 550                 """).format(path=abs_path)
 551         else:
 552             pyscript = dedent("""
 553                 import os
 554                 import stat
 555
 556                 print os.lstat("{path}").st_ino
 557                 """).format(path=abs_path)
 558
 559         proc = self._run_python(pyscript)
 560         proc.wait()
 561         return int(proc.stdout.getvalue().strip())
 562
 563     def path_to_nlink(self, fs_path):
 564         abs_path = os.path.join(self.mountpoint, fs_path)
 565
 566         pyscript = dedent("""
 567             import os
 568             import stat
 569
 570             print os.stat("{path}").st_nlink
 571             """).format(path=abs_path)
 572
 573         proc = self._run_python(pyscript)
 574         proc.wait()
 575         return int(proc.stdout.getvalue().strip())
 576
 577     def ls(self, path=None):
 578         """
 579         Wrap ls: return a list of strings
 580         """
 581         cmd = ["ls"]
 582         if path:
 583             cmd.append(path)
 584
 585         ls_text = self.run_shell(cmd).stdout.getvalue().strip()
 586
 587         if ls_text:
 588             return ls_text.split("\n")
 589         else:
 590             # Special case because otherwise split on empty string
 591             # gives you [''] instead of []
 592             return []
 593
 594     def setfattr(self, path, key, val):
 595         """
 596         Wrap setfattr.
 597
 598         :param path: relative to mount point
 599         :param key: xattr name
 600         :param val: xattr value
 601         :return: None
 602         """
 603         self.run_shell(["setfattr", "-n", key, "-v", val, path])
 604
 605     def getfattr(self, path, attr):
 606         """
 607         Wrap getfattr: return the values of a named xattr on one file, or
 608         None if the attribute is not found.
 609
 610         :return: a string
 611         """
 612         p = self.run_shell(["getfattr", "--only-values", "-n", attr, path], wait=False)
 613         try:
 614             p.wait()
 615         except CommandFailedError as e:
 616             if e.exitstatus == 1 and "No such attribute" in p.stderr.getvalue():
 617                 return None
 618             else:
 619                 raise
 620
 621         return p.stdout.getvalue()
 622
 623     def df(self):
 624         """
 625         Wrap df: return a dict of usage fields in bytes
 626         """
 627
 628         p = self.run_shell(["df", "-B1", "."])
 629         lines = p.stdout.getvalue().strip().split("\n")
 630         fs, total, used, avail = lines[1].split()[:4]
 631         log.warn(lines)
 632
 633         return {
 634             "total": int(total),
 635             "used": int(used),
 636             "available": int(avail)
 637         }