src/ceph/src/pybind/mgr/localpool/module.py

   1 from mgr_module import MgrModule, CommandResult
   2 import json
   3 import threading
   4
   5 class Module(MgrModule):
   6     def __init__(self, *args, **kwargs):
   7         super(Module, self).__init__(*args, **kwargs)
   8         self.serve_event = threading.Event()
   9
  10     def notify(self, notify_type, notify_id):
  11         if notify_type == 'osd_map':
  12             self.handle_osd_map()
  13
  14     def handle_osd_map(self):
  15         """
  16         Check pools on each OSDMap change
  17         """
  18         subtree_type = self.get_config('subtree') or 'rack'
  19         failure_domain = self.get_config('failure_domain') or 'host'
  20         pg_num = self.get_config('pg_num') or '128'
  21         num_rep = self.get_config('num_rep') or '3'
  22         min_size = self.get_config('min_size')
  23         prefix = self.get_config('prefix') or 'by-' + subtree_type + '-'
  24
  25         osdmap = self.get("osd_map")
  26         lpools = []
  27         for pool in osdmap['pools']:
  28             if pool['pool_name'].find(prefix) == 0:
  29                 lpools.append(pool['pool_name'])
  30
  31         self.log.debug('localized pools = %s', lpools)
  32         subtrees = []
  33         tree = self.get('osd_map_tree')
  34         for node in tree['nodes']:
  35             if node['type'] == subtree_type:
  36                 subtrees.append(node['name'])
  37                 pool_name = prefix + node['name']
  38                 if pool_name not in lpools:
  39                     self.log.info('Creating localized pool %s', pool_name)
  40                     #
  41                     result = CommandResult("")
  42                     self.send_command(result, "mon", "", json.dumps({
  43                         "prefix": "osd crush rule create-replicated",
  44                         "format": "json",
  45                         "name": pool_name,
  46                         "root": node['name'],
  47                         "type": failure_domain,
  48                     }), "")
  49                     r, outb, outs = result.wait()
  50
  51                     result = CommandResult("")
  52                     self.send_command(result, "mon", "", json.dumps({
  53                         "prefix": "osd pool create",
  54                         "format": "json",
  55                         "pool": pool_name,
  56                         'rule': pool_name,
  57                         'erasure_code_profile': pool_name,
  58                         "pool_type": 'replicated',
  59                         'pg_num': str(pg_num),
  60                     }), "")
  61                     r, outb, outs = result.wait()
  62
  63                     result = CommandResult("")
  64                     self.send_command(result, "mon", "", json.dumps({
  65                         "prefix": "osd pool set",
  66                         "format": "json",
  67                         "pool": pool_name,
  68                         'var': 'size',
  69                         "val": str(num_rep),
  70                     }), "")
  71                     r, outb, outs = result.wait()
  72
  73                     if min_size:
  74                         result = CommandResult("")
  75                         self.send_command(result, "mon", "", json.dumps({
  76                             "prefix": "osd pool set",
  77                             "format": "json",
  78                             "pool": pool_name,
  79                             'var': 'min_size',
  80                             "val": str(min_size),
  81                         }), "")
  82                         r, outb, outs = result.wait()
  83
  84         # TODO remove pools for hosts that don't exist?
  85
  86     def serve(self):
  87         self.handle_osd_map()
  88         self.serve_event.wait()
  89         self.serve_event.clear()
  90
  91     def shutdown(self):
  92         self.serve_event.set()