Skip to content

[crmsh-5.0] Dev: ui_sbd: Add and check the minimum SBD timeout value #1945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: crmsh-5.0
Choose a base branch
from
Open

[crmsh-5.0] Dev: ui_sbd: Add and check the minimum SBD timeout value #1945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bc295e0
Dev: ui_sbd: Add and check the minimum SBD timeout value
liangxin1300 Oct 16, 2025
b502bad
Dev: unittests: Adjust unit test for previous commit
liangxin1300 Oct 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 22 additions & 8 deletions crmsh/ui_sbd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,10 +54,10 @@ def sbd_configure_completer(completed_list: typing.List[str]) -> typing.List[str
show_types, timeout_types = (), ()
if is_diskbased:
show_types = SBD.SHOW_TYPES
timeout_types = SBD.TIMEOUT_TYPES
timeout_types = tuple(SBD.TIMEOUT_TYPE_MINIMUMS.keys())
elif is_diskless:
show_types = SBD.DISKLESS_SHOW_TYPES
timeout_types = SBD.DISKLESS_TIMEOUT_TYPES
timeout_types = tuple(SBD.DISKLESS_TIMEOUT_TYPE_MINIMUMS.keys())

if completed_list[1] == "show":
if len(completed_list) == 3:
Expand Down Expand Up @@ -89,8 +89,19 @@ class SBD(command.UI):
- sbd purge
'''
name = "sbd"
TIMEOUT_TYPES = ("watchdog", "allocate", "loop", "msgwait", "crashdump-watchdog")
DISKLESS_TIMEOUT_TYPES = ("watchdog", "crashdump-watchdog")
TIMEOUT_TYPE_MINIMUMS = {
# Type: Minimum value
"watchdog": 5,
"allocate": 2,
"loop": 1,
"msgwait": 10,
"crashdump-watchdog": 1
}
DISKLESS_TIMEOUT_TYPE_MINIMUMS = {
# Type: Minimum value
"watchdog": 5,
"crashdump-watchdog": 1
}
SHOW_TYPES = ("disk_metadata", "sysconfig", "property")
DISKLESS_SHOW_TYPES = ("sysconfig", "property")
PCMK_ATTRS = (
Expand Down Expand Up @@ -167,9 +178,9 @@ def configure_usage(self) -> str:
'''
timeout_types, show_types = (), ()
if sbd.SBDUtils.is_using_disk_based_sbd():
timeout_types, show_types = self.TIMEOUT_TYPES, self.SHOW_TYPES
timeout_types, show_types = tuple(self.TIMEOUT_TYPE_MINIMUMS.keys()), self.SHOW_TYPES
elif sbd.SBDUtils.is_using_diskless_sbd():
timeout_types, show_types = self.DISKLESS_TIMEOUT_TYPES, self.DISKLESS_SHOW_TYPES
timeout_types, show_types = tuple(self.DISKLESS_TIMEOUT_TYPE_MINIMUMS.keys()), self.DISKLESS_SHOW_TYPES
else:
return ""

Expand Down Expand Up @@ -265,9 +276,12 @@ def _parse_args(self, args: tuple[str, ...]) -> dict[str, int|str]:
raise self.SyntaxError(f"Invalid argument: {arg}")
key, suffix, value = match.groups()
# timeout related parameters
if key in self.TIMEOUT_TYPES and suffix and suffix == "timeout":
if key in self.TIMEOUT_TYPE_MINIMUMS and suffix and suffix == "timeout":
if not value.isdigit():
raise self.SyntaxError(f"Invalid timeout value: {value}")
min_value = self.TIMEOUT_TYPE_MINIMUMS[key]
if int(value) < min_value:
raise ValueError(f"The minimum value for {key}-timeout is {min_value}")
parameter_dict[key] = int(value)
# watchdog device parameter
elif key == "watchdog" and suffix == "device":
Expand Down Expand Up @@ -397,7 +411,7 @@ def _configure_diskbase(self, parameter_dict: dict):

timeout_dict = {
k: v for k, v in parameter_dict.items()
if k in self.TIMEOUT_TYPES and k != "crashdump-watchdog"
if k in self.TIMEOUT_TYPE_MINIMUMS and k != "crashdump-watchdog"
}
timeout_dict = self._adjust_timeout_dict(timeout_dict)
# merge runtime timeout dict into parameter timeout dict without overwriting
Expand Down
4 changes: 2 additions & 2 deletions test/unittests/test_ui_sbd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ def test_configure_usage_none(self, mock_is_using_disk_based_sbd, mock_is_using_
@mock.patch('crmsh.sbd.SBDUtils.is_using_disk_based_sbd')
def test_configure_usage_disk_diskbased(self, mock_is_using_disk_based_sbd, mock_is_using_diskless_sbd):
mock_is_using_disk_based_sbd.return_value = True
timeout_usage_str = " ".join([f"[{t}-timeout=<integer>]" for t in ui_sbd.SBD.TIMEOUT_TYPES])
timeout_usage_str = " ".join([f"[{t}-timeout=<integer>]" for t in ui_sbd.SBD.TIMEOUT_TYPE_MINIMUMS])
show_usage = f"crm sbd configure show [{'|'.join(ui_sbd.SBD.SHOW_TYPES)}]"
expected = f"Usage:\n{show_usage}\ncrm sbd configure {timeout_usage_str} [watchdog-device=<device>]\n"
self.assertEqual(self.sbd_instance_diskbased.configure_usage, expected)
Expand All @@ -165,7 +165,7 @@ def test_configure_usage_disk_diskbased(self, mock_is_using_disk_based_sbd, mock
def test_configure_usage_disk_diskless(self, mock_is_using_disk_based_sbd, mock_is_using_diskless_sbd):
mock_is_using_disk_based_sbd.return_value = False
mock_is_using_diskless_sbd.return_value = True
timeout_usage_str = " ".join([f"[{t}-timeout=<integer>]" for t in ui_sbd.SBD.DISKLESS_TIMEOUT_TYPES])
timeout_usage_str = " ".join([f"[{t}-timeout=<integer>]" for t in ui_sbd.SBD.DISKLESS_TIMEOUT_TYPE_MINIMUMS])
show_usage = f"crm sbd configure show [{'|'.join(ui_sbd.SBD.DISKLESS_SHOW_TYPES)}]"
expected = f"Usage:\n{show_usage}\ncrm sbd configure {timeout_usage_str} [watchdog-device=<device>]\n"
self.assertEqual(self.sbd_instance_diskless.configure_usage, expected)
Expand Down