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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
from ..services.exception import handle_send_command_error
from ..tools import find_object_in_list, str_to_bool
from . import APIDoc, APIRouter, EndpointDoc, RESTController, allow_empty_body
MGR_MODULE_SCHEMA = ([{
"name": (str, "Module Name"),
"enabled": (bool, "Is Module Enabled"),
"always_on": (bool, "Is it an always on module?"),
"options": ({
"Option_name": ({
"name": (str, "Name of the option"),
"type": (str, "Type of the option"),
"level": (str, "Option level"),
"flags": (int, "List of flags associated"),
"default_value": (int, "Default value for the option"),
"min": (str, "Minimum value"),
"max": (str, "Maximum value"),
"enum_allowed": ([str], ""),
"desc": (str, "Description of the option"),
"long_desc": (str, "Elaborated description"),
"tags": ([str], "Tags associated with the option"),
"see_also": ([str], "Related options")
}, "Options")
}, "Module Options")
}])
@APIRouter('/mgr/module', Scope.CONFIG_OPT)
@APIDoc("Get details of MGR Module", "MgrModule")
class MgrModules(RESTController):
ignore_modules = ['selftest']
@EndpointDoc("List Mgr modules",
responses={200: MGR_MODULE_SCHEMA})
def list(self):
"""
Get the list of managed modules.
:return: A list of objects with the fields 'enabled', 'name' and 'options'.
:rtype: list
"""
result = []
mgr_map = mgr.get('mgr_map')
always_on_modules = mgr_map['always_on_modules'].get(mgr.release_name, [])
for module_config in mgr_map['available_modules']:
module_name = module_config['name']
if module_name not in self.ignore_modules:
always_on = module_name in always_on_modules
enabled = module_name in mgr_map['modules'] or always_on
result.append({
'name': module_name,
'enabled': enabled,
'always_on': always_on,
'options': self._convert_module_options(
module_config['module_options'])
})
return result
def get(self, module_name):
"""
Retrieve the values of the persistent configuration settings.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
:return: The values of the module options.
:rtype: dict
"""
assert self._is_module_managed(module_name)
options = self._get_module_options(module_name)
result = {}
for name, option in options.items():
result[name] = mgr.get_module_option_ex(module_name, name,
option['default_value'])
return result
@RESTController.Resource('PUT')
def set(self, module_name, config):
"""
Set the values of the persistent configuration settings.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
:param config: The values of the module options to be stored.
:type config: dict
"""
assert self._is_module_managed(module_name)
options = self._get_module_options(module_name)
for name in options.keys():
if name in config:
mgr.set_module_option_ex(module_name, name, config[name])
@RESTController.Resource('POST')
@handle_send_command_error('mgr_modules')
@allow_empty_body
def enable(self, module_name):
"""
Enable the specified Ceph Mgr module.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
"""
assert self._is_module_managed(module_name)
CephService.send_command(
'mon', 'mgr module enable', module=module_name)
@RESTController.Resource('POST')
@handle_send_command_error('mgr_modules')
@allow_empty_body
def disable(self, module_name):
"""
Disable the specified Ceph Mgr module.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
"""
assert self._is_module_managed(module_name)
CephService.send_command(
'mon', 'mgr module disable', module=module_name)
@RESTController.Resource('GET')
def options(self, module_name):
"""
Get the module options of the specified Ceph Mgr module.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
:return: The module options as list of dicts.
:rtype: list
"""
assert self._is_module_managed(module_name)
return self._get_module_options(module_name)
def _is_module_managed(self, module_name):
"""
Check if the specified Ceph Mgr module is managed by this service.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
:return: Returns ``true`` if the Ceph Mgr module is managed by
this service, otherwise ``false``.
:rtype: bool
"""
if module_name in self.ignore_modules:
return False
mgr_map = mgr.get('mgr_map')
for module_config in mgr_map['available_modules']:
if module_name == module_config['name']:
return True
return False
def _get_module_config(self, module_name):
"""
Helper function to get detailed module configuration.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
:return: The module information, e.g. module name, can run,
error string and available module options.
:rtype: dict or None
"""
mgr_map = mgr.get('mgr_map')
return find_object_in_list('name', module_name,
mgr_map['available_modules'])
def _get_module_options(self, module_name):
"""
Helper function to get the module options.
:param module_name: The name of the Ceph Mgr module.
:type module_name: str
:return: The module options.
:rtype: dict
"""
options = self._get_module_config(module_name)['module_options']
return self._convert_module_options(options)
def _convert_module_options(self, options):
# Workaround a possible bug in the Ceph Mgr implementation.
# Various fields (e.g. default_value, min, max) are always
# returned as a string.
for option in options.values():
if option['type'] == 'str':
if option['default_value'] == 'None': # This is Python None
option['default_value'] = ''
elif option['type'] == 'bool':
if option['default_value'] == '':
option['default_value'] = False
else:
option['default_value'] = str_to_bool(
option['default_value'])
elif option['type'] in ['float', 'uint', 'int', 'size', 'secs']:
cls = {
'float': float
}.get(option['type'], int)
for name in ['default_value', 'min', 'max']:
if option[name] == 'None': # This is Python None
option[name] = None
elif option[name]: # Skip empty entries
option[name] = cls(option[name])
return options
|
