1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
from mgr_module import MgrModule, CommandResult, Option, NotifyType
import json
import threading
from typing import cast, Any
class Module(MgrModule):
MODULE_OPTIONS = [
Option(
name='subtree',
type='str',
default='rack',
desc='CRUSH level for which to create a local pool',
long_desc='which CRUSH subtree type the module should create a pool for.',
runtime=True),
Option(
name='failure_domain',
type='str',
default='host',
desc='failure domain for any created local pool',
long_desc='what failure domain we should separate data replicas across.',
runtime=True),
Option(
name='min_size',
type='int',
desc='default min_size for any created local pool',
long_desc='value to set min_size to (unchanged from Ceph\'s default if this option is not set)',
runtime=True),
Option(
name='num_rep',
type='int',
default=3,
desc='default replica count for any created local pool',
runtime=True),
Option(
name='pg_num',
type='int',
default=128,
desc='default pg_num for any created local pool',
runtime=True),
Option(
name='prefix',
type='str',
default='',
desc='name prefix for any created local pool',
runtime=True),
]
NOTIFY_TYPES = [NotifyType.osd_map]
def __init__(self, *args: Any, **kwargs: Any) -> None:
super(Module, self).__init__(*args, **kwargs)
self.serve_event = threading.Event()
def notify(self, notify_type: NotifyType, notify_id: str) -> None:
if notify_type == NotifyType.osd_map:
self.handle_osd_map()
def handle_osd_map(self) -> None:
"""
Check pools on each OSDMap change
"""
subtree_type = cast(str, self.get_module_option('subtree'))
failure_domain = self.get_module_option('failure_domain')
pg_num = self.get_module_option('pg_num')
num_rep = self.get_module_option('num_rep')
min_size = self.get_module_option('min_size')
prefix = cast(str, self.get_module_option('prefix')) or 'by-' + subtree_type + '-'
osdmap = self.get("osd_map")
lpools = []
for pool in osdmap['pools']:
if pool['pool_name'].find(prefix) == 0:
lpools.append(pool['pool_name'])
self.log.debug('localized pools = %s', lpools)
subtrees = []
tree = self.get('osd_map_tree')
for node in tree['nodes']:
if node['type'] == subtree_type:
subtrees.append(node['name'])
pool_name = prefix + node['name']
if pool_name not in lpools:
self.log.info('Creating localized pool %s', pool_name)
#
result = CommandResult("")
self.send_command(result, "mon", "", json.dumps({
"prefix": "osd crush rule create-replicated",
"format": "json",
"name": pool_name,
"root": node['name'],
"type": failure_domain,
}), "")
r, outb, outs = result.wait()
result = CommandResult("")
self.send_command(result, "mon", "", json.dumps({
"prefix": "osd pool create",
"format": "json",
"pool": pool_name,
'rule': pool_name,
"pool_type": 'replicated',
'pg_num': pg_num,
}), "")
r, outb, outs = result.wait()
result = CommandResult("")
self.send_command(result, "mon", "", json.dumps({
"prefix": "osd pool set",
"format": "json",
"pool": pool_name,
'var': 'size',
"val": str(num_rep),
}), "")
r, outb, outs = result.wait()
if min_size:
result = CommandResult("")
self.send_command(result, "mon", "", json.dumps({
"prefix": "osd pool set",
"format": "json",
"pool": pool_name,
'var': 'min_size',
"val": str(min_size),
}), "")
r, outb, outs = result.wait()
# TODO remove pools for hosts that don't exist?
def serve(self) -> None:
self.handle_osd_map()
self.serve_event.wait()
self.serve_event.clear()
def shutdown(self) -> None:
self.serve_event.set()
|
