ceph/src/pybind/mgr/dashboard/controllers/pool.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import time
   5 import cherrypy
   6
   7 from . import ApiController, RESTController, Endpoint, ReadPermission, Task, UiApiController
   8 from .. import mgr
   9 from ..security import Scope
  10 from ..services.ceph_service import CephService
  11 from ..services.rbd import RbdConfiguration
  12 from ..services.exception import handle_send_command_error
  13 from ..tools import str_to_bool, TaskManager
  14
  15
  16 def pool_task(name, metadata, wait_for=2.0):
  17     return Task("pool/{}".format(name), metadata, wait_for)
  18
  19
  20 @ApiController('/pool', Scope.POOL)
  21 class Pool(RESTController):
  22
  23     @staticmethod
  24     def _serialize_pool(pool, attrs):
  25         if not attrs or not isinstance(attrs, list):
  26             attrs = pool.keys()
  27
  28         crush_rules = {r['rule_id']: r["rule_name"] for r in mgr.get('osd_map_crush')['rules']}
  29
  30         res = {}
  31         for attr in attrs:
  32             if attr not in pool:
  33                 continue
  34             if attr == 'type':
  35                 res[attr] = {1: 'replicated', 3: 'erasure'}[pool[attr]]
  36             elif attr == 'crush_rule':
  37                 res[attr] = crush_rules[pool[attr]]
  38             elif attr == 'application_metadata':
  39                 res[attr] = list(pool[attr].keys())
  40             else:
  41                 res[attr] = pool[attr]
  42
  43         # pool_name is mandatory
  44         res['pool_name'] = pool['pool_name']
  45         return res
  46
  47     @classmethod
  48     def _pool_list(cls, attrs=None, stats=False):
  49         if attrs:
  50             attrs = attrs.split(',')
  51
  52         if str_to_bool(stats):
  53             pools = CephService.get_pool_list_with_stats()
  54         else:
  55             pools = CephService.get_pool_list()
  56
  57         return [cls._serialize_pool(pool, attrs) for pool in pools]
  58
  59     def list(self, attrs=None, stats=False):
  60         return self._pool_list(attrs, stats)
  61
  62     @classmethod
  63     def _get(cls, pool_name, attrs=None, stats=False):
  64         # type: (str, str, bool) -> dict
  65         pools = cls._pool_list(attrs, stats)
  66         pool = [p for p in pools if p['pool_name'] == pool_name]
  67         if not pool:
  68             raise cherrypy.NotFound('No such pool')
  69         return pool[0]
  70
  71     def get(self, pool_name, attrs=None, stats=False):
  72         # type: (str, str, bool) -> dict
  73         pool = self._get(pool_name, attrs, stats)
  74         pool['configuration'] = RbdConfiguration(pool_name).list()
  75         return pool
  76
  77     @pool_task('delete', ['{pool_name}'])
  78     @handle_send_command_error('pool')
  79     def delete(self, pool_name):
  80         return CephService.send_command('mon', 'osd pool delete', pool=pool_name, pool2=pool_name,
  81                                         yes_i_really_really_mean_it=True)
  82
  83     @pool_task('edit', ['{pool_name}'])
  84     def set(self, pool_name, flags=None, application_metadata=None, configuration=None, **kwargs):
  85         self._set_pool_values(pool_name, application_metadata, flags, True, kwargs)
  86         if kwargs.get('pool'):
  87             pool_name = kwargs['pool']
  88         RbdConfiguration(pool_name).set_configuration(configuration)
  89         self._wait_for_pgs(pool_name)
  90
  91     @pool_task('create', {'pool_name': '{pool}'})
  92     @handle_send_command_error('pool')
  93     def create(self, pool, pg_num, pool_type, erasure_code_profile=None, flags=None,
  94                application_metadata=None, rule_name=None, configuration=None, **kwargs):
  95         ecp = erasure_code_profile if erasure_code_profile else None
  96         CephService.send_command('mon', 'osd pool create', pool=pool, pg_num=int(pg_num),
  97                                  pgp_num=int(pg_num), pool_type=pool_type, erasure_code_profile=ecp,
  98                                  rule=rule_name)
  99         self._set_pool_values(pool, application_metadata, flags, False, kwargs)
 100         RbdConfiguration(pool).set_configuration(configuration)
 101         self._wait_for_pgs(pool)
 102
 103     def _set_pool_values(self, pool, application_metadata, flags, update_existing, kwargs):
 104         update_name = False
 105         current_pool = self._get(pool)
 106         if update_existing and kwargs.get('compression_mode') == 'unset':
 107             self._prepare_compression_removal(current_pool.get('options'), kwargs)
 108         if flags and 'ec_overwrites' in flags:
 109             CephService.send_command('mon', 'osd pool set', pool=pool, var='allow_ec_overwrites',
 110                                      val='true')
 111         if application_metadata is not None:
 112             def set_app(what, app):
 113                 CephService.send_command('mon', 'osd pool application ' + what, pool=pool, app=app,
 114                                          yes_i_really_mean_it=True)
 115             if update_existing:
 116                 original_app_metadata = set(
 117                     current_pool.get('application_metadata'))
 118             else:
 119                 original_app_metadata = set()
 120
 121             for app in original_app_metadata - set(application_metadata):
 122                 set_app('disable', app)
 123             for app in set(application_metadata) - original_app_metadata:
 124                 set_app('enable', app)
 125
 126         def set_key(key, value):
 127             CephService.send_command('mon', 'osd pool set', pool=pool, var=key, val=str(value))
 128
 129         quotas = {}
 130         quotas['max_objects'] = kwargs.pop('quota_max_objects', None)
 131         quotas['max_bytes'] = kwargs.pop('quota_max_bytes', None)
 132         self._set_quotas(pool, quotas)
 133
 134         for key, value in kwargs.items():
 135             if key == 'pool':
 136                 update_name = True
 137                 destpool = value
 138             else:
 139                 set_key(key, value)
 140                 if key == 'pg_num':
 141                     set_key('pgp_num', value)
 142         if update_name:
 143             CephService.send_command('mon', 'osd pool rename', srcpool=pool, destpool=destpool)
 144
 145     def _set_quotas(self, pool, quotas):
 146         for field, value in quotas.items():
 147             if value is not None:
 148                 CephService.send_command('mon', 'osd pool set-quota',
 149                                          pool=pool, field=field, val=str(value))
 150
 151     def _prepare_compression_removal(self, options, kwargs):
 152         """
 153         Presets payload with values to remove compression attributes in case they are not
 154         needed anymore.
 155
 156         In case compression is not needed the dashboard will send 'compression_mode' with the
 157         value 'unset'.
 158
 159         :param options: All set options for the current pool.
 160         :param kwargs: Payload of the PUT / POST call
 161         """
 162         if options is not None:
 163             def reset_arg(arg, value):
 164                 if options.get(arg):
 165                     kwargs[arg] = value
 166             for arg in ['compression_min_blob_size', 'compression_max_blob_size',
 167                         'compression_required_ratio']:
 168                 reset_arg(arg, '0')
 169             reset_arg('compression_algorithm', 'unset')
 170
 171     @classmethod
 172     def _wait_for_pgs(cls, pool_name):
 173         """
 174         Keep the task waiting for until all pg changes are complete
 175         :param pool_name: The name of the pool.
 176         :type pool_name: string
 177         """
 178         current_pool = cls._get(pool_name)
 179         initial_pgs = int(current_pool['pg_placement_num']) + int(current_pool['pg_num'])
 180         cls._pg_wait_loop(current_pool, initial_pgs)
 181
 182     @classmethod
 183     def _pg_wait_loop(cls, pool, initial_pgs):
 184         """
 185         Compares if all pg changes are completed, if not it will call itself
 186         until all changes are completed.
 187         :param pool: The dict that represents a pool.
 188         :type pool: dict
 189         :param initial_pgs: The pg and pg_num count before any change happened.
 190         :type initial_pgs: int
 191         """
 192         if 'pg_num_target' in pool:
 193             target = int(pool['pg_num_target']) + int(pool['pg_placement_num_target'])
 194             current = int(pool['pg_placement_num']) + int(pool['pg_num'])
 195             if current != target:
 196                 max_diff = abs(target - initial_pgs)
 197                 diff = max_diff - abs(target - current)
 198                 percentage = int(round(diff / float(max_diff) * 100))
 199                 TaskManager.current_task().set_progress(percentage)
 200                 time.sleep(4)
 201                 cls._pg_wait_loop(cls._get(pool['pool_name']), initial_pgs)
 202
 203     @RESTController.Resource()
 204     @ReadPermission
 205     def configuration(self, pool_name):
 206         return RbdConfiguration(pool_name).list()
 207
 208
 209 @UiApiController('/pool', Scope.POOL)
 210 class PoolUi(Pool):
 211     @Endpoint()
 212     @ReadPermission
 213     def info(self):
 214         """Used by the create-pool dialog"""
 215         osd_map_crush = mgr.get('osd_map_crush')
 216         options = mgr.get('config_options')['options']
 217
 218         def rules(pool_type):
 219             return [r
 220                     for r in osd_map_crush['rules']
 221                     if r['type'] == pool_type]
 222
 223         def all_bluestore():
 224             return all(o['osd_objectstore'] == 'bluestore'
 225                        for o in mgr.get('osd_metadata').values())
 226
 227         def get_config_option_enum(conf_name):
 228             return [[v for v in o['enum_values'] if len(v) > 0]
 229                     for o in options
 230                     if o['name'] == conf_name][0]
 231
 232         used_rules = {}
 233         pool_names = []
 234         for p in self._pool_list():
 235             name = p['pool_name']
 236             rule = p['crush_rule']
 237             pool_names.append(name)
 238             if rule in used_rules:
 239                 used_rules[rule].append(name)
 240             else:
 241                 used_rules[rule] = [name]
 242
 243         mgr_config = mgr.get('config')
 244         return {
 245             "pool_names": pool_names,
 246             "crush_rules_replicated": rules(1),
 247             "crush_rules_erasure": rules(3),
 248             "is_all_bluestore": all_bluestore(),
 249             "osd_count": len(mgr.get('osd_map')['osds']),
 250             "bluestore_compression_algorithm": mgr_config['bluestore_compression_algorithm'],
 251             "compression_algorithms": get_config_option_enum('bluestore_compression_algorithm'),
 252             "compression_modes": get_config_option_enum('bluestore_compression_mode'),
 253             "pg_autoscale_default_mode": mgr_config['osd_pool_default_pg_autoscale_mode'],
 254             "pg_autoscale_modes": get_config_option_enum('osd_pool_default_pg_autoscale_mode'),
 255             "erasure_code_profiles": CephService.get_erasure_code_profiles(),
 256             "used_rules": used_rules
 257         }