path: root/qa/tasks/mgr/dashboard/test_pool.py

# -*- coding: utf-8 -*-
from __future__ import absolute_import

import logging
import time
from contextlib import contextmanager

from .helper import DashboardTestCase, JAny, JList, JObj, JUnion

log = logging.getLogger(__name__)


class PoolTest(DashboardTestCase):
    AUTH_ROLES = ['pool-manager']

    pool_schema = JObj(sub_elems={
        'pool_name': str,
        'type': str,
        'application_metadata': JList(str),
        'flags': int,
        'flags_names': str,
    }, allow_unknown=True)

    pool_list_stat_schema = JObj(sub_elems={
        'latest': JUnion([int, float]),
        'rate': float,
        'rates': JList(JAny(none=False)),
    })

    pool_list_stats_schema = JObj(sub_elems={
        'avail_raw': pool_list_stat_schema,
        'bytes_used': pool_list_stat_schema,
        'max_avail': pool_list_stat_schema,
        'percent_used': pool_list_stat_schema,
        'rd_bytes': pool_list_stat_schema,
        'wr_bytes': pool_list_stat_schema,
        'rd': pool_list_stat_schema,
        'wr': pool_list_stat_schema,
    }, allow_unknown=True)

    pool_rbd_conf_schema = JList(JObj(sub_elems={
        'name': str,
        'value': str,
        'source': int
    }))

    @contextmanager
    def __yield_pool(self, name=None, data=None, deletion_name=None):
        """
        Use either just a name or whole description of a pool to create one.
        This also validates the correct creation and deletion after the pool was used.

        :param name: Name of the pool
        :param data: Describes the pool in full length
        :param deletion_name: Only needed if the pool was renamed
        :return:
        """
        data = self._create_pool(name, data)
        yield data
        self._delete_pool(deletion_name or data['pool'])

    def _create_pool(self, name, data):
        data = data or {
            'pool': name,
            'pg_num': '32',
            'pool_type': 'replicated',
            'compression_algorithm': 'snappy',
            'compression_mode': 'passive',
            'compression_max_blob_size': '131072',
            'compression_required_ratio': '0.875',
            'application_metadata': ['rbd'],
            'configuration': {
                'rbd_qos_bps_limit': 1024000,
                'rbd_qos_iops_limit': 5000,
            }
        }
        self._task_post('/api/pool/', data)
        self.assertStatus(201)
        self._validate_pool_properties(data, self._get_pool(data['pool']))
        return data

    def _delete_pool(self, name):
        self._task_delete('/api/pool/' + name)
        self.assertStatus(204)

    def _validate_pool_properties(self, data, pool, timeout=DashboardTestCase.TIMEOUT_HEALTH_CLEAR):
        # pylint: disable=too-many-branches
        for prop, value in data.items():
            if prop == 'pool_type':
                self.assertEqual(pool['type'], value)
            elif prop == 'size':
                self.assertEqual(pool[prop], int(value),
                                 '{}: {} != {}'.format(prop, pool[prop], value))
            elif prop == 'pg_num':
                self._check_pg_num(pool['pool_name'], int(value))
            elif prop == 'application_metadata':
                self.assertIsInstance(pool[prop], list)
                self.assertEqual(value, pool[prop])
            elif prop == 'pool':
                self.assertEqual(pool['pool_name'], value)
            elif prop.startswith('compression'):
                if value is not None:
                    if prop.endswith('size'):
                        value = int(value)
                    elif prop.endswith('ratio'):
                        value = float(value)
                    self.assertEqual(pool['options'][prop], value)
                else:
                    self.assertEqual(pool['options'], {})
            elif prop == 'configuration':
                # configuration cannot really be checked here for two reasons:
                #   1.  The default value cannot be given to this method, which becomes relevant
                #       when resetting a value, because it's not always zero.
                #   2.  The expected `source` cannot be given to this method, and it cannot
                #       reliably be determined (see 1)
                pass
            else:
                self.assertEqual(pool[prop], value, '{}: {} != {}'.format(prop, pool[prop], value))

        self.wait_for_health_clear(timeout)

    def _get_pool(self, pool_name):
        pool = self._get("/api/pool/" + pool_name)
        self.assertStatus(200)
        self.assertSchemaBody(self.pool_schema)
        return pool

    def _check_pg_num(self, pool_name, pg_num):
        """
        If both properties have not the same value, the cluster goes into a warning state, which
        will only happen during a pg update on an existing pool. The test that does that is
        currently commented out because our QA systems can't deal with the change. Feel free to test
        it locally.
        """
        self.wait_until_equal(
            lambda: self._get_pool(pool_name)['pg_placement_num'],
            expect_val=pg_num,
            timeout=180
        )

        pool = self._get_pool(pool_name)

        for prop in ['pg_num', 'pg_placement_num']:
            self.assertEqual(pool[prop], int(pg_num),
                             '{}: {} != {}'.format(prop, pool[prop], pg_num))

    @DashboardTestCase.RunAs('test', 'test', [{'pool': ['create', 'update', 'delete']}])
    def test_read_access_permissions(self):
        self._get('/api/pool')
        self.assertStatus(403)
        self._get('/api/pool/bla')
        self.assertStatus(403)

    @DashboardTestCase.RunAs('test', 'test', [{'pool': ['read', 'update', 'delete']}])
    def test_create_access_permissions(self):
        self._task_post('/api/pool/', {})
        self.assertStatus(403)

    @DashboardTestCase.RunAs('test', 'test', [{'pool': ['read', 'create', 'update']}])
    def test_delete_access_permissions(self):
        self._delete('/api/pool/ddd')
        self.assertStatus(403)

    def test_pool_configuration(self):
        pool_name = '.mgr'
        data = self._get('/api/pool/{}/configuration'.format(pool_name))
        self.assertStatus(200)
        self.assertSchema(data, JList(JObj({
            'name': str,
            'value': str,
            'source': int
        })))

    def test_pool_list(self):
        data = self._get("/api/pool")
        self.assertStatus(200)

        cluster_pools = self.ceph_cluster.mon_manager.list_pools()
        self.assertEqual(len(cluster_pools), len(data))
        self.assertSchemaBody(JList(self.pool_schema))
        for pool in data:
            self.assertNotIn('pg_status', pool)
            self.assertNotIn('stats', pool)
            self.assertIn(pool['pool_name'], cluster_pools)

    def test_pool_list_attrs(self):
        data = self._get("/api/pool?attrs=type,flags")
        self.assertStatus(200)

        cluster_pools = self.ceph_cluster.mon_manager.list_pools()
        self.assertEqual(len(cluster_pools), len(data))
        for pool in data:
            self.assertIn('pool_name', pool)
            self.assertIn('type', pool)
            self.assertIn('flags', pool)
            self.assertNotIn('flags_names', pool)
            self.assertNotIn('pg_status', pool)
            self.assertNotIn('stats', pool)
            self.assertIn(pool['pool_name'], cluster_pools)

    def test_pool_list_stats(self):
        data = self._get("/api/pool?stats=true")
        self.assertStatus(200)

        cluster_pools = self.ceph_cluster.mon_manager.list_pools()
        self.assertEqual(len(cluster_pools), len(data))
        self.assertSchemaBody(JList(self.pool_schema))
        for pool in data:
            self.assertIn('pool_name', pool)
            self.assertIn('type', pool)
            self.assertIn('application_metadata', pool)
            self.assertIn('flags', pool)
            self.assertIn('pg_status', pool)
            self.assertSchema(pool['stats'], self.pool_list_stats_schema)
            self.assertIn('flags_names', pool)
            self.assertIn(pool['pool_name'], cluster_pools)

    def test_pool_get(self):
        cluster_pools = self.ceph_cluster.mon_manager.list_pools()
        pool = self._get("/api/pool/{}?stats=true&attrs=type,flags,stats"
                         .format(cluster_pools[0]))
        self.assertEqual(pool['pool_name'], cluster_pools[0])
        self.assertIn('type', pool)
        self.assertIn('flags', pool)
        self.assertNotIn('pg_status', pool)
        self.assertSchema(pool['stats'], self.pool_list_stats_schema)
        self.assertNotIn('flags_names', pool)
        self.assertSchema(pool['configuration'], self.pool_rbd_conf_schema)

    def test_pool_create_with_two_applications(self):
        self.__yield_pool(None, {
            'pool': 'dashboard_pool1',
            'pg_num': '32',
            'pool_type': 'replicated',
            'application_metadata': ['rbd', 'sth'],
        })

    def test_pool_create_with_ecp_and_rule(self):
        self._ceph_cmd(['osd', 'crush', 'rule', 'create-erasure', 'ecrule'])
        self._ceph_cmd(
            ['osd', 'erasure-code-profile', 'set', 'ecprofile', 'crush-failure-domain=osd'])
        self.__yield_pool(None, {
            'pool': 'dashboard_pool2',
            'pg_num': '32',
            'pool_type': 'erasure',
            'application_metadata': ['rbd'],
            'erasure_code_profile': 'ecprofile',
            'crush_rule': 'ecrule',
        })
        self._ceph_cmd(['osd', 'erasure-code-profile', 'rm', 'ecprofile'])

    def test_pool_create_with_compression(self):
        pool = {
            'pool': 'dashboard_pool3',
            'pg_num': '32',
            'pool_type': 'replicated',
            'compression_algorithm': 'zstd',
            'compression_mode': 'aggressive',
            'compression_max_blob_size': '10000000',
            'compression_required_ratio': '0.8',
            'application_metadata': ['rbd'],
            'configuration': {
                'rbd_qos_bps_limit': 2048,
                'rbd_qos_iops_limit': None,
            },
        }
        with self.__yield_pool(None, pool):
            expected_configuration = [{
                'name': 'rbd_qos_bps_limit',
                'source': 1,
                'value': '2048',
            }, {
                'name': 'rbd_qos_iops_limit',
                'source': 0,
                'value': '0',
            }]
            new_pool = self._get_pool(pool['pool'])
            for conf in expected_configuration:
                self.assertIn(conf, new_pool['configuration'])

    def test_pool_create_with_quotas(self):
        pools = [
            {
                'pool_data': {
                    'pool': 'dashboard_pool_quota1',
                    'pg_num': '32',
                    'pool_type': 'replicated',
                    'application_metadata': ['rbd'],
                },
                'pool_quotas_to_check': {
                    'quota_max_objects': 0,
                    'quota_max_bytes': 0,
                }
            },
            {
                'pool_data': {
                    'pool': 'dashboard_pool_quota2',
                    'pg_num': '32',
                    'pool_type': 'replicated',
                    'application_metadata': ['rbd'],
                    'quota_max_objects': 1024,
                    'quota_max_bytes': 1000,
                },
                'pool_quotas_to_check': {
                    'quota_max_objects': 1024,
                    'quota_max_bytes': 1000,
                }
            }
        ]

        for pool in pools:
            pool_name = pool['pool_data']['pool']
            with self.__yield_pool(pool_name, pool['pool_data']):
                self._validate_pool_properties(pool['pool_quotas_to_check'],
                                               self._get_pool(pool_name))

    def test_pool_update_name(self):
        name = 'pool_update'
        updated_name = 'pool_updated_name'
        with self.__yield_pool(name, None, updated_name):
            props = {'pool': updated_name}
            self._task_put('/api/pool/{}'.format(name), props)
            time.sleep(5)
            self.assertStatus(200)
            self._validate_pool_properties(props, self._get_pool(updated_name))

    def test_pool_update_metadata(self):
        pool_name = 'pool_update_metadata'
        with self.__yield_pool(pool_name):
            props = {'application_metadata': ['rbd', 'sth']}
            self._task_put('/api/pool/{}'.format(pool_name), props)
            self._validate_pool_properties(props, self._get_pool(pool_name),
                                           self.TIMEOUT_HEALTH_CLEAR * 2)

            properties = {'application_metadata': ['rgw']}
            self._task_put('/api/pool/' + pool_name, properties)
            self._validate_pool_properties(properties, self._get_pool(pool_name),
                                           self.TIMEOUT_HEALTH_CLEAR * 2)

            properties = {'application_metadata': ['rbd', 'sth']}
            self._task_put('/api/pool/' + pool_name, properties)
            self._validate_pool_properties(properties, self._get_pool(pool_name),
                                           self.TIMEOUT_HEALTH_CLEAR * 2)

            properties = {'application_metadata': ['rgw']}
            self._task_put('/api/pool/' + pool_name, properties)
            self._validate_pool_properties(properties, self._get_pool(pool_name),
                                           self.TIMEOUT_HEALTH_CLEAR * 2)

    def test_pool_update_configuration(self):
        pool_name = 'pool_update_configuration'
        with self.__yield_pool(pool_name):
            configuration = {
                'rbd_qos_bps_limit': 1024,
                'rbd_qos_iops_limit': None,
            }
            expected_configuration = [{
                'name': 'rbd_qos_bps_limit',
                'source': 1,
                'value': '1024',
            }, {
                'name': 'rbd_qos_iops_limit',
                'source': 0,
                'value': '0',
            }]
            self._task_put('/api/pool/' + pool_name, {'configuration': configuration})
            time.sleep(5)
            pool_config = self._get_pool(pool_name)['configuration']
            for conf in expected_configuration:
                self.assertIn(conf, pool_config)

    def test_pool_update_compression(self):
        pool_name = 'pool_update_compression'
        with self.__yield_pool(pool_name):
            properties = {
                'compression_algorithm': 'zstd',
                'compression_mode': 'aggressive',
                'compression_max_blob_size': '10000000',
                'compression_required_ratio': '0.8',
            }
            self._task_put('/api/pool/' + pool_name, properties)
            time.sleep(5)
            self._validate_pool_properties(properties, self._get_pool(pool_name))

    def test_pool_update_unset_compression(self):
        pool_name = 'pool_update_unset_compression'
        with self.__yield_pool(pool_name):
            self._task_put('/api/pool/' + pool_name, {'compression_mode': 'unset'})
            time.sleep(5)
            self._validate_pool_properties({
                'compression_algorithm': None,
                'compression_mode': None,
                'compression_max_blob_size': None,
                'compression_required_ratio': None,
            }, self._get_pool(pool_name))

    def test_pool_update_quotas(self):
        pool_name = 'pool_update_quotas'
        with self.__yield_pool(pool_name):
            properties = {
                'quota_max_objects': 1024,
                'quota_max_bytes': 1000,
            }
            self._task_put('/api/pool/' + pool_name, properties)
            time.sleep(5)
            self._validate_pool_properties(properties, self._get_pool(pool_name))

    def test_pool_create_fail(self):
        data = {'pool_type': u'replicated', 'rule_name': u'dnf', 'pg_num': u'8', 'pool': u'sadfs'}
        self._task_post('/api/pool/', data)
        self.assertStatus(400)
        self.assertJsonBody({
            'component': 'pool',
            'code': "2",
            'detail': "[errno -2] specified rule dnf doesn't exist"
        })

    def test_pool_info(self):
        self._get("/ui-api/pool/info")
        self.assertSchemaBody(JObj({
            'pool_names': JList(str),
            'compression_algorithms': JList(str),
            'compression_modes': JList(str),
            'is_all_bluestore': bool,
            'bluestore_compression_algorithm': str,
            'osd_count': int,
            'crush_rules_replicated': JList(JObj({}, allow_unknown=True)),
            'crush_rules_erasure': JList(JObj({}, allow_unknown=True)),
            'pg_autoscale_default_mode': str,
            'pg_autoscale_modes': JList(str),
            'erasure_code_profiles': JList(JObj({}, allow_unknown=True)),
            'used_rules': JObj({}, allow_unknown=True),
            'used_profiles': JObj({}, allow_unknown=True),
            'nodes': JList(JObj({}, allow_unknown=True)),
        }))