src/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_background file {0}".format(basename))
 234         rproc = self._run_python(pyscript)
 235         self.background_procs.append(rproc)
 236         return rproc
 237
 238     def lock_and_release(self, basename="background_file"):
 239         assert(self.is_mounted())
 240
 241         path = os.path.join(self.mountpoint, basename)
 242
 243         script = """
 244             import time
 245             import fcntl
 246             import struct
 247             f1 = open("{path}-1", 'w')
 248             fcntl.flock(f1, fcntl.LOCK_EX)
 249             f2 = open("{path}-2", 'w')
 250             lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 251             fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 252             """
 253         pyscript = dedent(script).format(path=path)
 254
 255         log.info("lock_and_release file {0}".format(basename))
 256         return self._run_python(pyscript)
 257
 258     def check_filelock(self, basename="background_file", do_flock=True):
 259         assert(self.is_mounted())
 260
 261         path = os.path.join(self.mountpoint, basename)
 262
 263         script_builder = """
 264             import fcntl
 265             import errno
 266             import struct"""
 267         if do_flock:
 268             script_builder += """
 269             f1 = open("{path}-1", 'r')
 270             try:
 271                 fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)
 272             except IOError, e:
 273                 if e.errno == errno.EAGAIN:
 274                     pass
 275             else:
 276                 raise RuntimeError("flock on file {path}-1 not found")"""
 277         script_builder += """
 278             f2 = open("{path}-2", 'r')
 279             try:
 280                 lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 281                 fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 282             except IOError, e:
 283                 if e.errno == errno.EAGAIN:
 284                     pass
 285             else:
 286                 raise RuntimeError("posix lock on file {path}-2 not found")
 287             """
 288         pyscript = dedent(script_builder).format(path=path)
 289
 290         log.info("check lock on file {0}".format(basename))
 291         self.client_remote.run(args=[
 292             'sudo', 'python', '-c', pyscript
 293         ])
 294
 295     def write_background(self, basename="background_file", loop=False):
 296         """
 297         Open a file for writing, complete as soon as you can
 298         :param basename:
 299         :return:
 300         """
 301         assert(self.is_mounted())
 302
 303         path = os.path.join(self.mountpoint, basename)
 304
 305         pyscript = dedent("""
 306             import os
 307             import time
 308
 309             fd = os.open("{path}", os.O_RDWR | os.O_CREAT, 0644)
 310             try:
 311                 while True:
 312                     os.write(fd, 'content')
 313                     time.sleep(1)
 314                     if not {loop}:
 315                         break
 316             except IOError, e:
 317                 pass
 318             os.close(fd)
 319             """).format(path=path, loop=str(loop))
 320
 321         rproc = self._run_python(pyscript)
 322         self.background_procs.append(rproc)
 323         return rproc
 324
 325     def write_n_mb(self, filename, n_mb, seek=0, wait=True):
 326         """
 327         Write the requested number of megabytes to a file
 328         """
 329         assert(self.is_mounted())
 330
 331         return self.run_shell(["dd", "if=/dev/urandom", "of={0}".format(filename),
 332                                "bs=1M", "conv=fdatasync",
 333                                "count={0}".format(n_mb),
 334                                "seek={0}".format(seek)
 335                                ], wait=wait)
 336
 337     def write_test_pattern(self, filename, size):
 338         log.info("Writing {0} bytes to {1}".format(size, filename))
 339         return self.run_python(dedent("""
 340             import zlib
 341             path = "{path}"
 342             f = open(path, 'w')
 343             for i in range(0, {size}):
 344                 val = zlib.crc32("%s" % i) & 7
 345                 f.write(chr(val))
 346             f.close()
 347         """.format(
 348             path=os.path.join(self.mountpoint, filename),
 349             size=size
 350         )))
 351
 352     def validate_test_pattern(self, filename, size):
 353         log.info("Validating {0} bytes from {1}".format(size, filename))
 354         return self.run_python(dedent("""
 355             import zlib
 356             path = "{path}"
 357             f = open(path, 'r')
 358             bytes = f.read()
 359             f.close()
 360             if len(bytes) != {size}:
 361                 raise RuntimeError("Bad length {{0}} vs. expected {{1}}".format(
 362                     len(bytes), {size}
 363                 ))
 364             for i, b in enumerate(bytes):
 365                 val = zlib.crc32("%s" % i) & 7
 366                 if b != chr(val):
 367                     raise RuntimeError("Bad data at offset {{0}}".format(i))
 368         """.format(
 369             path=os.path.join(self.mountpoint, filename),
 370             size=size
 371         )))
 372
 373     def open_n_background(self, fs_path, count):
 374         """
 375         Open N files for writing, hold them open in a background process
 376
 377         :param fs_path: Path relative to CephFS root, e.g. "foo/bar"
 378         :return: a RemoteProcess
 379         """
 380         assert(self.is_mounted())
 381
 382         abs_path = os.path.join(self.mountpoint, fs_path)
 383
 384         pyscript = dedent("""
 385             import sys
 386             import time
 387             import os
 388
 389             n = {count}
 390             abs_path = "{abs_path}"
 391
 392             if not os.path.exists(os.path.dirname(abs_path)):
 393                 os.makedirs(os.path.dirname(abs_path))
 394
 395             handles = []
 396             for i in range(0, n):
 397                 fname = "{{0}}_{{1}}".format(abs_path, i)
 398                 handles.append(open(fname, 'w'))
 399
 400             while True:
 401                 time.sleep(1)
 402             """).format(abs_path=abs_path, count=count)
 403
 404         rproc = self._run_python(pyscript)
 405         self.background_procs.append(rproc)
 406         return rproc
 407
 408     def create_n_files(self, fs_path, count, sync=False):
 409         assert(self.is_mounted())
 410
 411         abs_path = os.path.join(self.mountpoint, fs_path)
 412
 413         pyscript = dedent("""
 414             import sys
 415             import time
 416             import os
 417
 418             n = {count}
 419             abs_path = "{abs_path}"
 420
 421             if not os.path.exists(os.path.dirname(abs_path)):
 422                 os.makedirs(os.path.dirname(abs_path))
 423
 424             for i in range(0, n):
 425                 fname = "{{0}}_{{1}}".format(abs_path, i)
 426                 h = open(fname, 'w')
 427                 h.write('content')
 428                 if {sync}:
 429                     h.flush()
 430                     os.fsync(h.fileno())
 431                 h.close()
 432             """).format(abs_path=abs_path, count=count, sync=str(sync))
 433
 434         self.run_python(pyscript)
 435
 436     def teardown(self):
 437         for p in self.background_procs:
 438             log.info("Terminating background process")
 439             self._kill_background(p)
 440
 441         self.background_procs = []
 442
 443     def _kill_background(self, p):
 444         if p.stdin:
 445             p.stdin.close()
 446             try:
 447                 p.wait()
 448             except (CommandFailedError, ConnectionLostError):
 449                 pass
 450
 451     def kill_background(self, p):
 452         """
 453         For a process that was returned by one of the _background member functions,
 454         kill it hard.
 455         """
 456         self._kill_background(p)
 457         self.background_procs.remove(p)
 458
 459     def get_global_id(self):
 460         raise NotImplementedError()
 461
 462     def get_osd_epoch(self):
 463         raise NotImplementedError()
 464
 465     def stat(self, fs_path, wait=True):
 466         """
 467         stat a file, and return the result as a dictionary like this:
 468         {
 469           "st_ctime": 1414161137.0,
 470           "st_mtime": 1414161137.0,
 471           "st_nlink": 33,
 472           "st_gid": 0,
 473           "st_dev": 16777218,
 474           "st_size": 1190,
 475           "st_ino": 2,
 476           "st_uid": 0,
 477           "st_mode": 16877,
 478           "st_atime": 1431520593.0
 479         }
 480
 481         Raises exception on absent file.
 482         """
 483         abs_path = os.path.join(self.mountpoint, fs_path)
 484
 485         pyscript = dedent("""
 486             import os
 487             import stat
 488             import json
 489             import sys
 490
 491             try:
 492                 s = os.stat("{path}")
 493             except OSError as e:
 494                 sys.exit(e.errno)
 495
 496             attrs = ["st_mode", "st_ino", "st_dev", "st_nlink", "st_uid", "st_gid", "st_size", "st_atime", "st_mtime", "st_ctime"]
 497             print json.dumps(
 498                 dict([(a, getattr(s, a)) for a in attrs]),
 499                 indent=2)
 500             """).format(path=abs_path)
 501         proc = self._run_python(pyscript)
 502         if wait:
 503             proc.wait()
 504             return json.loads(proc.stdout.getvalue().strip())
 505         else:
 506             return proc
 507
 508     def touch(self, fs_path):
 509         """
 510         Create a dentry if it doesn't already exist.  This python
 511         implementation exists because the usual command line tool doesn't
 512         pass through error codes like EIO.
 513
 514         :param fs_path:
 515         :return:
 516         """
 517         abs_path = os.path.join(self.mountpoint, fs_path)
 518         pyscript = dedent("""
 519             import sys
 520             import errno
 521
 522             try:
 523                 f = open("{path}", "w")
 524                 f.close()
 525             except IOError as e:
 526                 sys.exit(errno.EIO)
 527             """).format(path=abs_path)
 528         proc = self._run_python(pyscript)
 529         proc.wait()
 530
 531     def path_to_ino(self, fs_path, follow_symlinks=True):
 532         abs_path = os.path.join(self.mountpoint, fs_path)
 533
 534         if follow_symlinks:
 535             pyscript = dedent("""
 536                 import os
 537                 import stat
 538
 539                 print os.stat("{path}").st_ino
 540                 """).format(path=abs_path)
 541         else:
 542             pyscript = dedent("""
 543                 import os
 544                 import stat
 545
 546                 print os.lstat("{path}").st_ino
 547                 """).format(path=abs_path)
 548
 549         proc = self._run_python(pyscript)
 550         proc.wait()
 551         return int(proc.stdout.getvalue().strip())
 552
 553     def path_to_nlink(self, fs_path):
 554         abs_path = os.path.join(self.mountpoint, fs_path)
 555
 556         pyscript = dedent("""
 557             import os
 558             import stat
 559
 560             print os.stat("{path}").st_nlink
 561             """).format(path=abs_path)
 562
 563         proc = self._run_python(pyscript)
 564         proc.wait()
 565         return int(proc.stdout.getvalue().strip())
 566
 567     def ls(self, path=None):
 568         """
 569         Wrap ls: return a list of strings
 570         """
 571         cmd = ["ls"]
 572         if path:
 573             cmd.append(path)
 574
 575         ls_text = self.run_shell(cmd).stdout.getvalue().strip()
 576
 577         if ls_text:
 578             return ls_text.split("\n")
 579         else:
 580             # Special case because otherwise split on empty string
 581             # gives you [''] instead of []
 582             return []
 583
 584     def setfattr(self, path, key, val):
 585         """
 586         Wrap setfattr.
 587
 588         :param path: relative to mount point
 589         :param key: xattr name
 590         :param val: xattr value
 591         :return: None
 592         """
 593         self.run_shell(["setfattr", "-n", key, "-v", val, path])
 594
 595     def getfattr(self, path, attr):
 596         """
 597         Wrap getfattr: return the values of a named xattr on one file, or
 598         None if the attribute is not found.
 599
 600         :return: a string
 601         """
 602         p = self.run_shell(["getfattr", "--only-values", "-n", attr, path], wait=False)
 603         try:
 604             p.wait()
 605         except CommandFailedError as e:
 606             if e.exitstatus == 1 and "No such attribute" in p.stderr.getvalue():
 607                 return None
 608             else:
 609                 raise
 610
 611         return p.stdout.getvalue()
 612
 613     def df(self):
 614         """
 615         Wrap df: return a dict of usage fields in bytes
 616         """
 617
 618         p = self.run_shell(["df", "-B1", "."])
 619         lines = p.stdout.getvalue().strip().split("\n")
 620         fs, total, used, avail = lines[1].split()[:4]
 621         log.warn(lines)
 622
 623         return {
 624             "total": int(total),
 625             "used": int(used),
 626             "available": int(avail)
 627         }