ceph/qa/tasks/cephfs/mount.py

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