from .qos_conf import QOSType, QOSParams, QOSBandwidthControl, QOSOpsControl, QOS
-def export_dict_bw_checks(cluster_id: str,
- mgr_obj: Any,
- qos_enable: bool,
- qos_dict: dict) -> None:
- enable_bw_ctrl = qos_dict.get('enable_bw_control')
- combined_bw_ctrl = qos_dict.get('combined_rw_bw_control')
- bandwith_param_exists = any(key.endswith('bw') for key in qos_dict)
+def export_dict_bw_checks(
+ cluster_id: str, mgr_obj: Any, qos_enable: bool, qos_dict: dict
+) -> None:
+ enable_bw_ctrl = qos_dict.get("enable_bw_control")
+ combined_bw_ctrl = qos_dict.get("combined_rw_bw_control")
+ bandwith_param_exists = any(key.endswith("bw") for key in qos_dict)
if enable_bw_ctrl is None:
if combined_bw_ctrl and bandwith_param_exists:
- raise Exception('Bandwidth control is not enabled but associated parameters exists')
+ raise Exception(
+ "Bandwidth control is not enabled but associated parameters exists"
+ )
return
if combined_bw_ctrl is None:
combined_bw_ctrl = False
if not qos_enable and enable_bw_ctrl:
- raise Exception('To enable bandwidth control, qos_enable and enable_bw_control should be true.')
+ raise Exception(
+ "To enable bandwidth control, qos_enable and enable_bw_control should be true."
+ )
if not (isinstance(enable_bw_ctrl, bool) and isinstance(combined_bw_ctrl, bool)):
- raise Exception('Invalid values for the enable_bw_ctrl and combined_bw_ctrl parameters.')
+ raise Exception(
+ "Invalid values for the enable_bw_ctrl and combined_bw_ctrl parameters."
+ )
# if qos bandwidth control is disabled, then bandwidths should not be set and no need to bandwidth checks
if not enable_bw_ctrl:
if bandwith_param_exists:
- raise Exception('Bandwidths should not be passed when enable_bw_control is false.')
+ raise Exception(
+ "Bandwidths should not be passed when enable_bw_control is false."
+ )
return
if enable_bw_ctrl and not bandwith_param_exists:
- raise Exception('Bandwidths should be set when enable_bw_control is true.')
- bw_obj = QOSBandwidthControl(enable_bw_ctrl,
- combined_bw_ctrl,
- export_writebw=qos_dict.get(QOSParams.export_writebw.value, '0'),
- export_readbw=qos_dict.get(QOSParams.export_readbw.value, '0'),
- client_writebw=qos_dict.get(QOSParams.client_writebw.value, '0'),
- client_readbw=qos_dict.get(QOSParams.client_readbw.value, '0'),
- export_rw_bw=qos_dict.get(QOSParams.export_rw_bw.value, '0'),
- client_rw_bw=qos_dict.get(QOSParams.client_rw_bw.value, '0'))
+ raise Exception("Bandwidths should be set when enable_bw_control is true.")
+ bw_obj = QOSBandwidthControl(
+ enable_bw_ctrl,
+ combined_bw_ctrl,
+ export_writebw=qos_dict.get(QOSParams.export_writebw.value, "0"),
+ export_readbw=qos_dict.get(QOSParams.export_readbw.value, "0"),
+ client_writebw=qos_dict.get(QOSParams.client_writebw.value, "0"),
+ client_readbw=qos_dict.get(QOSParams.client_readbw.value, "0"),
+ export_rw_bw=qos_dict.get(QOSParams.export_rw_bw.value, "0"),
+ client_rw_bw=qos_dict.get(QOSParams.client_rw_bw.value, "0"),
+ )
export_qos_bw_checks(cluster_id, mgr_obj, bw_obj)
-def export_dict_ops_checks(cluster_id: str,
- mgr_obj: Any,
- qos_enable: bool,
- qos_dict: dict) -> None:
+def export_dict_ops_checks(
+ cluster_id: str, mgr_obj: Any, qos_enable: bool, qos_dict: dict
+) -> None:
enable_iops_ctrl = qos_dict.get(QOSParams.enable_iops_ctrl.value)
if enable_iops_ctrl is None:
return
if not isinstance(enable_iops_ctrl, bool):
- raise Exception(f'Invalid values for the {QOSParams.enable_iops_ctrl.value} parameter')
- ops_param_exists = any(key.endswith('iops') for key in qos_dict)
+ raise Exception(
+ f"Invalid values for the {QOSParams.enable_iops_ctrl.value} parameter"
+ )
+ ops_param_exists = any(key.endswith("iops") for key in qos_dict)
if not enable_iops_ctrl:
if ops_param_exists:
- raise Exception(f'IOPS count parameters should not be passed when {QOSParams.enable_iops_ctrl.value} is false.')
+ raise Exception(
+ f"IOPS count parameters should not be passed when {QOSParams.enable_iops_ctrl.value} is false."
+ )
return
if enable_iops_ctrl and not ops_param_exists:
- raise Exception(f'IOPS count parameters should be set when {QOSParams.enable_iops_ctrl.value} is true.')
- ops_obj = QOSOpsControl(enable_iops_ctrl,
- max_export_iops=qos_dict.get(QOSParams.max_export_iops.value, 0),
- max_client_iops=qos_dict.get(QOSParams.max_client_iops.value, 0))
+ raise Exception(
+ f"IOPS count parameters should be set when {QOSParams.enable_iops_ctrl.value} is true."
+ )
+ ops_obj = QOSOpsControl(
+ enable_iops_ctrl,
+ max_export_iops=qos_dict.get(QOSParams.max_export_iops.value, 0),
+ max_client_iops=qos_dict.get(QOSParams.max_client_iops.value, 0),
+ )
export_qos_ops_checks(cluster_id, mgr_obj, ops_obj)
-def export_dict_qos_bw_ops_checks(cluster_id: str,
- mgr_obj: Any,
- qos_dict: dict,
- old_qos_block: dict = {}) -> None:
+def export_dict_qos_bw_ops_checks(
+ cluster_id: str, mgr_obj: Any, qos_dict: dict, old_qos_block: dict = {}
+) -> None:
"""Validate the qos block of dict passed to apply_export method"""
- qos_enable = qos_dict.get('enable_qos')
+ qos_enable = qos_dict.get("enable_qos")
if qos_enable is None:
- raise Exception('The QoS block requires at least the enable_qos parameter')
+ raise Exception("The QoS block requires at least the enable_qos parameter")
if not isinstance(qos_enable, bool):
- raise Exception('Invalid value for the enable_qos parameter')
+ raise Exception("Invalid value for the enable_qos parameter")
# if cluster level bandwidth or ops control is disabled or qos type changed to PerClient
# but old qos block still has those values we should accept it in apply command
validate_bw = True
if not clust_qos_obj.enable_qos or clust_qos_obj.qos_type == QOSType.PerClient:
if old_qos_block == qos_dict:
return
- if clust_qos_obj.enable_qos and clust_qos_obj.bw_obj and not clust_qos_obj.bw_obj.enable_bw_ctrl:
- keys = [QOSParams.export_writebw.value, QOSParams.export_readbw.value,
- QOSParams.client_writebw.value, QOSParams.client_readbw.value,
- QOSParams.export_rw_bw.value, QOSParams.client_rw_bw.value,
- QOSParams.enable_bw_ctrl.value, QOSParams.combined_bw_ctrl.value]
+ if (
+ clust_qos_obj.enable_qos
+ and clust_qos_obj.bw_obj
+ and not clust_qos_obj.bw_obj.enable_bw_ctrl
+ ):
+ keys = [
+ QOSParams.export_writebw.value,
+ QOSParams.export_readbw.value,
+ QOSParams.client_writebw.value,
+ QOSParams.client_readbw.value,
+ QOSParams.export_rw_bw.value,
+ QOSParams.client_rw_bw.value,
+ QOSParams.enable_bw_ctrl.value,
+ QOSParams.combined_bw_ctrl.value,
+ ]
if all(old_qos_block.get(key) == qos_dict.get(key) for key in keys):
validate_bw = False
- if clust_qos_obj.enable_qos and clust_qos_obj.ops_obj and not clust_qos_obj.ops_obj.enable_iops_ctrl:
- keys = [QOSParams.max_export_iops.value, QOSParams.max_client_iops.value,
- QOSParams.enable_iops_ctrl.value]
+ if (
+ clust_qos_obj.enable_qos
+ and clust_qos_obj.ops_obj
+ and not clust_qos_obj.ops_obj.enable_iops_ctrl
+ ):
+ keys = [
+ QOSParams.max_export_iops.value,
+ QOSParams.max_client_iops.value,
+ QOSParams.enable_iops_ctrl.value,
+ ]
if all(old_qos_block.get(key) == qos_dict.get(key) for key in keys):
validate_ops = False
if validate_bw:
export_dict_ops_checks(cluster_id, mgr_obj, qos_enable, qos_dict)
-def export_qos_bw_checks(cluster_id: str,
- mgr_obj: Any,
- bw_obj: QOSBandwidthControl,
- nfs_clust_obj: Any = None) -> None:
+def export_qos_bw_checks(
+ cluster_id: str,
+ mgr_obj: Any,
+ bw_obj: QOSBandwidthControl,
+ nfs_clust_obj: Any = None,
+) -> None:
"""check cluster level qos bandwidth control is enabled to enable export level qos
bandwidth control and validate bandwidths"""
if not nfs_clust_obj:
nfs_clust_obj = NFSCluster(mgr_obj)
clust_qos_obj = nfs_clust_obj.get_cluster_qos_config(cluster_id)
- if not clust_qos_obj or (clust_qos_obj and not (clust_qos_obj.enable_qos
- and clust_qos_obj.bw_obj
- and clust_qos_obj.bw_obj.enable_bw_ctrl)):
- raise Exception(f'To configure bandwidth control for export, you must first enable bandwidth control at the cluster level for {cluster_id}.')
+ if not clust_qos_obj or (
+ clust_qos_obj
+ and not (
+ clust_qos_obj.enable_qos
+ and clust_qos_obj.bw_obj
+ and clust_qos_obj.bw_obj.enable_bw_ctrl
+ )
+ ):
+ raise Exception(
+ f"To configure bandwidth control for export, you must first enable bandwidth control at the cluster level for {cluster_id}."
+ )
if clust_qos_obj.qos_type:
if clust_qos_obj.qos_type == QOSType.PerClient:
- raise Exception(f'Export-level QoS bandwidth control cannot be enabled if the QoS type at the cluster {cluster_id} level is set to PerClient.')
+ raise Exception(
+ f"Export-level QoS bandwidth control cannot be enabled if the QoS type at the cluster {cluster_id} level is set to PerClient."
+ )
bw_obj.qos_bandwidth_checks(clust_qos_obj.qos_type)
-def export_qos_ops_checks(cluster_id: str,
- mgr_obj: Any,
- ops_obj: QOSOpsControl,
- nfs_clust_obj: Any = None) -> None:
+def export_qos_ops_checks(
+ cluster_id: str, mgr_obj: Any, ops_obj: QOSOpsControl, nfs_clust_obj: Any = None
+) -> None:
"""check cluster level qos IOPS is enabled to enable export level qos IOPS and validate IOPS count"""
if not nfs_clust_obj:
nfs_clust_obj = NFSCluster(mgr_obj)
clust_qos_obj = nfs_clust_obj.get_cluster_qos_config(cluster_id)
- if not clust_qos_obj or (clust_qos_obj and not (clust_qos_obj.enable_qos
- and clust_qos_obj.ops_obj
- and clust_qos_obj.ops_obj.enable_iops_ctrl)):
- raise Exception(f'To configure IOPS control for export, you must first enable IOPS control at the cluster level {cluster_id}.')
+ if not clust_qos_obj or (
+ clust_qos_obj
+ and not (
+ clust_qos_obj.enable_qos
+ and clust_qos_obj.ops_obj
+ and clust_qos_obj.ops_obj.enable_iops_ctrl
+ )
+ ):
+ raise Exception(
+ f"To configure IOPS control for export, you must first enable IOPS control at the cluster level {cluster_id}."
+ )
if clust_qos_obj.qos_type:
if clust_qos_obj.qos_type == QOSType.PerClient:
- raise Exception(f'Export-level QoS IOPS control cannot be enabled if the QoS type at the cluster {cluster_id} level is set to PerClient.')
+ raise Exception(
+ f"Export-level QoS IOPS control cannot be enabled if the QoS type at the cluster {cluster_id} level is set to PerClient."
+ )
ops_obj.qos_ops_checks(clust_qos_obj.qos_type)
-from typing import List, Dict, Any, Optional
+from typing import Dict, Any, Optional
from enum import Enum
from ceph.utils import bytes_to_human, with_units_to_int
-
-
-class RawBlock():
- def __init__(self, block_name: str, blocks: List['RawBlock'] = [], values: Dict[str, Any] = {}):
- if not values: # workaround mutable default argument
- values = {}
- if not blocks: # workaround mutable default argument
- blocks = []
- self.block_name = block_name
- self.blocks = blocks
- self.values = values
-
- def __eq__(self, other: Any) -> bool:
- if not isinstance(other, RawBlock):
- return False
- return self.block_name == other.block_name and \
- self.blocks == other.blocks and \
- self.values == other.values
-
- def __repr__(self) -> str:
- return f'RawBlock({self.block_name!r}, {self.blocks!r}, {self.values!r})'
+from .ganesha_raw_conf import RawBlock
class QOSParams(Enum):
class UserQoSType(Enum):
- per_share = 'PerShare'
- per_client = 'PerClient'
- per_share_per_client = 'PerShare_PerClient'
+ per_share = "PerShare"
+ per_client = "PerClient"
+ per_share_per_client = "PerShare_PerClient"
class QOSType(Enum):
max_bw = 4000000000 # 4GB
bw_bytes = with_units_to_int(bandwidth)
if bw_bytes != 0 and (bw_bytes < min_bw or bw_bytes > max_bw):
- raise Exception(f"Provided bandwidth value is not in range, Please enter a value between {min_bw} (1MB) and {max_bw} (4GB) bytes per second.")
+ raise Exception(
+ f"Provided bandwidth value is not in range, Please enter a value between {min_bw} (1MB) and {max_bw} (4GB) bytes per second."
+ )
return bw_bytes
min_cnt = 10
max_cnt = 16384
if count != 0 and (count < min_cnt or count > max_cnt):
- raise Exception(f"Provided IOS count value is not in range, Please enter a value between {min_cnt} and {max_cnt} bytes per second.")
+ raise Exception(
+ f"Provided IOS count value is not in range, Please enter a value between {min_cnt} and {max_cnt} bytes per second."
+ )
return count
QOS_REQ_BW_PARAMS = {
- 'combined_bw_disabled': {
- 'PerShare': ['max_export_write_bw', 'max_export_read_bw'],
- 'PerClient': ['max_client_write_bw', 'max_client_read_bw'],
- 'PerShare_PerClient': ['max_export_write_bw', 'max_export_read_bw', 'max_client_write_bw', 'max_client_read_bw']
+ "combined_bw_disabled": {
+ "PerShare": ["max_export_write_bw", "max_export_read_bw"],
+ "PerClient": ["max_client_write_bw", "max_client_read_bw"],
+ "PerShare_PerClient": [
+ "max_export_write_bw",
+ "max_export_read_bw",
+ "max_client_write_bw",
+ "max_client_read_bw",
+ ],
+ },
+ "combined_bw_enabled": {
+ "PerShare": ["max_export_combined_bw"],
+ "PerClient": ["max_client_combined_bw"],
+ "PerShare_PerClient": ["max_export_combined_bw", "max_client_combined_bw"],
},
- 'combined_bw_enabled': {
- 'PerShare': ['max_export_combined_bw'],
- 'PerClient': ['max_client_combined_bw'],
- 'PerShare_PerClient': ['max_export_combined_bw', 'max_client_combined_bw'],
- }
}
QOS_REQ_OPS_PARAMS = {
- 'PerShare': ['max_export_iops'],
- 'PerClient': ['max_client_iops'],
- 'PerShare_PerClient': ['max_export_iops', 'max_client_iops']
+ "PerShare": ["max_export_iops"],
+ "PerClient": ["max_client_iops"],
+ "PerShare_PerClient": ["max_export_iops", "max_client_iops"],
}
class QOSBandwidthControl(object):
- def __init__(self,
- enable_bw_ctrl: bool = False,
- combined_bw_ctrl: bool = False,
- export_writebw: str = '0',
- export_readbw: str = '0',
- client_writebw: str = '0',
- client_readbw: str = '0',
- export_rw_bw: str = '0',
- client_rw_bw: str = '0'
- ) -> None:
+ def __init__(
+ self,
+ enable_bw_ctrl: bool = False,
+ combined_bw_ctrl: bool = False,
+ export_writebw: str = "0",
+ export_readbw: str = "0",
+ client_writebw: str = "0",
+ client_readbw: str = "0",
+ export_rw_bw: str = "0",
+ client_rw_bw: str = "0",
+ ) -> None:
self.enable_bw_ctrl = enable_bw_ctrl
self.combined_bw_ctrl = combined_bw_ctrl
try:
raise Exception(f"Invalid bandwidth value. {e}")
@classmethod
- def from_dict(cls, qos_dict: Dict[str, Any]) -> 'QOSBandwidthControl':
+ def from_dict(cls, qos_dict: Dict[str, Any]) -> "QOSBandwidthControl":
# qos dict has bandwidths in human readable format(str)
bw_kwargs = {
- 'enable_bw_ctrl': qos_dict.get(QOSParams.enable_bw_ctrl.value, False),
- 'combined_bw_ctrl': qos_dict.get(QOSParams.combined_bw_ctrl.value, False),
- 'export_writebw': qos_dict.get(QOSParams.export_writebw.value, '0'),
- 'export_readbw': qos_dict.get(QOSParams.export_readbw.value, '0'),
- 'client_writebw': qos_dict.get(QOSParams.client_writebw.value, '0'),
- 'client_readbw': qos_dict.get(QOSParams.client_readbw.value, '0'),
- 'export_rw_bw': qos_dict.get(QOSParams.export_rw_bw.value, '0'),
- 'client_rw_bw': qos_dict.get(QOSParams.client_rw_bw.value, '0')
+ "enable_bw_ctrl": qos_dict.get(QOSParams.enable_bw_ctrl.value, False),
+ "combined_bw_ctrl": qos_dict.get(QOSParams.combined_bw_ctrl.value, False),
+ "export_writebw": qos_dict.get(QOSParams.export_writebw.value, "0"),
+ "export_readbw": qos_dict.get(QOSParams.export_readbw.value, "0"),
+ "client_writebw": qos_dict.get(QOSParams.client_writebw.value, "0"),
+ "client_readbw": qos_dict.get(QOSParams.client_readbw.value, "0"),
+ "export_rw_bw": qos_dict.get(QOSParams.export_rw_bw.value, "0"),
+ "client_rw_bw": qos_dict.get(QOSParams.client_rw_bw.value, "0"),
}
return cls(**bw_kwargs)
@classmethod
- def from_qos_block(cls, qos_block: RawBlock) -> 'QOSBandwidthControl':
+ def from_qos_block(cls, qos_block: RawBlock) -> "QOSBandwidthControl":
# qos block has bandwidths in bytes(int)
bw_kwargs = {
- 'enable_bw_ctrl': qos_block.values.get(QOSParams.enable_bw_ctrl.value, False),
- 'combined_bw_ctrl': qos_block.values.get(QOSParams.combined_bw_ctrl.value, False),
- 'export_writebw': str(qos_block.values.get(QOSParams.export_writebw.value, 0)),
- 'export_readbw': str(qos_block.values.get(QOSParams.export_readbw.value, 0)),
- 'client_writebw': str(qos_block.values.get(QOSParams.client_writebw.value, 0)),
- 'client_readbw': str(qos_block.values.get(QOSParams.client_readbw.value, 0)),
- 'export_rw_bw': str(qos_block.values.get(QOSParams.export_rw_bw.value, 0)),
- 'client_rw_bw': str(qos_block.values.get(QOSParams.client_rw_bw.value, 0))
+ "enable_bw_ctrl": qos_block.values.get(
+ QOSParams.enable_bw_ctrl.value, False
+ ),
+ "combined_bw_ctrl": qos_block.values.get(
+ QOSParams.combined_bw_ctrl.value, False
+ ),
+ "export_writebw": str(
+ qos_block.values.get(QOSParams.export_writebw.value, 0)
+ ),
+ "export_readbw": str(
+ qos_block.values.get(QOSParams.export_readbw.value, 0)
+ ),
+ "client_writebw": str(
+ qos_block.values.get(QOSParams.client_writebw.value, 0)
+ ),
+ "client_readbw": str(
+ qos_block.values.get(QOSParams.client_readbw.value, 0)
+ ),
+ "export_rw_bw": str(qos_block.values.get(QOSParams.export_rw_bw.value, 0)),
+ "client_rw_bw": str(qos_block.values.get(QOSParams.client_rw_bw.value, 0)),
}
return cls(**bw_kwargs)
def to_qos_block(self) -> RawBlock:
- result = RawBlock('qos_bandwidths_control')
+ result = RawBlock("qos_bandwidths_control")
result.values[QOSParams.enable_bw_ctrl.value] = self.enable_bw_ctrl
result.values[QOSParams.combined_bw_ctrl.value] = self.combined_bw_ctrl
if self.export_writebw:
r[QOSParams.enable_bw_ctrl.value] = self.enable_bw_ctrl
r[QOSParams.combined_bw_ctrl.value] = self.combined_bw_ctrl
if self.export_writebw:
- r[QOSParams.export_writebw.value] = self.bw_for_to_dict(self.export_writebw, ret_bw_in_bytes)
+ r[QOSParams.export_writebw.value] = self.bw_for_to_dict(
+ self.export_writebw, ret_bw_in_bytes
+ )
if self.export_readbw:
- r[QOSParams.export_readbw.value] = self.bw_for_to_dict(self.export_readbw, ret_bw_in_bytes)
+ r[QOSParams.export_readbw.value] = self.bw_for_to_dict(
+ self.export_readbw, ret_bw_in_bytes
+ )
if self.client_writebw:
- r[QOSParams.client_writebw.value] = self.bw_for_to_dict(self.client_writebw, ret_bw_in_bytes)
+ r[QOSParams.client_writebw.value] = self.bw_for_to_dict(
+ self.client_writebw, ret_bw_in_bytes
+ )
if self.client_readbw:
- r[QOSParams.client_readbw.value] = self.bw_for_to_dict(self.client_readbw, ret_bw_in_bytes)
+ r[QOSParams.client_readbw.value] = self.bw_for_to_dict(
+ self.client_readbw, ret_bw_in_bytes
+ )
if self.export_rw_bw:
- r[QOSParams.export_rw_bw.value] = self.bw_for_to_dict(self.export_rw_bw, ret_bw_in_bytes)
+ r[QOSParams.export_rw_bw.value] = self.bw_for_to_dict(
+ self.export_rw_bw, ret_bw_in_bytes
+ )
if self.client_rw_bw:
- r[QOSParams.client_rw_bw.value] = self.bw_for_to_dict(self.client_rw_bw, ret_bw_in_bytes)
+ r[QOSParams.client_rw_bw.value] = self.bw_for_to_dict(
+ self.client_rw_bw, ret_bw_in_bytes
+ )
return r
def qos_bandwidth_checks(self, qos_type: QOSType) -> None:
params = {}
d = vars(self)
for key in d:
- if key.endswith('bw'):
+ if key.endswith("bw"):
params[QOSParams[key].value] = d[key]
if not self.combined_bw_ctrl:
- req_params = QOS_REQ_BW_PARAMS['combined_bw_disabled'][qos_type.name]
+ req_params = QOS_REQ_BW_PARAMS["combined_bw_disabled"][qos_type.name]
else:
- req_params = QOS_REQ_BW_PARAMS['combined_bw_enabled'][qos_type.name]
+ req_params = QOS_REQ_BW_PARAMS["combined_bw_enabled"][qos_type.name]
allowed_params = []
not_allowed_params = []
for key in params:
elif key not in req_params and params[key] != 0:
not_allowed_params.append(key)
if allowed_params or not_allowed_params:
- raise Exception(f"When combined_rw_bw is {'enabled' if self.combined_bw_ctrl else 'disabled'} "
- f"and qos_type is {qos_type.name}, "
- f"{'attribute ' + ', '.join(allowed_params) + ' required' if allowed_params else ''} "
- f"{'attribute ' + ', '.join(not_allowed_params) + ' are not allowed' if not_allowed_params else ''}.")
+ raise Exception(
+ f"When combined_rw_bw is {'enabled' if self.combined_bw_ctrl else 'disabled'} "
+ f"and qos_type is {qos_type.name}, "
+ f"{'attribute ' + ', '.join(allowed_params) + ' required' if allowed_params else ''} "
+ f"{'attribute ' + ', '.join(not_allowed_params) + ' are not allowed' if not_allowed_params else ''}."
+ )
class QOSOpsControl(object):
- def __init__(self,
- enable_iops_ctrl: bool = False,
- max_export_iops: int = 0,
- max_client_iops: int = 0
- ) -> None:
+ def __init__(
+ self,
+ enable_iops_ctrl: bool = False,
+ max_export_iops: int = 0,
+ max_client_iops: int = 0,
+ ) -> None:
self.enable_iops_ctrl = enable_iops_ctrl
self.max_export_iops = _validate_qos_ops(max_export_iops)
self.max_client_iops = _validate_qos_ops(max_client_iops)
@classmethod
- def from_dict(cls, qos_dict: Dict[str, Any]) -> 'QOSOpsControl':
+ def from_dict(cls, qos_dict: Dict[str, Any]) -> "QOSOpsControl":
kwargs: dict[str, Any] = {}
- kwargs['enable_iops_ctrl'] = qos_dict.get(QOSParams.enable_iops_ctrl.value, False)
- kwargs['max_export_iops'] = qos_dict.get(QOSParams.max_export_iops.value, 0)
- kwargs['max_client_iops'] = qos_dict.get(QOSParams.max_client_iops.value, 0)
+ kwargs["enable_iops_ctrl"] = qos_dict.get(
+ QOSParams.enable_iops_ctrl.value, False
+ )
+ kwargs["max_export_iops"] = qos_dict.get(QOSParams.max_export_iops.value, 0)
+ kwargs["max_client_iops"] = qos_dict.get(QOSParams.max_client_iops.value, 0)
return cls(**kwargs)
@classmethod
- def from_qos_block(cls, qos_block: RawBlock) -> 'QOSOpsControl':
+ def from_qos_block(cls, qos_block: RawBlock) -> "QOSOpsControl":
kwargs: dict[str, Any] = {}
- kwargs['enable_iops_ctrl'] = qos_block.values.get(QOSParams.enable_iops_ctrl.value, False)
- kwargs['max_export_iops'] = qos_block.values.get(QOSParams.max_export_iops.value, 0)
- kwargs['max_client_iops'] = qos_block.values.get(QOSParams.max_client_iops.value, 0)
+ kwargs["enable_iops_ctrl"] = qos_block.values.get(
+ QOSParams.enable_iops_ctrl.value, False
+ )
+ kwargs["max_export_iops"] = qos_block.values.get(
+ QOSParams.max_export_iops.value, 0
+ )
+ kwargs["max_client_iops"] = qos_block.values.get(
+ QOSParams.max_client_iops.value, 0
+ )
return cls(**kwargs)
def to_qos_block(self) -> RawBlock:
- result = RawBlock('qos_ops_control')
+ result = RawBlock("qos_ops_control")
result.values[QOSParams.enable_iops_ctrl.value] = self.enable_iops_ctrl
if self.max_export_iops:
result.values[QOSParams.max_export_iops.value] = self.max_export_iops
params = {}
d = vars(self)
for key in d:
- if key.endswith('iops'):
+ if key.endswith("iops"):
params[QOSParams[key].value] = d[key]
req_params = QOS_REQ_OPS_PARAMS[qos_type.name]
allowed_params = []
elif key not in req_params and params[key] != 0:
not_allowed_params.append(key)
if allowed_params or not_allowed_params:
- raise Exception(f"When qos_type is {qos_type.name}, "
- f"{'attribute ' + ', '.join(allowed_params) + ' required' if allowed_params else ''} "
- f"{'attribute ' + ', '.join(not_allowed_params) + ' are not allowed' if not_allowed_params else ''}.")
+ raise Exception(
+ f"When qos_type is {qos_type.name}, "
+ f"{'attribute ' + ', '.join(allowed_params) + ' required' if allowed_params else ''} "
+ f"{'attribute ' + ', '.join(not_allowed_params) + ' are not allowed' if not_allowed_params else ''}."
+ )
class QOS(object):
- def __init__(self,
- cluster_op: bool = False,
- enable_qos: bool = False,
- qos_type: Optional[QOSType] = None,
- bw_obj: Optional[QOSBandwidthControl] = None,
- ops_obj: Optional[QOSOpsControl] = None
- ) -> None:
+ def __init__(
+ self,
+ cluster_op: bool = False,
+ enable_qos: bool = False,
+ qos_type: Optional[QOSType] = None,
+ bw_obj: Optional[QOSBandwidthControl] = None,
+ ops_obj: Optional[QOSOpsControl] = None,
+ ) -> None:
self.cluster_op = cluster_op
self.enable_qos = enable_qos
self.qos_type = qos_type
self.ops_obj = ops_obj
@classmethod
- def from_dict(cls, qos_dict: Dict[str, Any], cluster_op: bool = False) -> 'QOS':
+ def from_dict(cls, qos_dict: Dict[str, Any], cluster_op: bool = False) -> "QOS":
kwargs: dict[str, Any] = {}
# qos dict will have qos type as enum name
if cluster_op:
qos_type = qos_dict.get(QOSParams.qos_type.value)
if qos_type:
- kwargs['qos_type'] = QOSType[qos_type]
- kwargs['enable_qos'] = qos_dict.get(QOSParams.enable_qos.value)
- kwargs['bw_obj'] = QOSBandwidthControl.from_dict(qos_dict)
- kwargs['ops_obj'] = QOSOpsControl.from_dict(qos_dict)
+ kwargs["qos_type"] = QOSType[qos_type]
+ kwargs["enable_qos"] = qos_dict.get(QOSParams.enable_qos.value)
+ kwargs["bw_obj"] = QOSBandwidthControl.from_dict(qos_dict)
+ kwargs["ops_obj"] = QOSOpsControl.from_dict(qos_dict)
return cls(cluster_op, **kwargs)
@classmethod
- def from_qos_block(cls, qos_block: RawBlock, cluster_op: bool = False) -> 'QOS':
+ def from_qos_block(cls, qos_block: RawBlock, cluster_op: bool = False) -> "QOS":
kwargs: dict[str, Any] = {}
# qos block will have qos type as enum value
if cluster_op:
qos_type = qos_block.values.get(QOSParams.qos_type.value)
if qos_type:
- kwargs['qos_type'] = QOSType(qos_type)
- kwargs['enable_qos'] = qos_block.values.get(QOSParams.enable_qos.value)
- kwargs['bw_obj'] = QOSBandwidthControl.from_qos_block(qos_block)
- kwargs['ops_obj'] = QOSOpsControl.from_qos_block(qos_block)
+ kwargs["qos_type"] = QOSType(qos_type)
+ kwargs["enable_qos"] = qos_block.values.get(QOSParams.enable_qos.value)
+ kwargs["bw_obj"] = QOSBandwidthControl.from_qos_block(qos_block)
+ kwargs["ops_obj"] = QOSOpsControl.from_qos_block(qos_block)
return cls(cluster_op, **kwargs)
def to_qos_block(self) -> RawBlock:
r.update(res)
return r
- def get_enable_qos_val(self, disable_bw: bool = False, disable_ops: bool = False) -> bool:
+ def get_enable_qos_val(
+ self, disable_bw: bool = False, disable_ops: bool = False
+ ) -> bool:
if not (self.enable_qos) or not (disable_bw or disable_ops):
return False
# check if ops control is enabled
projects / ceph.git / commitdiff