import json import logging import time import os from textwrap import dedent from tasks.cephfs.cephfs_test_case import CephFSTestCase from tasks.cephfs.fuse_mount import FuseMount from teuthology.exceptions import CommandFailedError log = logging.getLogger(__name__) class TestVolumeClient(CephFSTestCase): # One for looking at the global filesystem, one for being # the VolumeClient, two for mounting the created shares CLIENTS_REQUIRED = 4 py_version = 'python' def setUp(self): CephFSTestCase.setUp(self) self.py_version = self.ctx.config.get('overrides', {}).get('python', 'python') log.info("using python version: %s".format(self.py_version)) def _volume_client_python(self, client, script, vol_prefix=None, ns_prefix=None): # Can't dedent this *and* the script we pass in, because they might have different # levels of indentation to begin with, so leave this string zero-indented if vol_prefix: vol_prefix = "\"" + vol_prefix + "\"" if ns_prefix: ns_prefix = "\"" + ns_prefix + "\"" return client.run_python(""" from __future__ import print_function from ceph_volume_client import CephFSVolumeClient, VolumePath import logging log = logging.getLogger("ceph_volume_client") log.addHandler(logging.StreamHandler()) log.setLevel(logging.DEBUG) vc = CephFSVolumeClient("manila", "{conf_path}", "ceph", {vol_prefix}, {ns_prefix}) vc.connect() {payload} vc.disconnect() """.format(payload=script, conf_path=client.config_path, vol_prefix=vol_prefix, ns_prefix=ns_prefix), self.py_version) def _sudo_write_file(self, remote, path, data): """ Write data to a remote file as super user :param remote: Remote site. :param path: Path on the remote being written to. :param data: Data to be written. Both perms and owner are passed directly to chmod. """ remote.run( args=[ 'sudo', 'python', '-c', 'import shutil, sys; shutil.copyfileobj(sys.stdin, file(sys.argv[1], "wb"))', path, ], stdin=data, ) def _configure_vc_auth(self, mount, id_name): """ Set up auth credentials for the VolumeClient user """ out = self.fs.mon_manager.raw_cluster_cmd( "auth", "get-or-create", "client.{name}".format(name=id_name), "mds", "allow *", "osd", "allow rw", "mon", "allow *" ) mount.client_id = id_name self._sudo_write_file(mount.client_remote, mount.get_keyring_path(), out) self.set_conf("client.{name}".format(name=id_name), "keyring", mount.get_keyring_path()) def _configure_guest_auth(self, volumeclient_mount, guest_mount, guest_entity, mount_path, namespace_prefix=None, readonly=False, tenant_id=None): """ Set up auth credentials for the guest client to mount a volume. :param volumeclient_mount: mount used as the handle for driving volumeclient. :param guest_mount: mount used by the guest client. :param guest_entity: auth ID used by the guest client. :param mount_path: path of the volume. :param namespace_prefix: name prefix of the RADOS namespace, which is used for the volume's layout. :param readonly: defaults to False. If set to 'True' only read-only mount access is granted to the guest. :param tenant_id: (OpenStack) tenant ID of the guest client. """ head, volume_id = os.path.split(mount_path) head, group_id = os.path.split(head) head, volume_prefix = os.path.split(head) volume_prefix = "/" + volume_prefix # Authorize the guest client's auth ID to mount the volume. key = self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") auth_result = vc.authorize(vp, "{guest_entity}", readonly={readonly}, tenant_id="{tenant_id}") print(auth_result['auth_key']) """.format( group_id=group_id, volume_id=volume_id, guest_entity=guest_entity, readonly=readonly, tenant_id=tenant_id)), volume_prefix, namespace_prefix ) # CephFSVolumeClient's authorize() does not return the secret # key to a caller who isn't multi-tenant aware. Explicitly # query the key for such a client. if not tenant_id: key = self.fs.mon_manager.raw_cluster_cmd( "auth", "get-key", "client.{name}".format(name=guest_entity), ) # The guest auth ID should exist. existing_ids = [a['entity'] for a in self.auth_list()] self.assertIn("client.{0}".format(guest_entity), existing_ids) # Create keyring file for the guest client. keyring_txt = dedent(""" [client.{guest_entity}] key = {key} """.format( guest_entity=guest_entity, key=key )) guest_mount.client_id = guest_entity self._sudo_write_file(guest_mount.client_remote, guest_mount.get_keyring_path(), keyring_txt) # Add a guest client section to the ceph config file. self.set_conf("client.{0}".format(guest_entity), "client quota", "True") self.set_conf("client.{0}".format(guest_entity), "debug client", "20") self.set_conf("client.{0}".format(guest_entity), "debug objecter", "20") self.set_conf("client.{0}".format(guest_entity), "keyring", guest_mount.get_keyring_path()) def test_default_prefix(self): group_id = "grpid" volume_id = "volid" DEFAULT_VOL_PREFIX = "volumes" DEFAULT_NS_PREFIX = "fsvolumens_" self.mount_b.umount_wait() self._configure_vc_auth(self.mount_b, "manila") #create a volume with default prefix self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.create_volume(vp, 10, data_isolated=True) """.format( group_id=group_id, volume_id=volume_id, ))) # The dir should be created self.mount_a.stat(os.path.join(DEFAULT_VOL_PREFIX, group_id, volume_id)) #namespace should be set ns_in_attr = self.mount_a.getfattr(os.path.join(DEFAULT_VOL_PREFIX, group_id, volume_id), "ceph.dir.layout.pool_namespace") namespace = "{0}{1}".format(DEFAULT_NS_PREFIX, volume_id) self.assertEqual(namespace, ns_in_attr) def test_lifecycle(self): """ General smoke test for create, extend, destroy """ # I'm going to use mount_c later as a guest for mounting the created # shares self.mounts[2].umount_wait() # I'm going to leave mount_b unmounted and just use it as a handle for # driving volumeclient. It's a little hacky but we don't have a more # general concept for librados/libcephfs clients as opposed to full # blown mounting clients. self.mount_b.umount_wait() self._configure_vc_auth(self.mount_b, "manila") guest_entity = "guest" group_id = "grpid" volume_id = "volid" volume_prefix = "/myprefix" namespace_prefix = "mynsprefix_" # Create a 100MB volume volume_size = 100 mount_path = self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") create_result = vc.create_volume(vp, 1024*1024*{volume_size}) print(create_result['mount_path']) """.format( group_id=group_id, volume_id=volume_id, volume_size=volume_size )), volume_prefix, namespace_prefix) # The dir should be created self.mount_a.stat(os.path.join("myprefix", group_id, volume_id)) # Authorize and configure credentials for the guest to mount the # the volume. self._configure_guest_auth(self.mount_b, self.mounts[2], guest_entity, mount_path, namespace_prefix) self.mounts[2].mount(mount_path=mount_path) # The kernel client doesn't have the quota-based df behaviour, # or quotas at all, so only exercise the client behaviour when # running fuse. if isinstance(self.mounts[2], FuseMount): # df should see volume size, same as the quota set on volume's dir self.assertEqual(self.mounts[2].df()['total'], volume_size * 1024 * 1024) self.assertEqual( self.mount_a.getfattr( os.path.join(volume_prefix.strip("/"), group_id, volume_id), "ceph.quota.max_bytes"), "%s" % (volume_size * 1024 * 1024)) # df granularity is 4MB block so have to write at least that much data_bin_mb = 4 self.mounts[2].write_n_mb("data.bin", data_bin_mb) # Write something outside volume to check this space usage is # not reported in the volume's DF. other_bin_mb = 8 self.mount_a.write_n_mb("other.bin", other_bin_mb) # global: df should see all the writes (data + other). This is a > # rather than a == because the global spaced used includes all pools def check_df(): used = self.mount_a.df()['used'] return used >= (other_bin_mb * 1024 * 1024) self.wait_until_true(check_df, timeout=30) # Hack: do a metadata IO to kick rstats self.mounts[2].run_shell(["touch", "foo"]) # volume: df should see the data_bin_mb consumed from quota, same # as the rbytes for the volume's dir self.wait_until_equal( lambda: self.mounts[2].df()['used'], data_bin_mb * 1024 * 1024, timeout=60) self.wait_until_equal( lambda: self.mount_a.getfattr( os.path.join(volume_prefix.strip("/"), group_id, volume_id), "ceph.dir.rbytes"), "%s" % (data_bin_mb * 1024 * 1024), timeout=60) # sync so that file data are persist to rados self.mounts[2].run_shell(["sync"]) # Our data should stay in particular rados namespace pool_name = self.mount_a.getfattr(os.path.join("myprefix", group_id, volume_id), "ceph.dir.layout.pool") namespace = "{0}{1}".format(namespace_prefix, volume_id) ns_in_attr = self.mount_a.getfattr(os.path.join("myprefix", group_id, volume_id), "ceph.dir.layout.pool_namespace") self.assertEqual(namespace, ns_in_attr) objects_in_ns = set(self.fs.rados(["ls"], pool=pool_name, namespace=namespace).split("\n")) self.assertNotEqual(objects_in_ns, set()) # De-authorize the guest self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.deauthorize(vp, "{guest_entity}") vc.evict("{guest_entity}") """.format( group_id=group_id, volume_id=volume_id, guest_entity=guest_entity )), volume_prefix, namespace_prefix) # Once deauthorized, the client should be unable to do any more metadata ops # The way that the client currently behaves here is to block (it acts like # it has lost network, because there is nothing to tell it that is messages # are being dropped because it's identity is gone) background = self.mounts[2].write_n_mb("rogue.bin", 1, wait=False) try: background.wait() except CommandFailedError: # command failed with EBLACKLISTED? if "transport endpoint shutdown" in background.stderr.getvalue(): pass else: raise # After deauthorisation, the client ID should be gone (this was the only # volume it was authorised for) self.assertNotIn("client.{0}".format(guest_entity), [e['entity'] for e in self.auth_list()]) # Clean up the dead mount (ceph-fuse's behaviour here is a bit undefined) self.mounts[2].umount_wait() self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.delete_volume(vp) vc.purge_volume(vp) """.format( group_id=group_id, volume_id=volume_id, )), volume_prefix, namespace_prefix) def test_idempotency(self): """ That the volumeclient interface works when calling everything twice """ self.mount_b.umount_wait() self._configure_vc_auth(self.mount_b, "manila") guest_entity = "guest" group_id = "grpid" volume_id = "volid" self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.create_volume(vp, 10) vc.create_volume(vp, 10) vc.authorize(vp, "{guest_entity}") vc.authorize(vp, "{guest_entity}") vc.deauthorize(vp, "{guest_entity}") vc.deauthorize(vp, "{guest_entity}") vc.delete_volume(vp) vc.delete_volume(vp) vc.purge_volume(vp) vc.purge_volume(vp) vc.create_volume(vp, 10, data_isolated=True) vc.create_volume(vp, 10, data_isolated=True) vc.authorize(vp, "{guest_entity}") vc.authorize(vp, "{guest_entity}") vc.deauthorize(vp, "{guest_entity}") vc.deauthorize(vp, "{guest_entity}") vc.evict("{guest_entity}") vc.evict("{guest_entity}") vc.delete_volume(vp, data_isolated=True) vc.delete_volume(vp, data_isolated=True) vc.purge_volume(vp, data_isolated=True) vc.purge_volume(vp, data_isolated=True) vc.create_volume(vp, 10, namespace_isolated=False) vc.create_volume(vp, 10, namespace_isolated=False) vc.authorize(vp, "{guest_entity}") vc.authorize(vp, "{guest_entity}") vc.deauthorize(vp, "{guest_entity}") vc.deauthorize(vp, "{guest_entity}") vc.evict("{guest_entity}") vc.evict("{guest_entity}") vc.delete_volume(vp) vc.delete_volume(vp) vc.purge_volume(vp) vc.purge_volume(vp) """.format( group_id=group_id, volume_id=volume_id, guest_entity=guest_entity ))) def test_data_isolated(self): """ That data isolated shares get their own pool :return: """ # Because the teuthology config template sets mon_max_pg_per_osd to # 10000 (i.e. it just tries to ignore health warnings), reset it to something # sane before using volume_client, to avoid creating pools with absurdly large # numbers of PGs. self.set_conf("global", "mon max pg per osd", "300") for mon_daemon_state in self.ctx.daemons.iter_daemons_of_role('mon'): mon_daemon_state.restart() self.mount_b.umount_wait() self._configure_vc_auth(self.mount_b, "manila") # Calculate how many PGs we'll expect the new volume pool to have osd_map = json.loads(self.fs.mon_manager.raw_cluster_cmd('osd', 'dump', '--format=json-pretty')) max_per_osd = int(self.fs.get_config('mon_max_pg_per_osd')) osd_count = len(osd_map['osds']) max_overall = osd_count * max_per_osd existing_pg_count = 0 for p in osd_map['pools']: existing_pg_count += p['pg_num'] expected_pg_num = (max_overall - existing_pg_count) / 10 log.info("max_per_osd {0}".format(max_per_osd)) log.info("osd_count {0}".format(osd_count)) log.info("max_overall {0}".format(max_overall)) log.info("existing_pg_count {0}".format(existing_pg_count)) log.info("expected_pg_num {0}".format(expected_pg_num)) pools_a = json.loads(self.fs.mon_manager.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['pools'] group_id = "grpid" volume_id = "volid" self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.create_volume(vp, 10, data_isolated=True) """.format( group_id=group_id, volume_id=volume_id, ))) pools_b = json.loads(self.fs.mon_manager.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['pools'] # Should have created one new pool new_pools = set(p['pool_name'] for p in pools_b) - set([p['pool_name'] for p in pools_a]) self.assertEqual(len(new_pools), 1) # It should have followed the heuristic for PG count # (this is an overly strict test condition, so we may want to remove # it at some point as/when the logic gets fancier) created_pg_num = self.fs.mon_manager.get_pool_property(list(new_pools)[0], "pg_num") self.assertEqual(expected_pg_num, created_pg_num) def test_15303(self): """ Reproducer for #15303 "Client holds incorrect complete flag on dir after losing caps" (http://tracker.ceph.com/issues/15303) """ for m in self.mounts: m.umount_wait() # Create a dir on mount A self.mount_a.mount() self.mount_a.run_shell(["mkdir", "parent1"]) self.mount_a.run_shell(["mkdir", "parent2"]) self.mount_a.run_shell(["mkdir", "parent1/mydir"]) # Put some files in it from mount B self.mount_b.mount() self.mount_b.run_shell(["touch", "parent1/mydir/afile"]) self.mount_b.umount_wait() # List the dir's contents on mount A self.assertListEqual(self.mount_a.ls("parent1/mydir"), ["afile"]) def test_evict_client(self): """ That a volume client can be evicted based on its auth ID and the volume path it has mounted. """ if not isinstance(self.mount_a, FuseMount): self.skipTest("Requires FUSE client to inject client metadata") # mounts[1] would be used as handle for driving VolumeClient. mounts[2] # and mounts[3] would be used as guests to mount the volumes/shares. for i in range(1, 4): self.mounts[i].umount_wait() volumeclient_mount = self.mounts[1] self._configure_vc_auth(volumeclient_mount, "manila") guest_mounts = (self.mounts[2], self.mounts[3]) guest_entity = "guest" group_id = "grpid" mount_paths = [] volume_ids = [] # Create two volumes. Authorize 'guest' auth ID to mount the two # volumes. Mount the two volumes. Write data to the volumes. for i in range(2): # Create volume. volume_ids.append("volid_{0}".format(str(i))) mount_paths.append( self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") create_result = vc.create_volume(vp, 10 * 1024 * 1024) print(create_result['mount_path']) """.format( group_id=group_id, volume_id=volume_ids[i] )))) # Authorize 'guest' auth ID to mount the volume. self._configure_guest_auth(volumeclient_mount, guest_mounts[i], guest_entity, mount_paths[i]) # Mount the volume. guest_mounts[i].mountpoint_dir_name = 'mnt.{id}.{suffix}'.format( id=guest_entity, suffix=str(i)) guest_mounts[i].mount(mount_path=mount_paths[i]) guest_mounts[i].write_n_mb("data.bin", 1) # Evict client, guest_mounts[0], using auth ID 'guest' and has mounted # one volume. self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.deauthorize(vp, "{guest_entity}") vc.evict("{guest_entity}", volume_path=vp) """.format( group_id=group_id, volume_id=volume_ids[0], guest_entity=guest_entity ))) # Evicted guest client, guest_mounts[0], should not be able to do # anymore metadata ops. It should start failing all operations # when it sees that its own address is in the blacklist. try: guest_mounts[0].write_n_mb("rogue.bin", 1) except CommandFailedError: pass else: raise RuntimeError("post-eviction write should have failed!") # The blacklisted guest client should now be unmountable guest_mounts[0].umount_wait() # Guest client, guest_mounts[1], using the same auth ID 'guest', but # has mounted the other volume, should be able to use its volume # unaffected. guest_mounts[1].write_n_mb("data.bin.1", 1) # Cleanup. for i in range(2): self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.deauthorize(vp, "{guest_entity}") vc.delete_volume(vp) vc.purge_volume(vp) """.format( group_id=group_id, volume_id=volume_ids[i], guest_entity=guest_entity ))) def test_purge(self): """ Reproducer for #15266, exception trying to purge volumes that contain non-ascii filenames. Additionally test any other purge corner cases here. """ # I'm going to leave mount_b unmounted and just use it as a handle for # driving volumeclient. It's a little hacky but we don't have a more # general concept for librados/libcephfs clients as opposed to full # blown mounting clients. self.mount_b.umount_wait() self._configure_vc_auth(self.mount_b, "manila") group_id = "grpid" # Use a unicode volume ID (like Manila), to reproduce #15266 volume_id = u"volid" # Create mount_path = self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", u"{volume_id}") create_result = vc.create_volume(vp, 10) print(create_result['mount_path']) """.format( group_id=group_id, volume_id=volume_id ))) # Strip leading "/" mount_path = mount_path[1:] # A file with non-ascii characters self.mount_a.run_shell(["touch", os.path.join(mount_path, u"b\u00F6b")]) # A file with no permissions to do anything self.mount_a.run_shell(["touch", os.path.join(mount_path, "noperms")]) self.mount_a.run_shell(["chmod", "0000", os.path.join(mount_path, "noperms")]) self._volume_client_python(self.mount_b, dedent(""" vp = VolumePath("{group_id}", u"{volume_id}") vc.delete_volume(vp) vc.purge_volume(vp) """.format( group_id=group_id, volume_id=volume_id ))) # Check it's really gone self.assertEqual(self.mount_a.ls("volumes/_deleting"), []) self.assertEqual(self.mount_a.ls("volumes/"), ["_deleting", group_id]) def test_readonly_authorization(self): """ That guest clients can be restricted to read-only mounts of volumes. """ volumeclient_mount = self.mounts[1] guest_mount = self.mounts[2] volumeclient_mount.umount_wait() guest_mount.umount_wait() # Configure volumeclient_mount as the handle for driving volumeclient. self._configure_vc_auth(volumeclient_mount, "manila") guest_entity = "guest" group_id = "grpid" volume_id = "volid" # Create a volume. mount_path = self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") create_result = vc.create_volume(vp, 1024*1024*10) print(create_result['mount_path']) """.format( group_id=group_id, volume_id=volume_id, ))) # Authorize and configure credentials for the guest to mount the # the volume with read-write access. self._configure_guest_auth(volumeclient_mount, guest_mount, guest_entity, mount_path, readonly=False) # Mount the volume, and write to it. guest_mount.mount(mount_path=mount_path) guest_mount.write_n_mb("data.bin", 1) # Change the guest auth ID's authorization to read-only mount access. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.deauthorize(vp, "{guest_entity}") """.format( group_id=group_id, volume_id=volume_id, guest_entity=guest_entity ))) self._configure_guest_auth(volumeclient_mount, guest_mount, guest_entity, mount_path, readonly=True) # The effect of the change in access level to read-only is not # immediate. The guest sees the change only after a remount of # the volume. guest_mount.umount_wait() guest_mount.mount(mount_path=mount_path) # Read existing content of the volume. self.assertListEqual(guest_mount.ls(guest_mount.mountpoint), ["data.bin"]) # Cannot write into read-only volume. with self.assertRaises(CommandFailedError): guest_mount.write_n_mb("rogue.bin", 1) def test_get_authorized_ids(self): """ That for a volume, the authorized IDs and their access levels can be obtained using CephFSVolumeClient's get_authorized_ids(). """ volumeclient_mount = self.mounts[1] volumeclient_mount.umount_wait() # Configure volumeclient_mount as the handle for driving volumeclient. self._configure_vc_auth(volumeclient_mount, "manila") group_id = "grpid" volume_id = "volid" guest_entity_1 = "guest1" guest_entity_2 = "guest2" log.info("print(group ID: {0})".format(group_id)) # Create a volume. auths = self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.create_volume(vp, 1024*1024*10) auths = vc.get_authorized_ids(vp) print(auths) """.format( group_id=group_id, volume_id=volume_id, ))) # Check the list of authorized IDs for the volume. expected_result = None self.assertEqual(str(expected_result), auths) # Allow two auth IDs access to the volume. auths = self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.authorize(vp, "{guest_entity_1}", readonly=False) vc.authorize(vp, "{guest_entity_2}", readonly=True) auths = vc.get_authorized_ids(vp) print(auths) """.format( group_id=group_id, volume_id=volume_id, guest_entity_1=guest_entity_1, guest_entity_2=guest_entity_2, ))) # Check the list of authorized IDs and their access levels. if self.py_version == 'python3': expected_result = [('guest1', 'rw'), ('guest2', 'r')] else: expected_result = [(u'guest1', u'rw'), (u'guest2', u'r')] self.assertItemsEqual(str(expected_result), auths) # Disallow both the auth IDs' access to the volume. auths = self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.deauthorize(vp, "{guest_entity_1}") vc.deauthorize(vp, "{guest_entity_2}") auths = vc.get_authorized_ids(vp) print(auths) """.format( group_id=group_id, volume_id=volume_id, guest_entity_1=guest_entity_1, guest_entity_2=guest_entity_2, ))) # Check the list of authorized IDs for the volume. expected_result = None self.assertItemsEqual(str(expected_result), auths) def test_multitenant_volumes(self): """ That volume access can be restricted to a tenant. That metadata used to enforce tenant isolation of volumes is stored as a two-way mapping between auth IDs and volumes that they're authorized to access. """ volumeclient_mount = self.mounts[1] volumeclient_mount.umount_wait() # Configure volumeclient_mount as the handle for driving volumeclient. self._configure_vc_auth(volumeclient_mount, "manila") group_id = "groupid" volume_id = "volumeid" # Guest clients belonging to different tenants, but using the same # auth ID. auth_id = "guest" guestclient_1 = { "auth_id": auth_id, "tenant_id": "tenant1", } guestclient_2 = { "auth_id": auth_id, "tenant_id": "tenant2", } # Create a volume. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.create_volume(vp, 1024*1024*10) """.format( group_id=group_id, volume_id=volume_id, ))) # Check that volume metadata file is created on volume creation. vol_metadata_filename = "_{0}:{1}.meta".format(group_id, volume_id) self.assertIn(vol_metadata_filename, self.mounts[0].ls("volumes")) # Authorize 'guestclient_1', using auth ID 'guest' and belonging to # 'tenant1', with 'rw' access to the volume. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.authorize(vp, "{auth_id}", tenant_id="{tenant_id}") """.format( group_id=group_id, volume_id=volume_id, auth_id=guestclient_1["auth_id"], tenant_id=guestclient_1["tenant_id"] ))) # Check that auth metadata file for auth ID 'guest', is # created on authorizing 'guest' access to the volume. auth_metadata_filename = "${0}.meta".format(guestclient_1["auth_id"]) self.assertIn(auth_metadata_filename, self.mounts[0].ls("volumes")) # Verify that the auth metadata file stores the tenant ID that the # auth ID belongs to, the auth ID's authorized access levels # for different volumes, versioning details, etc. expected_auth_metadata = { "version": 2, "compat_version": 1, "dirty": False, "tenant_id": "tenant1", "volumes": { "groupid/volumeid": { "dirty": False, "access_level": "rw" } } } auth_metadata = self._volume_client_python(volumeclient_mount, dedent(""" import json vp = VolumePath("{group_id}", "{volume_id}") auth_metadata = vc._auth_metadata_get("{auth_id}") print(json.dumps(auth_metadata)) """.format( group_id=group_id, volume_id=volume_id, auth_id=guestclient_1["auth_id"], ))) auth_metadata = json.loads(auth_metadata) self.assertGreaterEqual(auth_metadata["version"], expected_auth_metadata["version"]) del expected_auth_metadata["version"] del auth_metadata["version"] self.assertEqual(expected_auth_metadata, auth_metadata) # Verify that the volume metadata file stores info about auth IDs # and their access levels to the volume, versioning details, etc. expected_vol_metadata = { "version": 2, "compat_version": 1, "auths": { "guest": { "dirty": False, "access_level": "rw" } } } vol_metadata = self._volume_client_python(volumeclient_mount, dedent(""" import json vp = VolumePath("{group_id}", "{volume_id}") volume_metadata = vc._volume_metadata_get(vp) print(json.dumps(volume_metadata)) """.format( group_id=group_id, volume_id=volume_id, ))) vol_metadata = json.loads(vol_metadata) self.assertGreaterEqual(vol_metadata["version"], expected_vol_metadata["version"]) del expected_vol_metadata["version"] del vol_metadata["version"] self.assertEqual(expected_vol_metadata, vol_metadata) # Cannot authorize 'guestclient_2' to access the volume. # It uses auth ID 'guest', which has already been used by a # 'guestclient_1' belonging to an another tenant for accessing # the volume. with self.assertRaises(CommandFailedError): self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.authorize(vp, "{auth_id}", tenant_id="{tenant_id}") """.format( group_id=group_id, volume_id=volume_id, auth_id=guestclient_2["auth_id"], tenant_id=guestclient_2["tenant_id"] ))) # Check that auth metadata file is cleaned up on removing # auth ID's only access to a volume. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.deauthorize(vp, "{guest_entity}") """.format( group_id=group_id, volume_id=volume_id, guest_entity=guestclient_1["auth_id"] ))) self.assertNotIn(auth_metadata_filename, self.mounts[0].ls("volumes")) # Check that volume metadata file is cleaned up on volume deletion. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.delete_volume(vp) """.format( group_id=group_id, volume_id=volume_id, ))) self.assertNotIn(vol_metadata_filename, self.mounts[0].ls("volumes")) def test_recover_metadata(self): """ That volume client can recover from partial auth updates using metadata files, which store auth info and its update status info. """ volumeclient_mount = self.mounts[1] volumeclient_mount.umount_wait() # Configure volumeclient_mount as the handle for driving volumeclient. self._configure_vc_auth(volumeclient_mount, "manila") group_id = "groupid" volume_id = "volumeid" guestclient = { "auth_id": "guest", "tenant_id": "tenant", } # Create a volume. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.create_volume(vp, 1024*1024*10) """.format( group_id=group_id, volume_id=volume_id, ))) # Authorize 'guestclient' access to the volume. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.authorize(vp, "{auth_id}", tenant_id="{tenant_id}") """.format( group_id=group_id, volume_id=volume_id, auth_id=guestclient["auth_id"], tenant_id=guestclient["tenant_id"] ))) # Check that auth metadata file for auth ID 'guest' is created. auth_metadata_filename = "${0}.meta".format(guestclient["auth_id"]) self.assertIn(auth_metadata_filename, self.mounts[0].ls("volumes")) # Induce partial auth update state by modifying the auth metadata file, # and then run recovery procedure. self._volume_client_python(volumeclient_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") auth_metadata = vc._auth_metadata_get("{auth_id}") auth_metadata['dirty'] = True vc._auth_metadata_set("{auth_id}", auth_metadata) vc.recover() """.format( group_id=group_id, volume_id=volume_id, auth_id=guestclient["auth_id"], ))) def test_put_object(self): vc_mount = self.mounts[1] vc_mount.umount_wait() self._configure_vc_auth(vc_mount, "manila") obj_data = 'test data' obj_name = 'test_vc_obj_1' pool_name = self.fs.get_data_pool_names()[0] self._volume_client_python(vc_mount, dedent(""" vc.put_object("{pool_name}", "{obj_name}", b"{obj_data}") """.format( pool_name = pool_name, obj_name = obj_name, obj_data = obj_data ))) read_data = self.fs.rados(['get', obj_name, '-'], pool=pool_name) self.assertEqual(obj_data, read_data) def test_get_object(self): vc_mount = self.mounts[1] vc_mount.umount_wait() self._configure_vc_auth(vc_mount, "manila") obj_data = 'test_data' obj_name = 'test_vc_ob_2' pool_name = self.fs.get_data_pool_names()[0] self.fs.rados(['put', obj_name, '-'], pool=pool_name, stdin_data=obj_data) self._volume_client_python(vc_mount, dedent(""" data_read = vc.get_object("{pool_name}", "{obj_name}") assert data_read == b"{obj_data}" """.format( pool_name = pool_name, obj_name = obj_name, obj_data = obj_data ))) def test_put_object_versioned(self): vc_mount = self.mounts[1] vc_mount.umount_wait() self._configure_vc_auth(vc_mount, "manila") obj_data = 'test_data' obj_name = 'test_vc_ob_2' pool_name = self.fs.get_data_pool_names()[0] self.fs.rados(['put', obj_name, '-'], pool=pool_name, stdin_data=obj_data) # Test if put_object_versioned() crosschecks the version of the # given object. Being a negative test, an exception is expected. with self.assertRaises(CommandFailedError): self._volume_client_python(vc_mount, dedent(""" data, version = vc.get_object_and_version("{pool_name}", "{obj_name}") data += 'm1' vc.put_object("{pool_name}", "{obj_name}", data) data += 'm2' vc.put_object_versioned("{pool_name}", "{obj_name}", data, version) """).format(pool_name=pool_name, obj_name=obj_name)) def test_delete_object(self): vc_mount = self.mounts[1] vc_mount.umount_wait() self._configure_vc_auth(vc_mount, "manila") obj_data = 'test data' obj_name = 'test_vc_obj_3' pool_name = self.fs.get_data_pool_names()[0] self.fs.rados(['put', obj_name, '-'], pool=pool_name, stdin_data=obj_data) self._volume_client_python(vc_mount, dedent(""" data_read = vc.delete_object("{pool_name}", "{obj_name}") """.format( pool_name = pool_name, obj_name = obj_name, ))) with self.assertRaises(CommandFailedError): self.fs.rados(['stat', obj_name], pool=pool_name) # Check idempotency -- no error raised trying to delete non-existent # object self._volume_client_python(vc_mount, dedent(""" data_read = vc.delete_object("{pool_name}", "{obj_name}") """.format( pool_name = pool_name, obj_name = obj_name, ))) def test_21501(self): """ Reproducer for #21501 "ceph_volume_client: sets invalid caps for existing IDs with no caps" (http://tracker.ceph.com/issues/21501) """ vc_mount = self.mounts[1] vc_mount.umount_wait() # Configure vc_mount as the handle for driving volumeclient self._configure_vc_auth(vc_mount, "manila") # Create a volume group_id = "grpid" volume_id = "volid" mount_path = self._volume_client_python(vc_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") create_result = vc.create_volume(vp, 1024*1024*10) print(create_result['mount_path']) """.format( group_id=group_id, volume_id=volume_id ))) # Create an auth ID with no caps guest_id = '21501' self.fs.mon_manager.raw_cluster_cmd_result( 'auth', 'get-or-create', 'client.{0}'.format(guest_id)) guest_mount = self.mounts[2] guest_mount.umount_wait() # Set auth caps for the auth ID using the volumeclient self._configure_guest_auth(vc_mount, guest_mount, guest_id, mount_path) # Mount the volume in the guest using the auth ID to assert that the # auth caps are valid guest_mount.mount(mount_path=mount_path) def test_volume_without_namespace_isolation(self): """ That volume client can create volumes that do not have separate RADOS namespace layouts. """ vc_mount = self.mounts[1] vc_mount.umount_wait() # Configure vc_mount as the handle for driving volumeclient self._configure_vc_auth(vc_mount, "manila") # Create a volume volume_prefix = "/myprefix" group_id = "grpid" volume_id = "volid" mount_path = self._volume_client_python(vc_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") create_result = vc.create_volume(vp, 1024*1024*10, namespace_isolated=False) print(create_result['mount_path']) """.format( group_id=group_id, volume_id=volume_id )), volume_prefix) # The CephFS volume should be created self.mounts[0].stat(os.path.join("myprefix", group_id, volume_id)) vol_namespace = self.mounts[0].getfattr( os.path.join("myprefix", group_id, volume_id), "ceph.dir.layout.pool_namespace") assert not vol_namespace self._volume_client_python(vc_mount, dedent(""" vp = VolumePath("{group_id}", "{volume_id}") vc.delete_volume(vp) vc.purge_volume(vp) """.format( group_id=group_id, volume_id=volume_id, )), volume_prefix)