[ceph.git] / ceph / src / pybind / mgr / selftest / module.py


from mgr_module import MgrModule, CommandResult, HandleCommandResult, CLICommand, Option
import enum
import json
import random
import sys
import threading
from code import InteractiveInterpreter
from contextlib import redirect_stderr, redirect_stdout
from io import StringIO
from typing import Any, Dict, List, Optional, Tuple


# These workloads are things that can be requested to run inside the
# serve() function
class Workload(enum.Enum):
    COMMAND_SPAM = 'command_spam'
    THROW_EXCEPTION = 'throw_exception'
    SHUTDOWN = 'shutdown'


class Module(MgrModule):
    """
    This module is for testing the ceph-mgr python interface from within
    a running ceph-mgr daemon.

    It implements a sychronous self-test command for calling the functions
    in the MgrModule interface one by one, and a background "workload"
    command for causing the module to perform some thrashing-type
    activities in its serve() thread.
    """

    # The test code in qa/ relies on these options existing -- they
    # are of course not really used for anything in the module
    MODULE_OPTIONS = [
        Option(name='testkey'),
        Option(name='testlkey'),
        Option(name='testnewline'),
        Option(name='roption1'),
        Option(name='roption2',
               type='str',
               default='xyz'),
        Option(name='rwoption1'),
        Option(name='rwoption2',
               type='int'),
        Option(name='rwoption3',
               type='float'),
        Option(name='rwoption4',
               type='str'),
        Option(name='rwoption5',
               type='bool'),
        Option(name='rwoption6',
               type='bool',
               default=True),
        Option(name='rwoption7',
               type='int',
               min=1,
               max=42),
    ]

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super(Module, self).__init__(*args, **kwargs)
        self._event = threading.Event()
        self._workload: Optional[Workload] = None
        self._health: Dict[str, Dict[str, Any]] = {}
        self._repl = InteractiveInterpreter(dict(mgr=self))

    @CLICommand('mgr self-test python-version', perm='r')
    def python_version(self) -> Tuple[int, str, str]:
        '''
        Query the version of the embedded Python runtime
        '''
        major = sys.version_info.major
        minor = sys.version_info.minor
        micro = sys.version_info.micro
        return 0, f'{major}.{minor}.{micro}', ''

    @CLICommand('mgr self-test run')
    def run(self) -> Tuple[int, str, str]:
        '''
        Run mgr python interface tests
        '''
        self._self_test()
        return 0, '', 'Self-test succeeded'

    @CLICommand('mgr self-test background start')
    def backgroun_start(self, workload: Workload) -> Tuple[int, str, str]:
        '''
        Activate a background workload (one of command_spam, throw_exception)
        '''
        self._workload = workload
        self._event.set()
        return 0, '', 'Running `{0}` in background'.format(self._workload)

    @CLICommand('mgr self-test background stop')
    def background_stop(self) -> Tuple[int, str, str]:
        '''
        Stop background workload if any is running
        '''
        if self._workload:
            was_running = self._workload
            self._workload = None
            self._event.set()
            return 0, '', 'Stopping background workload `{0}`'.format(
                was_running)
        else:
            return 0, '', 'No background workload was running'

    @CLICommand('mgr self-test config get')
    def config_get(self, key: str) -> Tuple[int, str, str]:
        '''
        Peek at a configuration value
        '''
        return 0, str(self.get_module_option(key)), ''

    @CLICommand('mgr self-test config get_localized')
    def config_get_localized(self, key: str) -> Tuple[int, str, str]:
        '''
        Peek at a configuration value (localized variant)
        '''
        return 0, str(self.get_localized_module_option(key)), ''

    @CLICommand('mgr self-test remote')
    def test_remote(self) -> Tuple[int, str, str]:
        '''
        Test inter-module calls
        '''
        self._test_remote_calls()
        return 0, '', 'Successfully called'

    @CLICommand('mgr self-test module')
    def module(self, module: str) -> Tuple[int, str, str]:
        '''
        Run another module's self_test() method
        '''
        try:
            r = self.remote(module, "self_test")
        except RuntimeError as e:
            return -1, '', "Test failed: {0}".format(e)
        else:
            return 0, str(r), "Self-test OK"

    @CLICommand('mgr self-test cluster-log')
    def do_cluster_log(self,
                       channel: str,
                       priority: str,
                       message: str) -> Tuple[int, str, str]:
        '''
        Create an audit log record.
        '''
        priority_map = {
            'info': self.ClusterLogPrio.INFO,
            'security': self.ClusterLogPrio.SEC,
            'warning': self.ClusterLogPrio.WARN,
            'error': self.ClusterLogPrio.ERROR
        }
        self.cluster_log(channel,
                         priority_map[priority],
                         message)
        return 0, '', 'Successfully called'

    @CLICommand('mgr self-test health set')
    def health_set(self, checks: str) -> Tuple[int, str, str]:
        '''
        Set a health check from a JSON-formatted description.
        '''
        try:
            health_check = json.loads(checks)
        except Exception as e:
            return -1, "", "Failed to decode JSON input: {}".format(e)

        try:
            for check, info in health_check.items():
                self._health[check] = {
                    "severity": str(info["severity"]),
                    "summary": str(info["summary"]),
                    "count": 123,
                    "detail": [str(m) for m in info["detail"]]
                }
        except Exception as e:
            return -1, "", "Invalid health check format: {}".format(e)

        self.set_health_checks(self._health)
        return 0, "", ""

    @CLICommand('mgr self-test health clear')
    def health_clear(self, checks: Optional[List[str]] = None) -> Tuple[int, str, str]:
        '''
        Clear health checks by name. If no names provided, clear all.
        '''
        if checks is not None:
            for check in checks:
                if check in self._health:
                    del self._health[check]
        else:
            self._health = dict()

        self.set_health_checks(self._health)
        return 0, "", ""

    @CLICommand('mgr self-test insights_set_now_offset')
    def insights_set_now_offset(self, hours: int) -> Tuple[int, str, str]:
        '''
        Set the now time for the insights module.
        '''
        self.remote("insights", "testing_set_now_time_offset", hours)
        return 0, "", ""

    def _self_test(self) -> None:
        self.log.info("Running self-test procedure...")

        self._self_test_osdmap()
        self._self_test_getters()
        self._self_test_config()
        self._self_test_store()
        self._self_test_misc()
        self._self_test_perf_counters()

    def _self_test_getters(self) -> None:
        self.version
        self.get_context()
        self.get_mgr_id()

        # In this function, we will assume that the system is in a steady
        # state, i.e. if a server/service appears in one call, it will
        # not have gone by the time we call another function referring to it

        objects = [
            "fs_map",
            "osdmap_crush_map_text",
            "osd_map",
            "config",
            "mon_map",
            "service_map",
            "osd_metadata",
            "pg_summary",
            "pg_status",
            "pg_dump",
            "pg_ready",
            "df",
            "pg_stats",
            "pool_stats",
            "osd_stats",
            "osd_ping_times",
            "health",
            "mon_status",
            "mgr_map"
        ]
        for obj in objects:
            assert self.get(obj) is not None

        assert self.get("__OBJ_DNE__") is None

        servers = self.list_servers()
        for server in servers:
            self.get_server(server['hostname'])  # type: ignore

        osdmap = self.get('osd_map')
        for o in osdmap['osds']:
            osd_id = o['osd']
            self.get_metadata("osd", str(osd_id))

        self.get_daemon_status("osd", "0")

    def _self_test_config(self) -> None:
        # This is not a strong test (can't tell if values really
        # persisted), it's just for the python interface bit.

        self.set_module_option("testkey", "testvalue")
        assert self.get_module_option("testkey") == "testvalue"

        self.set_localized_module_option("testkey", "foo")
        assert self.get_localized_module_option("testkey") == "foo"

        # Must return the default value defined in MODULE_OPTIONS.
        value = self.get_localized_module_option("rwoption6")
        assert isinstance(value, bool)
        assert value is True

        # Use default value.
        assert self.get_module_option("roption1") is None
        assert self.get_module_option("roption1", "foobar") == "foobar"
        assert self.get_module_option("roption2") == "xyz"
        assert self.get_module_option("roption2", "foobar") == "xyz"

        # Option type is not defined => return as string.
        self.set_module_option("rwoption1", 8080)
        value = self.get_module_option("rwoption1")
        assert isinstance(value, str)
        assert value == "8080"

        # Option type is defined => return as integer.
        self.set_module_option("rwoption2", 10)
        value = self.get_module_option("rwoption2")
        assert isinstance(value, int)
        assert value == 10

        # Option type is defined => return as float.
        self.set_module_option("rwoption3", 1.5)
        value = self.get_module_option("rwoption3")
        assert isinstance(value, float)
        assert value == 1.5

        # Option type is defined => return as string.
        self.set_module_option("rwoption4", "foo")
        value = self.get_module_option("rwoption4")
        assert isinstance(value, str)
        assert value == "foo"

        # Option type is defined => return as bool.
        self.set_module_option("rwoption5", False)
        value = self.get_module_option("rwoption5")
        assert isinstance(value, bool)
        assert value is False

        # Option value range is specified
        try:
            self.set_module_option("rwoption7", 43)
        except Exception as e:
            assert isinstance(e, ValueError)
        else:
            message = "should raise if value is not in specified range"
            assert False, message

        # Specified module does not exist => return None.
        assert self.get_module_option_ex("foo", "bar") is None

        # Specified key does not exist => return None.
        assert self.get_module_option_ex("dashboard", "bar") is None

        self.set_module_option_ex("telemetry", "contact", "test@test.com")
        assert self.get_module_option_ex("telemetry", "contact") == "test@test.com"

        # No option default value, so use the specified one.
        assert self.get_module_option_ex("dashboard", "password") is None
        assert self.get_module_option_ex("dashboard", "password", "foobar") == "foobar"

        # Option type is not defined => return as string.
        self.set_module_option_ex("selftest", "rwoption1", 1234)
        value = self.get_module_option_ex("selftest", "rwoption1")
        assert isinstance(value, str)
        assert value == "1234"

        # Option type is defined => return as integer.
        self.set_module_option_ex("telemetry", "interval", 60)
        value = self.get_module_option_ex("telemetry", "interval")
        assert isinstance(value, int)
        assert value == 60

        # Option type is defined => return as bool.
        self.set_module_option_ex("telemetry", "leaderboard", True)
        value = self.get_module_option_ex("telemetry", "leaderboard")
        assert isinstance(value, bool)
        assert value is True

    def _self_test_store(self) -> None:
        existing_keys = set(self.get_store_prefix("test").keys())
        self.set_store("testkey", "testvalue")
        assert self.get_store("testkey") == "testvalue"

        assert (set(self.get_store_prefix("test").keys())
                == {"testkey"} | existing_keys)

    def _self_test_perf_counters(self) -> None:
        self.get_perf_schema("osd", "0")
        self.get_counter("osd", "0", "osd.op")
        # get_counter
        # get_all_perf_coutners

    def _self_test_misc(self) -> None:
        self.set_uri("http://this.is.a.test.com")
        self.set_health_checks({})

    def _self_test_osdmap(self) -> None:
        osdmap = self.get_osdmap()
        osdmap.get_epoch()
        osdmap.get_crush_version()
        osdmap.dump()

        inc = osdmap.new_incremental()
        osdmap.apply_incremental(inc)
        inc.get_epoch()
        inc.dump()

        crush = osdmap.get_crush()
        crush.dump()
        crush.get_item_name(-1)
        crush.get_item_weight(-1)
        crush.find_takes()
        crush.get_take_weight_osd_map(-1)

        # osdmap.get_pools_by_take()
        # osdmap.calc_pg_upmaps()
        # osdmap.map_pools_pgs_up()

        # inc.set_osd_reweights
        # inc.set_crush_compat_weight_set_weights

        self.log.info("Finished self-test procedure.")

    def _test_remote_calls(self) -> None:
        # Test making valid call
        self.remote("influx", "self_test")

        # Test calling module that exists but isn't enabled
        # (arbitrarily pick a non-always-on module to use)
        disabled_module = "telegraf"
        mgr_map = self.get("mgr_map")
        assert disabled_module not in mgr_map['modules']

        # (This works until the Z release in about 2027)
        latest_release = sorted(mgr_map['always_on_modules'].keys())[-1]
        assert disabled_module not in mgr_map['always_on_modules'][latest_release]

        try:
            self.remote(disabled_module, "handle_command", {"prefix": "influx self-test"})
        except ImportError:
            pass
        else:
            raise RuntimeError("ImportError not raised for disabled module")

        # Test calling module that doesn't exist
        try:
            self.remote("idontexist", "self_test")
        except ImportError:
            pass
        else:
            raise RuntimeError("ImportError not raised for nonexistent module")

        # Test calling method that doesn't exist
        try:
            self.remote("influx", "idontexist")
        except NameError:
            pass
        else:
            raise RuntimeError("KeyError not raised")

    def remote_from_orchestrator_cli_self_test(self, what: str) -> Any:
        import orchestrator
        if what == 'OrchestratorError':
            return orchestrator.OrchResult(result=None, exception=orchestrator.OrchestratorError('hello, world'))
        elif what == "ZeroDivisionError":
            return orchestrator.OrchResult(result=None, exception=ZeroDivisionError('hello, world'))
        assert False, repr(what)

    def shutdown(self) -> None:
        self._workload = Workload.SHUTDOWN
        self._event.set()

    def _command_spam(self) -> None:
        self.log.info("Starting command_spam workload...")
        while not self._event.is_set():
            osdmap = self.get_osdmap()
            dump = osdmap.dump()
            count = len(dump['osds'])
            i = int(random.random() * count)
            w = random.random()

            result = CommandResult('')
            self.send_command(result, 'mon', '', json.dumps({
                'prefix': 'osd reweight',
                'id': i,
                'weight': w}), '')

            _ = osdmap.get_crush().dump()
            r, outb, outs = result.wait()

        self._event.clear()
        self.log.info("Ended command_spam workload...")

    @CLICommand('mgr self-test eval')
    def eval(self,
             s: Optional[str] = None,
             inbuf: Optional[str] = None) -> HandleCommandResult:
        '''
        eval given source
        '''
        source = s or inbuf
        if source is None:
            return HandleCommandResult(-1, '', 'source is not specified')

        err = StringIO()
        out = StringIO()
        with redirect_stderr(err), redirect_stdout(out):
            needs_more = self._repl.runsource(source)
            if needs_more:
                retval = 2
                stdout = ''
                stderr = ''
            else:
                retval = 0
                stdout = out.getvalue()
                stderr = err.getvalue()
            return HandleCommandResult(retval, stdout, stderr)

    def serve(self) -> None:
        while True:
            if self._workload == Workload.COMMAND_SPAM:
                self._command_spam()
            elif self._workload == Workload.SHUTDOWN:
                self.log.info("Shutting down...")
                break
            elif self._workload == Workload.THROW_EXCEPTION:
                raise RuntimeError("Synthetic exception in serve")
            else:
                self.log.info("Waiting for workload request...")
                self._event.wait()
                self._event.clear()
Commit	Line	Data
3efd9988	1
20effc67 TL	2	from mgr_module import MgrModule, CommandResult, HandleCommandResult, CLICommand, Option
20effc67 TL	3	import enum
a4b75251 TL	4	import json
	5	import random
	6	import sys
	7	import threading
20effc67 TL	8	from code import InteractiveInterpreter
	9	from contextlib import redirect_stderr, redirect_stdout
	10	from io import StringIO
	11	from typing import Any, Dict, List, Optional, Tuple
	12
	13
	14	# These workloads are things that can be requested to run inside the
	15	# serve() function
	16	class Workload(enum.Enum):
	17	COMMAND_SPAM = 'command_spam'
	18	THROW_EXCEPTION = 'throw_exception'
	19	SHUTDOWN = 'shutdown'
3efd9988 FG	20
	21
	22	class Module(MgrModule):
	23	"""
	24	This module is for testing the ceph-mgr python interface from within
	25	a running ceph-mgr daemon.
	26
	27	It implements a sychronous self-test command for calling the functions
	28	in the MgrModule interface one by one, and a background "workload"
	29	command for causing the module to perform some thrashing-type
	30	activities in its serve() thread.
	31	"""
	32
11fdf7f2 TL	33	# The test code in qa/ relies on these options existing -- they
	34	# are of course not really used for anything in the module
	35	MODULE_OPTIONS = [
20effc67 TL	36	Option(name='testkey'),
	37	Option(name='testlkey'),
	38	Option(name='testnewline'),
	39	Option(name='roption1'),
	40	Option(name='roption2',
	41	type='str',
	42	default='xyz'),
	43	Option(name='rwoption1'),
	44	Option(name='rwoption2',
	45	type='int'),
	46	Option(name='rwoption3',
	47	type='float'),
	48	Option(name='rwoption4',
	49	type='str'),
	50	Option(name='rwoption5',
	51	type='bool'),
	52	Option(name='rwoption6',
	53	type='bool',
	54	default=True),
	55	Option(name='rwoption7',
	56	type='int',
	57	min=1,
	58	max=42),
11fdf7f2	59	]
3efd9988	60
20effc67	61	def __init__(self, args: Any, *kwargs: Any) -> None:
3efd9988 FG	62	super(Module, self).__init__(args, *kwargs)
3efd9988 FG	63	self._event = threading.Event()
20effc67 TL	64	self._workload: Optional[Workload] = None
	65	self._health: Dict[str, Dict[str, Any]] = {}
	66	self._repl = InteractiveInterpreter(dict(mgr=self))
	67
	68	@CLICommand('mgr self-test python-version', perm='r')
	69	def python_version(self) -> Tuple[int, str, str]:
	70	'''
	71	Query the version of the embedded Python runtime
	72	'''
	73	major = sys.version_info.major
	74	minor = sys.version_info.minor
	75	micro = sys.version_info.micro
	76	return 0, f'{major}.{minor}.{micro}', ''
	77
	78	@CLICommand('mgr self-test run')
	79	def run(self) -> Tuple[int, str, str]:
	80	'''
	81	Run mgr python interface tests
	82	'''
	83	self._self_test()
	84	return 0, '', 'Self-test succeeded'
	85
	86	@CLICommand('mgr self-test background start')
	87	def backgroun_start(self, workload: Workload) -> Tuple[int, str, str]:
	88	'''
	89	Activate a background workload (one of command_spam, throw_exception)
	90	'''
	91	self._workload = workload
	92	self._event.set()
	93	return 0, '', 'Running `{0}` in background'.format(self._workload)
	94
	95	@CLICommand('mgr self-test background stop')
	96	def background_stop(self) -> Tuple[int, str, str]:
	97	'''
	98	Stop background workload if any is running
	99	'''
	100	if self._workload:
	101	was_running = self._workload
	102	self._workload = None
3efd9988	103	self._event.set()
20effc67 TL	104	return 0, '', 'Stopping background workload `{0}`'.format(
20effc67 TL	105	was_running)
3efd9988	106	else:
20effc67 TL	107	return 0, '', 'No background workload was running'
	108
	109	@CLICommand('mgr self-test config get')
	110	def config_get(self, key: str) -> Tuple[int, str, str]:
	111	'''
	112	Peek at a configuration value
	113	'''
	114	return 0, str(self.get_module_option(key)), ''
	115
	116	@CLICommand('mgr self-test config get_localized')
	117	def config_get_localized(self, key: str) -> Tuple[int, str, str]:
	118	'''
	119	Peek at a configuration value (localized variant)
	120	'''
	121	return 0, str(self.get_localized_module_option(key)), ''
	122
	123	@CLICommand('mgr self-test remote')
	124	def test_remote(self) -> Tuple[int, str, str]:
	125	'''
	126	Test inter-module calls
	127	'''
	128	self._test_remote_calls()
	129	return 0, '', 'Successfully called'
	130
	131	@CLICommand('mgr self-test module')
	132	def module(self, module: str) -> Tuple[int, str, str]:
	133	'''
	134	Run another module's self_test() method
	135	'''
	136	try:
	137	r = self.remote(module, "self_test")
	138	except RuntimeError as e:
	139	return -1, '', "Test failed: {0}".format(e)
	140	else:
	141	return 0, str(r), "Self-test OK"
	142
	143	@CLICommand('mgr self-test cluster-log')
	144	def do_cluster_log(self,
	145	channel: str,
	146	priority: str,
	147	message: str) -> Tuple[int, str, str]:
	148	'''
	149	Create an audit log record.
	150	'''
	151	priority_map = {
	152	'info': self.ClusterLogPrio.INFO,
	153	'security': self.ClusterLogPrio.SEC,
	154	'warning': self.ClusterLogPrio.WARN,
	155	'error': self.ClusterLogPrio.ERROR
	156	}
	157	self.cluster_log(channel,
	158	priority_map[priority],
	159	message)
	160	return 0, '', 'Successfully called'
	161
	162	@CLICommand('mgr self-test health set')
	163	def health_set(self, checks: str) -> Tuple[int, str, str]:
	164	'''
	165	Set a health check from a JSON-formatted description.
	166	'''
11fdf7f2	167	try:
20effc67	168	health_check = json.loads(checks)
11fdf7f2	169	except Exception as e:
494da23a	170	return -1, "", "Failed to decode JSON input: {}".format(e)
11fdf7f2 TL	171
11fdf7f2 TL	172	try:
20effc67	173	for check, info in health_check.items():
11fdf7f2 TL	174	self._health[check] = {
	175	"severity": str(info["severity"]),
	176	"summary": str(info["summary"]),
9f95a23c	177	"count": 123,
11fdf7f2 TL	178	"detail": [str(m) for m in info["detail"]]
	179	}
	180	except Exception as e:
494da23a	181	return -1, "", "Invalid health check format: {}".format(e)
11fdf7f2 TL	182
	183	self.set_health_checks(self._health)
	184	return 0, "", ""
	185
20effc67 TL	186	@CLICommand('mgr self-test health clear')
	187	def health_clear(self, checks: Optional[List[str]] = None) -> Tuple[int, str, str]:
	188	'''
	189	Clear health checks by name. If no names provided, clear all.
	190	'''
	191	if checks is not None:
	192	for check in checks:
11fdf7f2 TL	193	if check in self._health:
	194	del self._health[check]
	195	else:
	196	self._health = dict()
	197
	198	self.set_health_checks(self._health)
	199	return 0, "", ""
	200
20effc67 TL	201	@CLICommand('mgr self-test insights_set_now_offset')
	202	def insights_set_now_offset(self, hours: int) -> Tuple[int, str, str]:
	203	'''
	204	Set the now time for the insights module.
	205	'''
11fdf7f2 TL	206	self.remote("insights", "testing_set_now_time_offset", hours)
	207	return 0, "", ""
	208
20effc67	209	def _self_test(self) -> None:
3efd9988 FG	210	self.log.info("Running self-test procedure...")
	211
	212	self._self_test_osdmap()
	213	self._self_test_getters()
	214	self._self_test_config()
11fdf7f2	215	self._self_test_store()
3efd9988 FG	216	self._self_test_misc()
	217	self._self_test_perf_counters()
	218
20effc67	219	def _self_test_getters(self) -> None:
3efd9988 FG	220	self.version
	221	self.get_context()
	222	self.get_mgr_id()
	223
	224	# In this function, we will assume that the system is in a steady
	225	# state, i.e. if a server/service appears in one call, it will
	226	# not have gone by the time we call another function referring to it
	227
	228	objects = [
20effc67 TL	229	"fs_map",
	230	"osdmap_crush_map_text",
	231	"osd_map",
	232	"config",
	233	"mon_map",
	234	"service_map",
	235	"osd_metadata",
	236	"pg_summary",
	237	"pg_status",
	238	"pg_dump",
	239	"pg_ready",
	240	"df",
	241	"pg_stats",
	242	"pool_stats",
	243	"osd_stats",
	244	"osd_ping_times",
	245	"health",
	246	"mon_status",
	247	"mgr_map"
	248	]
3efd9988	249	for obj in objects:
11fdf7f2 TL	250	assert self.get(obj) is not None
	251
	252	assert self.get("__OBJ_DNE__") is None
3efd9988 FG	253
	254	servers = self.list_servers()
	255	for server in servers:
20effc67	256	self.get_server(server['hostname']) # type: ignore
3efd9988 FG	257
	258	osdmap = self.get('osd_map')
	259	for o in osdmap['osds']:
	260	osd_id = o['osd']
	261	self.get_metadata("osd", str(osd_id))
	262
	263	self.get_daemon_status("osd", "0")
3efd9988	264
20effc67	265	def _self_test_config(self) -> None:
3efd9988 FG	266	# This is not a strong test (can't tell if values really
	267	# persisted), it's just for the python interface bit.
	268
11fdf7f2 TL	269	self.set_module_option("testkey", "testvalue")
	270	assert self.get_module_option("testkey") == "testvalue"
	271
	272	self.set_localized_module_option("testkey", "foo")
	273	assert self.get_localized_module_option("testkey") == "foo"
	274
	275	# Must return the default value defined in MODULE_OPTIONS.
	276	value = self.get_localized_module_option("rwoption6")
	277	assert isinstance(value, bool)
	278	assert value is True
	279
	280	# Use default value.
	281	assert self.get_module_option("roption1") is None
	282	assert self.get_module_option("roption1", "foobar") == "foobar"
	283	assert self.get_module_option("roption2") == "xyz"
	284	assert self.get_module_option("roption2", "foobar") == "xyz"
	285
	286	# Option type is not defined => return as string.
	287	self.set_module_option("rwoption1", 8080)
	288	value = self.get_module_option("rwoption1")
	289	assert isinstance(value, str)
	290	assert value == "8080"
	291
	292	# Option type is defined => return as integer.
	293	self.set_module_option("rwoption2", 10)
	294	value = self.get_module_option("rwoption2")
	295	assert isinstance(value, int)
	296	assert value == 10
	297
	298	# Option type is defined => return as float.
	299	self.set_module_option("rwoption3", 1.5)
	300	value = self.get_module_option("rwoption3")
	301	assert isinstance(value, float)
	302	assert value == 1.5
	303
	304	# Option type is defined => return as string.
	305	self.set_module_option("rwoption4", "foo")
	306	value = self.get_module_option("rwoption4")
	307	assert isinstance(value, str)
	308	assert value == "foo"
	309
	310	# Option type is defined => return as bool.
	311	self.set_module_option("rwoption5", False)
	312	value = self.get_module_option("rwoption5")
	313	assert isinstance(value, bool)
	314	assert value is False
	315
20effc67 TL	316	# Option value range is specified
	317	try:
	318	self.set_module_option("rwoption7", 43)
	319	except Exception as e:
	320	assert isinstance(e, ValueError)
	321	else:
	322	message = "should raise if value is not in specified range"
	323	assert False, message
	324
11fdf7f2 TL	325	# Specified module does not exist => return None.
	326	assert self.get_module_option_ex("foo", "bar") is None
	327
	328	# Specified key does not exist => return None.
	329	assert self.get_module_option_ex("dashboard", "bar") is None
	330
	331	self.set_module_option_ex("telemetry", "contact", "test@test.com")
	332	assert self.get_module_option_ex("telemetry", "contact") == "test@test.com"
	333
	334	# No option default value, so use the specified one.
	335	assert self.get_module_option_ex("dashboard", "password") is None
	336	assert self.get_module_option_ex("dashboard", "password", "foobar") == "foobar"
	337
	338	# Option type is not defined => return as string.
	339	self.set_module_option_ex("selftest", "rwoption1", 1234)
	340	value = self.get_module_option_ex("selftest", "rwoption1")
	341	assert isinstance(value, str)
	342	assert value == "1234"
	343
	344	# Option type is defined => return as integer.
	345	self.set_module_option_ex("telemetry", "interval", 60)
	346	value = self.get_module_option_ex("telemetry", "interval")
	347	assert isinstance(value, int)
	348	assert value == 60
	349
	350	# Option type is defined => return as bool.
	351	self.set_module_option_ex("telemetry", "leaderboard", True)
	352	value = self.get_module_option_ex("telemetry", "leaderboard")
	353	assert isinstance(value, bool)
	354	assert value is True
	355
20effc67	356	def _self_test_store(self) -> None:
11fdf7f2 TL	357	existing_keys = set(self.get_store_prefix("test").keys())
	358	self.set_store("testkey", "testvalue")
	359	assert self.get_store("testkey") == "testvalue"
	360
20effc67 TL	361	assert (set(self.get_store_prefix("test").keys())
20effc67 TL	362	== {"testkey"} \| existing_keys)
3efd9988	363
20effc67	364	def _self_test_perf_counters(self) -> None:
3efd9988 FG	365	self.get_perf_schema("osd", "0")
3efd9988 FG	366	self.get_counter("osd", "0", "osd.op")
20effc67 TL	367	# get_counter
20effc67 TL	368	# get_all_perf_coutners
3efd9988	369
20effc67	370	def _self_test_misc(self) -> None:
3efd9988 FG	371	self.set_uri("http://this.is.a.test.com")
	372	self.set_health_checks({})
	373
20effc67	374	def _self_test_osdmap(self) -> None:
3efd9988 FG	375	osdmap = self.get_osdmap()
	376	osdmap.get_epoch()
	377	osdmap.get_crush_version()
	378	osdmap.dump()
	379
	380	inc = osdmap.new_incremental()
	381	osdmap.apply_incremental(inc)
	382	inc.get_epoch()
	383	inc.dump()
	384
	385	crush = osdmap.get_crush()
	386	crush.dump()
	387	crush.get_item_name(-1)
	388	crush.get_item_weight(-1)
	389	crush.find_takes()
	390	crush.get_take_weight_osd_map(-1)
	391
20effc67 TL	392	# osdmap.get_pools_by_take()
	393	# osdmap.calc_pg_upmaps()
	394	# osdmap.map_pools_pgs_up()
3efd9988	395
20effc67 TL	396	# inc.set_osd_reweights
20effc67 TL	397	# inc.set_crush_compat_weight_set_weights
3efd9988 FG	398
	399	self.log.info("Finished self-test procedure.")
	400
20effc67	401	def _test_remote_calls(self) -> None:
11fdf7f2	402	# Test making valid call
20effc67	403	self.remote("influx", "self_test")
11fdf7f2 TL	404
	405	# Test calling module that exists but isn't enabled
	406	# (arbitrarily pick a non-always-on module to use)
	407	disabled_module = "telegraf"
	408	mgr_map = self.get("mgr_map")
	409	assert disabled_module not in mgr_map['modules']
	410
	411	# (This works until the Z release in about 2027)
	412	latest_release = sorted(mgr_map['always_on_modules'].keys())[-1]
	413	assert disabled_module not in mgr_map['always_on_modules'][latest_release]
	414
	415	try:
	416	self.remote(disabled_module, "handle_command", {"prefix": "influx self-test"})
	417	except ImportError:
	418	pass
	419	else:
	420	raise RuntimeError("ImportError not raised for disabled module")
	421
	422	# Test calling module that doesn't exist
	423	try:
20effc67	424	self.remote("idontexist", "self_test")
11fdf7f2 TL	425	except ImportError:
	426	pass
	427	else:
	428	raise RuntimeError("ImportError not raised for nonexistent module")
	429
	430	# Test calling method that doesn't exist
	431	try:
20effc67	432	self.remote("influx", "idontexist")
11fdf7f2 TL	433	except NameError:
	434	pass
	435	else:
	436	raise RuntimeError("KeyError not raised")
	437
20effc67	438	def remote_from_orchestrator_cli_self_test(self, what: str) -> Any:
9f95a23c TL	439	import orchestrator
9f95a23c TL	440	if what == 'OrchestratorError':
20effc67	441	return orchestrator.OrchResult(result=None, exception=orchestrator.OrchestratorError('hello, world'))
9f95a23c	442	elif what == "ZeroDivisionError":
20effc67	443	return orchestrator.OrchResult(result=None, exception=ZeroDivisionError('hello, world'))
9f95a23c	444	assert False, repr(what)
11fdf7f2	445
20effc67 TL	446	def shutdown(self) -> None:
20effc67 TL	447	self._workload = Workload.SHUTDOWN
3efd9988 FG	448	self._event.set()
3efd9988 FG	449
20effc67	450	def _command_spam(self) -> None:
3efd9988 FG	451	self.log.info("Starting command_spam workload...")
	452	while not self._event.is_set():
	453	osdmap = self.get_osdmap()
	454	dump = osdmap.dump()
	455	count = len(dump['osds'])
	456	i = int(random.random() * count)
	457	w = random.random()
	458
	459	result = CommandResult('')
	460	self.send_command(result, 'mon', '', json.dumps({
	461	'prefix': 'osd reweight',
	462	'id': i,
20effc67	463	'weight': w}), '')
3efd9988	464
20effc67	465	_ = osdmap.get_crush().dump()
3efd9988 FG	466	r, outb, outs = result.wait()
	467
	468	self._event.clear()
	469	self.log.info("Ended command_spam workload...")
	470
20effc67 TL	471	@CLICommand('mgr self-test eval')
	472	def eval(self,
	473	s: Optional[str] = None,
	474	inbuf: Optional[str] = None) -> HandleCommandResult:
	475	'''
	476	eval given source
	477	'''
	478	source = s or inbuf
	479	if source is None:
	480	return HandleCommandResult(-1, '', 'source is not specified')
	481
	482	err = StringIO()
	483	out = StringIO()
	484	with redirect_stderr(err), redirect_stdout(out):
	485	needs_more = self._repl.runsource(source)
	486	if needs_more:
	487	retval = 2
	488	stdout = ''
	489	stderr = ''
	490	else:
	491	retval = 0
	492	stdout = out.getvalue()
	493	stderr = err.getvalue()
	494	return HandleCommandResult(retval, stdout, stderr)
	495
	496	def serve(self) -> None:
3efd9988	497	while True:
20effc67	498	if self._workload == Workload.COMMAND_SPAM:
3efd9988	499	self._command_spam()
20effc67	500	elif self._workload == Workload.SHUTDOWN:
3efd9988 FG	501	self.log.info("Shutting down...")
3efd9988 FG	502	break
20effc67	503	elif self._workload == Workload.THROW_EXCEPTION:
11fdf7f2	504	raise RuntimeError("Synthetic exception in serve")
3efd9988 FG	505	else:
	506	self.log.info("Waiting for workload request...")
	507	self._event.wait()
	508	self._event.clear()