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, Raw
  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_payload(self, payload, **kwargs):
 201         return self.run_shell(["bash", "-c", Raw(f"'{payload}'")], **kwargs)
 202
 203     def run_shell(self, args, wait=True, stdin=None, check_status=True,
 204                   omit_sudo=True, timeout=10800):
 205         if isinstance(args, str):
 206             args = args.split()
 207
 208         args = ["cd", self.mountpoint, run.Raw('&&'), "sudo"] + args
 209         return self.client_remote.run(args=args, stdout=StringIO(),
 210                                       stderr=StringIO(), wait=wait,
 211                                       stdin=stdin, check_status=check_status,
 212                                       omit_sudo=omit_sudo,
 213                                       timeout=timeout)
 214
 215     def open_no_data(self, basename):
 216         """
 217         A pure metadata operation
 218         """
 219         assert(self.is_mounted())
 220
 221         path = os.path.join(self.mountpoint, basename)
 222
 223         p = self._run_python(dedent(
 224             """
 225             f = open("{path}", 'w')
 226             """.format(path=path)
 227         ))
 228         p.wait()
 229
 230     def open_background(self, basename="background_file", write=True):
 231         """
 232         Open a file for writing, then block such that the client
 233         will hold a capability.
 234
 235         Don't return until the remote process has got as far as opening
 236         the file, then return the RemoteProcess instance.
 237         """
 238         assert(self.is_mounted())
 239
 240         path = os.path.join(self.mountpoint, basename)
 241
 242         if write:
 243             pyscript = dedent("""
 244                 import time
 245
 246                 with open("{path}", 'w') as f:
 247                     f.write('content')
 248                     f.flush()
 249                     f.write('content2')
 250                     while True:
 251                         time.sleep(1)
 252                 """).format(path=path)
 253         else:
 254             pyscript = dedent("""
 255                 import time
 256
 257                 with open("{path}", 'r') as f:
 258                     while True:
 259                         time.sleep(1)
 260                 """).format(path=path)
 261
 262         rproc = self._run_python(pyscript)
 263         self.background_procs.append(rproc)
 264
 265         # This wait would not be sufficient if the file had already
 266         # existed, but it's simple and in practice users of open_background
 267         # are not using it on existing files.
 268         self.wait_for_visible(basename)
 269
 270         return rproc
 271
 272     def wait_for_dir_empty(self, dirname, timeout=30):
 273         i = 0
 274         dirpath = os.path.join(self.mountpoint, dirname)
 275         while i < timeout:
 276             nr_entries = int(self.getfattr(dirpath, "ceph.dir.entries"))
 277             if nr_entries == 0:
 278                 log.debug("Directory {0} seen empty from {1} after {2}s ".format(
 279                     dirname, self.client_id, i))
 280                 return
 281             else:
 282                 time.sleep(1)
 283                 i += 1
 284
 285         raise RuntimeError("Timed out after {0}s waiting for {1} to become empty from {2}".format(
 286             i, dirname, self.client_id))
 287
 288     def wait_for_visible(self, basename="background_file", timeout=30):
 289         i = 0
 290         while i < timeout:
 291             r = self.client_remote.run(args=[
 292                 'sudo', 'ls', os.path.join(self.mountpoint, basename)
 293             ], check_status=False)
 294             if r.exitstatus == 0:
 295                 log.debug("File {0} became visible from {1} after {2}s".format(
 296                     basename, self.client_id, i))
 297                 return
 298             else:
 299                 time.sleep(1)
 300                 i += 1
 301
 302         raise RuntimeError("Timed out after {0}s waiting for {1} to become visible from {2}".format(
 303             i, basename, self.client_id))
 304
 305     def lock_background(self, basename="background_file", do_flock=True):
 306         """
 307         Open and lock a files for writing, hold the lock in a background process
 308         """
 309         assert(self.is_mounted())
 310
 311         path = os.path.join(self.mountpoint, basename)
 312
 313         script_builder = """
 314             import time
 315             import fcntl
 316             import struct"""
 317         if do_flock:
 318             script_builder += """
 319             f1 = open("{path}-1", 'w')
 320             fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)"""
 321         script_builder += """
 322             f2 = open("{path}-2", 'w')
 323             lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 324             fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 325             while True:
 326                 time.sleep(1)
 327             """
 328
 329         pyscript = dedent(script_builder).format(path=path)
 330
 331         log.info("lock_background file {0}".format(basename))
 332         rproc = self._run_python(pyscript)
 333         self.background_procs.append(rproc)
 334         return rproc
 335
 336     def lock_and_release(self, basename="background_file"):
 337         assert(self.is_mounted())
 338
 339         path = os.path.join(self.mountpoint, basename)
 340
 341         script = """
 342             import time
 343             import fcntl
 344             import struct
 345             f1 = open("{path}-1", 'w')
 346             fcntl.flock(f1, fcntl.LOCK_EX)
 347             f2 = open("{path}-2", 'w')
 348             lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 349             fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 350             """
 351         pyscript = dedent(script).format(path=path)
 352
 353         log.info("lock_and_release file {0}".format(basename))
 354         return self._run_python(pyscript)
 355
 356     def check_filelock(self, basename="background_file", do_flock=True):
 357         assert(self.is_mounted())
 358
 359         path = os.path.join(self.mountpoint, basename)
 360
 361         script_builder = """
 362             import fcntl
 363             import errno
 364             import struct"""
 365         if do_flock:
 366             script_builder += """
 367             f1 = open("{path}-1", 'r')
 368             try:
 369                 fcntl.flock(f1, fcntl.LOCK_EX | fcntl.LOCK_NB)
 370             except IOError as e:
 371                 if e.errno == errno.EAGAIN:
 372                     pass
 373             else:
 374                 raise RuntimeError("flock on file {path}-1 not found")"""
 375         script_builder += """
 376             f2 = open("{path}-2", 'r')
 377             try:
 378                 lockdata = struct.pack('hhllhh', fcntl.F_WRLCK, 0, 0, 0, 0, 0)
 379                 fcntl.fcntl(f2, fcntl.F_SETLK, lockdata)
 380             except IOError as e:
 381                 if e.errno == errno.EAGAIN:
 382                     pass
 383             else:
 384                 raise RuntimeError("posix lock on file {path}-2 not found")
 385             """
 386         pyscript = dedent(script_builder).format(path=path)
 387
 388         log.info("check lock on file {0}".format(basename))
 389         self.client_remote.run(args=[
 390             'sudo', 'python3', '-c', pyscript
 391         ])
 392
 393     def write_background(self, basename="background_file", loop=False):
 394         """
 395         Open a file for writing, complete as soon as you can
 396         :param basename:
 397         :return:
 398         """
 399         assert(self.is_mounted())
 400
 401         path = os.path.join(self.mountpoint, basename)
 402
 403         pyscript = dedent("""
 404             import os
 405             import time
 406
 407             fd = os.open("{path}", os.O_RDWR | os.O_CREAT, 0o644)
 408             try:
 409                 while True:
 410                     os.write(fd, b'content')
 411                     time.sleep(1)
 412                     if not {loop}:
 413                         break
 414             except IOError as e:
 415                 pass
 416             os.close(fd)
 417             """).format(path=path, loop=str(loop))
 418
 419         rproc = self._run_python(pyscript)
 420         self.background_procs.append(rproc)
 421         return rproc
 422
 423     def write_n_mb(self, filename, n_mb, seek=0, wait=True):
 424         """
 425         Write the requested number of megabytes to a file
 426         """
 427         assert(self.is_mounted())
 428
 429         return self.run_shell(["dd", "if=/dev/urandom", "of={0}".format(filename),
 430                                "bs=1M", "conv=fdatasync",
 431                                "count={0}".format(int(n_mb)),
 432                                "seek={0}".format(int(seek))
 433                                ], wait=wait)
 434
 435     def write_test_pattern(self, filename, size):
 436         log.info("Writing {0} bytes to {1}".format(size, filename))
 437         return self.run_python(dedent("""
 438             import zlib
 439             path = "{path}"
 440             with open(path, 'w') as f:
 441                 for i in range(0, {size}):
 442                     val = zlib.crc32(str(i).encode('utf-8')) & 7
 443                     f.write(chr(val))
 444         """.format(
 445             path=os.path.join(self.mountpoint, filename),
 446             size=size
 447         )))
 448
 449     def validate_test_pattern(self, filename, size):
 450         log.info("Validating {0} bytes from {1}".format(size, filename))
 451         return self.run_python(dedent("""
 452             import zlib
 453             path = "{path}"
 454             with open(path, 'r') as f:
 455                 bytes = f.read()
 456             if len(bytes) != {size}:
 457                 raise RuntimeError("Bad length {{0}} vs. expected {{1}}".format(
 458                     len(bytes), {size}
 459                 ))
 460             for i, b in enumerate(bytes):
 461                 val = zlib.crc32(str(i).encode('utf-8')) & 7
 462                 if b != chr(val):
 463                     raise RuntimeError("Bad data at offset {{0}}".format(i))
 464         """.format(
 465             path=os.path.join(self.mountpoint, filename),
 466             size=size
 467         )))
 468
 469     def open_n_background(self, fs_path, count):
 470         """
 471         Open N files for writing, hold them open in a background process
 472
 473         :param fs_path: Path relative to CephFS root, e.g. "foo/bar"
 474         :return: a RemoteProcess
 475         """
 476         assert(self.is_mounted())
 477
 478         abs_path = os.path.join(self.mountpoint, fs_path)
 479
 480         pyscript = dedent("""
 481             import sys
 482             import time
 483             import os
 484
 485             n = {count}
 486             abs_path = "{abs_path}"
 487
 488             if not os.path.exists(abs_path):
 489                 os.makedirs(abs_path)
 490
 491             handles = []
 492             for i in range(0, n):
 493                 fname = "file_"+str(i)
 494                 path = os.path.join(abs_path, fname)
 495                 handles.append(open(path, 'w'))
 496
 497             while True:
 498                 time.sleep(1)
 499             """).format(abs_path=abs_path, count=count)
 500
 501         rproc = self._run_python(pyscript)
 502         self.background_procs.append(rproc)
 503         return rproc
 504
 505     def create_n_files(self, fs_path, count, sync=False):
 506         assert(self.is_mounted())
 507
 508         abs_path = os.path.join(self.mountpoint, fs_path)
 509
 510         pyscript = dedent("""
 511             import sys
 512             import time
 513             import os
 514
 515             n = {count}
 516             abs_path = "{abs_path}"
 517
 518             if not os.path.exists(os.path.dirname(abs_path)):
 519                 os.makedirs(os.path.dirname(abs_path))
 520
 521             for i in range(0, n):
 522                 fname = "{{0}}_{{1}}".format(abs_path, i)
 523                 with open(fname, 'w') as f:
 524                     f.write('content')
 525                     if {sync}:
 526                         f.flush()
 527                         os.fsync(f.fileno())
 528             """).format(abs_path=abs_path, count=count, sync=str(sync))
 529
 530         self.run_python(pyscript)
 531
 532     def teardown(self):
 533         for p in self.background_procs:
 534             log.info("Terminating background process")
 535             self._kill_background(p)
 536
 537         self.background_procs = []
 538
 539     def _kill_background(self, p):
 540         if p.stdin:
 541             p.stdin.close()
 542             try:
 543                 p.wait()
 544             except (CommandFailedError, ConnectionLostError):
 545                 pass
 546
 547     def kill_background(self, p):
 548         """
 549         For a process that was returned by one of the _background member functions,
 550         kill it hard.
 551         """
 552         self._kill_background(p)
 553         self.background_procs.remove(p)
 554
 555     def send_signal(self, signal):
 556         signal = signal.lower()
 557         if signal.lower() not in ['sigstop', 'sigcont', 'sigterm', 'sigkill']:
 558             raise NotImplementedError
 559
 560         self.client_remote.run(args=['sudo', 'kill', '-{0}'.format(signal),
 561                                 self.client_pid], omit_sudo=False)
 562
 563     def get_global_id(self):
 564         raise NotImplementedError()
 565
 566     def get_global_inst(self):
 567         raise NotImplementedError()
 568
 569     def get_global_addr(self):
 570         raise NotImplementedError()
 571
 572     def get_osd_epoch(self):
 573         raise NotImplementedError()
 574
 575     def lstat(self, fs_path, follow_symlinks=False, wait=True):
 576         return self.stat(fs_path, follow_symlinks=False, wait=True)
 577
 578     def stat(self, fs_path, follow_symlinks=True, wait=True):
 579         """
 580         stat a file, and return the result as a dictionary like this:
 581         {
 582           "st_ctime": 1414161137.0,
 583           "st_mtime": 1414161137.0,
 584           "st_nlink": 33,
 585           "st_gid": 0,
 586           "st_dev": 16777218,
 587           "st_size": 1190,
 588           "st_ino": 2,
 589           "st_uid": 0,
 590           "st_mode": 16877,
 591           "st_atime": 1431520593.0
 592         }
 593
 594         Raises exception on absent file.
 595         """
 596         abs_path = os.path.join(self.mountpoint, fs_path)
 597         if follow_symlinks:
 598             stat_call = "os.stat('" + abs_path + "')"
 599         else:
 600             stat_call = "os.lstat('" + abs_path + "')"
 601
 602         pyscript = dedent("""
 603             import os
 604             import stat
 605             import json
 606             import sys
 607
 608             try:
 609                 s = {stat_call}
 610             except OSError as e:
 611                 sys.exit(e.errno)
 612
 613             attrs = ["st_mode", "st_ino", "st_dev", "st_nlink", "st_uid", "st_gid", "st_size", "st_atime", "st_mtime", "st_ctime"]
 614             print(json.dumps(
 615                 dict([(a, getattr(s, a)) for a in attrs]),
 616                 indent=2))
 617             """).format(stat_call=stat_call)
 618         proc = self._run_python(pyscript)
 619         if wait:
 620             proc.wait()
 621             return json.loads(proc.stdout.getvalue().strip())
 622         else:
 623             return proc
 624
 625     def touch(self, fs_path):
 626         """
 627         Create a dentry if it doesn't already exist.  This python
 628         implementation exists because the usual command line tool doesn't
 629         pass through error codes like EIO.
 630
 631         :param fs_path:
 632         :return:
 633         """
 634         abs_path = os.path.join(self.mountpoint, fs_path)
 635         pyscript = dedent("""
 636             import sys
 637             import errno
 638
 639             try:
 640                 f = open("{path}", "w")
 641                 f.close()
 642             except IOError as e:
 643                 sys.exit(errno.EIO)
 644             """).format(path=abs_path)
 645         proc = self._run_python(pyscript)
 646         proc.wait()
 647
 648     def path_to_ino(self, fs_path, follow_symlinks=True):
 649         abs_path = os.path.join(self.mountpoint, fs_path)
 650
 651         if follow_symlinks:
 652             pyscript = dedent("""
 653                 import os
 654                 import stat
 655
 656                 print(os.stat("{path}").st_ino)
 657                 """).format(path=abs_path)
 658         else:
 659             pyscript = dedent("""
 660                 import os
 661                 import stat
 662
 663                 print(os.lstat("{path}").st_ino)
 664                 """).format(path=abs_path)
 665
 666         proc = self._run_python(pyscript)
 667         proc.wait()
 668         return int(proc.stdout.getvalue().strip())
 669
 670     def path_to_nlink(self, fs_path):
 671         abs_path = os.path.join(self.mountpoint, fs_path)
 672
 673         pyscript = dedent("""
 674             import os
 675             import stat
 676
 677             print(os.stat("{path}").st_nlink)
 678             """).format(path=abs_path)
 679
 680         proc = self._run_python(pyscript)
 681         proc.wait()
 682         return int(proc.stdout.getvalue().strip())
 683
 684     def ls(self, path=None):
 685         """
 686         Wrap ls: return a list of strings
 687         """
 688         cmd = ["ls"]
 689         if path:
 690             cmd.append(path)
 691
 692         ls_text = self.run_shell(cmd).stdout.getvalue().strip()
 693
 694         if ls_text:
 695             return ls_text.split("\n")
 696         else:
 697             # Special case because otherwise split on empty string
 698             # gives you [''] instead of []
 699             return []
 700
 701     def setfattr(self, path, key, val):
 702         """
 703         Wrap setfattr.
 704
 705         :param path: relative to mount point
 706         :param key: xattr name
 707         :param val: xattr value
 708         :return: None
 709         """
 710         self.run_shell(["setfattr", "-n", key, "-v", val, path])
 711
 712     def getfattr(self, path, attr):
 713         """
 714         Wrap getfattr: return the values of a named xattr on one file, or
 715         None if the attribute is not found.
 716
 717         :return: a string
 718         """
 719         p = self.run_shell(["getfattr", "--only-values", "-n", attr, path], wait=False)
 720         try:
 721             p.wait()
 722         except CommandFailedError as e:
 723             if e.exitstatus == 1 and "No such attribute" in p.stderr.getvalue():
 724                 return None
 725             else:
 726                 raise
 727
 728         return str(p.stdout.getvalue())
 729
 730     def df(self):
 731         """
 732         Wrap df: return a dict of usage fields in bytes
 733         """
 734
 735         p = self.run_shell(["df", "-B1", "."])
 736         lines = p.stdout.getvalue().strip().split("\n")
 737         fs, total, used, avail = lines[1].split()[:4]
 738         log.warning(lines)
 739
 740         return {
 741             "total": int(total),
 742             "used": int(used),
 743             "available": int(avail)
 744         }