From 1b0b32e44cddacbd7acf7024181247f4251c92f9 Mon Sep 17 00:00:00 2001 From: Kefu Chai Date: Mon, 28 Jun 2021 11:32:41 +0800 Subject: [PATCH] pybind/mgr/pg_autoscaler: add typing annotations Signed-off-by: Kefu Chai --- src/mypy.ini | 3 + src/pybind/mgr/pg_autoscaler/module.py | 185 +++++++++++++++---------- src/pybind/mgr/tox.ini | 1 + 3 files changed, 118 insertions(+), 71 deletions(-) diff --git a/src/mypy.ini b/src/mypy.ini index b689279ab0a..191dde3b4f3 100755 --- a/src/mypy.ini +++ b/src/mypy.ini @@ -67,6 +67,9 @@ disallow_untyped_defs = True [mypy-prometheus.*] disallow_untyped_defs = True +[mypy-pg_autoscaler.*] +disallow_untyped_defs = True + [mypy-rbd_support.*] disallow_untyped_defs = True diff --git a/src/pybind/mgr/pg_autoscaler/module.py b/src/pybind/mgr/pg_autoscaler/module.py index 2649f68e52e..9d0c41cca21 100644 --- a/src/pybind/mgr/pg_autoscaler/module.py +++ b/src/pybind/mgr/pg_autoscaler/module.py @@ -5,10 +5,10 @@ Automatically scale pg_num based on how much data is stored in each pool. import json import mgr_util import threading -from typing import Optional, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple, TYPE_CHECKING, Union import uuid from prettytable import PrettyTable -from mgr_module import CLIReadCommand, CLIWriteCommand, MgrModule, Option +from mgr_module import HealthChecksT, CLIReadCommand, CLIWriteCommand, CRUSHMap, MgrModule, Option, OSDMap """ Some terminology is made up for the purposes of this module: @@ -24,7 +24,17 @@ INTERVAL = 5 PG_NUM_MIN = 32 # unless specified on a per-pool basis -def nearest_power_of_two(n): +if TYPE_CHECKING: + import sys + if sys.version_info >= (3, 8): + from typing import Literal + else: + from typing_extensions import Literal + + ScaleModeT = Literal['scale-up', 'scale-down'] + + +def nearest_power_of_two(n: int) -> int: v = int(n) v -= 1 @@ -42,7 +52,11 @@ def nearest_power_of_two(n): return x if (v - n) > (n - x) else v -def effective_target_ratio(target_ratio, total_target_ratio, total_target_bytes, capacity): + +def effective_target_ratio(target_ratio: float, + total_target_ratio: float, + total_target_bytes: int, + capacity: int) -> float: """ Returns the target ratio after normalizing for ratios across pools and adjusting for capacity reserved by pools that have target_size_bytes set. @@ -57,20 +71,21 @@ def effective_target_ratio(target_ratio, total_target_ratio, total_target_bytes, return target_ratio + class PgAdjustmentProgress(object): """ Keeps the initial and target pg_num values """ - def __init__(self, pool_id, pg_num, pg_num_target): + def __init__(self, pool_id: int, pg_num: int, pg_num_target: int) -> None: self.ev_id = str(uuid.uuid4()) self.pool_id = pool_id self.reset(pg_num, pg_num_target) - def reset(self, pg_num, pg_num_target): + def reset(self, pg_num: int, pg_num_target: int) -> None: self.pg_num = pg_num self.pg_num_target = pg_num_target - def update(self, module, progress): + def update(self, module: MgrModule, progress: float) -> None: desc = 'increasing' if self.pg_num < self.pg_num_target else 'decreasing' module.remote('progress', 'update', self.ev_id, ev_msg="PG autoscaler %s pool %d PGs from %d to %d" % @@ -80,17 +95,17 @@ class PgAdjustmentProgress(object): class CrushSubtreeResourceStatus: - def __init__(self): - self.root_ids = [] - self.osds = set() - self.osd_count = None # Number of OSDs - self.pg_target = None # Ideal full-capacity PG count? + def __init__(self) -> None: + self.root_ids: List[int] = [] + self.osds: Set[int] = set() + self.osd_count: Optional[int] = None # Number of OSDs + self.pg_target: Optional[int] = None # Ideal full-capacity PG count? self.pg_current = 0 # How many PGs already? self.pg_left = 0 - self.capacity = None # Total capacity of OSDs in subtree - self.pool_ids = [] - self.pool_names = [] - self.pool_count = None + self.capacity: Optional[int] = None # Total capacity of OSDs in subtree + self.pool_ids: List[int] = [] + self.pool_names: List[str] = [] + self.pool_count: Optional[int] = None self.pool_used = 0 self.total_target_ratio = 0.0 self.total_target_bytes = 0 # including replication / EC overhead @@ -123,16 +138,20 @@ class PgAutoscaler(MgrModule): runtime=True), ] - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: super(PgAutoscaler, self).__init__(*args, **kwargs) self._shutdown = threading.Event() - self._event = {} + self._event: Dict[int, PgAdjustmentProgress] = {} # So much of what we do peeks at the osdmap that it's easiest # to just keep a copy of the pythonized version. self._osd_map = None + if TYPE_CHECKING: + self.autoscale_profile: 'ScaleModeT' = 'scale-up' + self.sleep_interval = 60 + self.mon_target_pg_per_osd = 0 - def config_notify(self): + def config_notify(self) -> None: for opt in self.NATIVE_OPTIONS: setattr(self, opt, @@ -243,32 +262,38 @@ class PgAutoscaler(MgrModule): self.set_module_option("autoscale_profile", "scale-down") return 0, "", "autoscale-profile is now scale-down" - def serve(self): + def serve(self) -> None: self.config_notify() while not self._shutdown.is_set(): self._maybe_adjust() self._update_progress_events() self._shutdown.wait(timeout=self.sleep_interval) - def shutdown(self): + def shutdown(self) -> None: self.log.info('Stopping pg_autoscaler') self._shutdown.set() - def get_subtree_resource_status(self, osdmap, crush): + def get_subtree_resource_status(self, + osdmap: OSDMap, + crush: CRUSHMap) -> Tuple[Dict[int, CrushSubtreeResourceStatus], + Dict[int, int]]: """ For each CRUSH subtree of interest (i.e. the roots under which we have pools), calculate the current resource usages and targets, such as how many PGs there are, vs. how many PGs we would like there to be. """ - result = {} + result: Dict[int, CrushSubtreeResourceStatus] = {} pool_root = {} roots = [] # identify subtrees (note that they may overlap!) for pool_id, pool in osdmap.get_pools().items(): - cr_name = crush.get_rule_by_id(pool['crush_rule'])['rule_name'] - root_id = int(crush.get_rule_root(cr_name)) + crush_rule = crush.get_rule_by_id(pool['crush_rule']) + assert crush_rule is not None + cr_name = crush_rule['rule_name'] + root_id = crush.get_rule_root(cr_name) + assert root_id is not None pool_root[pool_id] = root_id osds = set(crush.get_osds_under(root_id)) @@ -298,11 +323,12 @@ class PgAutoscaler(MgrModule): # finish subtrees all_stats = self.get('osd_stats') for s in roots: + assert s.osds is not None s.osd_count = len(s.osds) s.pg_target = s.osd_count * self.mon_target_pg_per_osd s.pg_left = s.pg_target s.pool_count = len(s.pool_ids) - capacity = 0.0 + capacity = 0 for osd_stats in all_stats['osd_stats']: if osd_stats['osd'] in s.osds: # Intentionally do not apply the OSD's reweight to @@ -323,16 +349,16 @@ class PgAutoscaler(MgrModule): def _calc_final_pg_target( self, - p, - pool_name, - root_map, - root_id, - capacity_ratio, - even_pools, - bias, - is_used, - profile, - ): + p: Dict[str, Any], + pool_name: str, + root_map: Dict[int, CrushSubtreeResourceStatus], + root_id: int, + capacity_ratio: float, + even_pools: Dict[str, Dict[str, Any]], + bias: float, + is_used: bool, + profile: 'ScaleModeT', + ) -> Union[Tuple[float, int, int], Tuple[None, None, None]]: """ `profile` determines behaviour of the autoscaler. `is_used` flag used to determine if this is the first @@ -344,13 +370,17 @@ class PgAutoscaler(MgrModule): if profile == "scale-up": final_ratio = capacity_ratio # So what proportion of pg allowance should we be using? - pool_pg_target = (final_ratio * root_map[root_id].pg_target) / p['size'] * bias + pg_target = root_map[root_id].pg_target + assert pg_target is not None + pool_pg_target = (final_ratio * pg_target) / p['size'] * bias final_pg_target = max(p.get('options', {}).get('pg_num_min', PG_NUM_MIN), nearest_power_of_two(pool_pg_target)) else: if is_used: - even_ratio = 1 / root_map[root_id].pool_count + pool_count = root_map[root_id].pool_count + assert pool_count is not None + even_ratio = 1 / pool_count used_ratio = capacity_ratio if used_ratio > even_ratio: @@ -362,12 +392,16 @@ class PgAutoscaler(MgrModule): return None, None, None final_ratio = max(used_ratio, even_ratio) - used_pg = final_ratio * root_map[root_id].pg_target - root_map[root_id].pg_left -= used_pg + pg_target = root_map[root_id].pg_target + assert pg_target is not None + used_pg = final_ratio * pg_target + root_map[root_id].pg_left -= int(used_pg) pool_pg_target = used_pg / p['size'] * bias else: - final_ratio = 1 / (root_map[root_id].pool_count - root_map[root_id].pool_used) + pool_count = root_map[root_id].pool_count + assert pool_count is not None + final_ratio = 1 / (pool_count - root_map[root_id].pool_used) pool_pg_target = (final_ratio * root_map[root_id].pg_left) / p['size'] * bias final_pg_target = max(p.get('options', {}).get('pg_num_min', PG_NUM_MIN), @@ -388,17 +422,17 @@ class PgAutoscaler(MgrModule): def _calc_pool_targets( self, - osdmap, - pools, - crush_map, - root_map, - pool_root, - pool_stats, - ret, - threshold, - is_used, - profile, - ): + osdmap: OSDMap, + pools: Dict[str, Dict[str, Any]], + crush_map: CRUSHMap, + root_map: Dict[int, CrushSubtreeResourceStatus], + pool_root: Dict[int, int], + pool_stats: Dict[int, Dict[str, int]], + ret: List[Dict[str, Any]], + threshold: float, + is_used: bool, + profile: 'ScaleModeT', + ) -> Tuple[List[Dict[str, Any]], Dict[str, Dict[str, Any]]]: """ Calculates final_pg_target of each pools and determine if it needs scaling, this depends on the profile of the autoscaler. For scale-down, @@ -406,7 +440,7 @@ class PgAutoscaler(MgrModule): pools needs more pgs due to increased usage. For scale-up, we start out with the minimal amount of pgs and only scale when there is increase in usage. """ - even_pools = {} + even_pools: Dict[str, Dict[str, Any]] = {} for pool_name, p in pools.items(): pool_id = p['pool'] if pool_id not in pool_stats: @@ -415,10 +449,14 @@ class PgAutoscaler(MgrModule): # FIXME: we assume there is only one take per pool, but that # may not be true. - cr_name = crush_map.get_rule_by_id(p['crush_rule'])['rule_name'] - root_id = int(crush_map.get_rule_root(cr_name)) + crush_rule = crush_map.get_rule_by_id(p['crush_rule']) + assert crush_rule is not None + cr_name = crush_rule['rule_name'] + root_id = crush_map.get_rule_root(cr_name) + assert root_id is not None capacity = root_map[root_id].capacity + assert capacity is not None if capacity == 0: self.log.debug('skipping empty subtree %s', cr_name) continue @@ -456,7 +494,7 @@ class PgAutoscaler(MgrModule): profile, ) - if final_ratio == None: + if final_ratio is None: continue adjust = False @@ -466,6 +504,7 @@ class PgAutoscaler(MgrModule): final_ratio <= 1.0: adjust = True + assert pool_pg_target is not None ret.append({ 'pool_id': pool_id, 'pool_name': p['pool_name'], @@ -492,11 +531,13 @@ class PgAutoscaler(MgrModule): def _get_pool_status( self, - osdmap, - pools, - profile, - threshold=3.0, - ): + osdmap: OSDMap, + pools: Dict[str, Dict[str, Any]], + profile: 'ScaleModeT', + threshold: float = 3.0, + ) -> Tuple[List[Dict[str, Any]], + Dict[int, CrushSubtreeResourceStatus], + Dict[int, int]]: assert threshold >= 2.0 crush_map = osdmap.get_crush() @@ -504,7 +545,7 @@ class PgAutoscaler(MgrModule): df = self.get('df') pool_stats = dict([(p['id'], p['stats']) for p in df['pools']]) - ret = [] + ret: List[Dict[str, Any]] = [] # Iterate over all pools to determine how they should be sized. # First call of _calc_pool_targets() is to find/adjust pools that uses more capacaity than @@ -523,7 +564,7 @@ class PgAutoscaler(MgrModule): return (ret, root_map, pool_root) - def _update_progress_events(self): + def _update_progress_events(self) -> None: osdmap = self.get_osdmap() pools = osdmap.get_pools() for pool_id in list(self._event): @@ -536,7 +577,7 @@ class PgAutoscaler(MgrModule): continue ev.update(self, (ev.pg_num - pool_data['pg_num']) / (ev.pg_num - ev.pg_num_target)) - def _maybe_adjust(self): + def _maybe_adjust(self) -> None: self.log.info('_maybe_adjust') osdmap = self.get_osdmap() if osdmap.get_require_osd_release() < 'nautilus': @@ -550,11 +591,11 @@ class PgAutoscaler(MgrModule): too_few = [] too_many = [] bytes_and_ratio = [] - health_checks = {} + health_checks: Dict[str, Dict[str, Union[int, str, List[str]]]] = {} total_bytes = dict([(r, 0) for r in iter(root_map)]) total_target_bytes = dict([(r, 0.0) for r in iter(root_map)]) - target_bytes_pools = dict([(r, []) for r in iter(root_map)]) + target_bytes_pools: Dict[int, List[int]] = dict([(r, []) for r in iter(root_map)]) for p in ps: pool_id = p['pool_id'] @@ -630,23 +671,25 @@ class PgAutoscaler(MgrModule): too_much_target_bytes = [] for root_id, total in total_bytes.items(): - total_target = total_target_bytes[root_id] - if total_target > 0 and total > root_map[root_id].capacity and root_map[root_id].capacity: + total_target = int(total_target_bytes[root_id]) + capacity = root_map[root_id].capacity + assert capacity is not None + if total_target > 0 and total > capacity and capacity: too_much_target_bytes.append( 'Pools %s overcommit available storage by %.03fx due to ' 'target_size_bytes %s on pools %s' % ( root_map[root_id].pool_names, - total / root_map[root_id].capacity, + total / capacity, mgr_util.format_bytes(total_target, 5, colored=False), target_bytes_pools[root_id] ) ) - elif total_target > root_map[root_id].capacity and root_map[root_id].capacity: + elif total_target > capacity and capacity: too_much_target_bytes.append( 'Pools %s overcommit available storage by %.03fx due to ' 'collective target_size_bytes of %s' % ( root_map[root_id].pool_names, - total / root_map[root_id].capacity, + total / capacity, mgr_util.format_bytes(total_target, 5, colored=False), ) ) diff --git a/src/pybind/mgr/tox.ini b/src/pybind/mgr/tox.ini index 660062d2fd6..b47dec31558 100644 --- a/src/pybind/mgr/tox.ini +++ b/src/pybind/mgr/tox.ini @@ -88,6 +88,7 @@ commands = -m mirroring \ -m nfs \ -m orchestrator \ + -m pg_autoscaler \ -m progress \ -m prometheus \ -m rbd_support \ -- 2.39.5