Skip to content

Commit d3e1551

Browse files
committed
feat(core): add DRBD helpers (#11)
Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>
1 parent d57f698 commit d3e1551

9 files changed

Lines changed: 675 additions & 0 deletions

File tree

‎src/xcp_storage/backends/__init__.py‎

Whitespace-only changes.
Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
# Copyright (C) 2026 Vates SAS
2+
#
3+
# This program is free software: you can redistribute it and/or modify
4+
# it under the terms of the GNU General Public License as published by
5+
# the Free Software Foundation, either version 3 of the License, or
6+
# (at your option) any later version.
7+
# This program is distributed in the hope that it will be useful,
8+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10+
# GNU General Public License for more details.
11+
#
12+
# You should have received a copy of the GNU General Public License
13+
# along with this program. If not, see <https://www.gnu.org/licenses/>.
14+
15+
import contextlib
16+
from dataclasses import dataclass
17+
import json
18+
from pathlib import Path
19+
import re
20+
21+
import xcp_storage.log as log
22+
from xcp_storage.utils.process import (
23+
get_process_cmdline,
24+
run_command,
25+
)
26+
27+
from xcp_storage.typing import (
28+
Any,
29+
Dict,
30+
Final,
31+
Iterator,
32+
List,
33+
)
34+
35+
# ==============================================================================
36+
37+
logger = log.get_logger() # Use default logger.
38+
39+
# ------------------------------------------------------------------------------
40+
41+
DRBD_BY_RES_PATH: Final = "/dev/drbd/by-res/"
42+
43+
DRBD_PORT_RANGE: Final = (7000, 8000)
44+
45+
# ------------------------------------------------------------------------------
46+
47+
_EXEC_PATH_DRBDSETUP: Final = "/usr/sbin/drbdsetup"
48+
49+
_REGEX_DRBD_OPENER_LINE: Final = re.compile(r"(.*)\s+(\d+)\s+(\d+)")
50+
51+
# ------------------------------------------------------------------------------
52+
53+
def build_drbd_path(resource_name: str, volume_number: int) -> str:
54+
return f"{DRBD_BY_RES_PATH}{resource_name}/{volume_number}"
55+
56+
def get_drbd_name_from_path(path: str) -> str:
57+
# Assume that we have a path like this:
58+
# - "/dev/drbd/by-res/xcp-volume-<UUID>/0"
59+
# - "../xcp-volume-<UUID>/0"
60+
if path.startswith(DRBD_BY_RES_PATH):
61+
prefix_len = len(DRBD_BY_RES_PATH)
62+
elif path.startswith("../"):
63+
prefix_len = 3
64+
else:
65+
return ""
66+
67+
res_name_end = path.find("/", prefix_len)
68+
if res_name_end == -1:
69+
return ""
70+
71+
return path[prefix_len:res_name_end]
72+
73+
# ------------------------------------------------------------------------------
74+
75+
@contextlib.contextmanager
76+
def _handle_drbd_json_error() -> Iterator[None]:
77+
"""
78+
Log instead of raising an error while reading the JSON status of DRBD.
79+
80+
The log message suggests that the JSON format may have changed BUT it's also logged for valid states that
81+
don't have the expected keys or items, like a resource without connection or even a connection without path.
82+
"""
83+
try:
84+
yield
85+
except KeyError as e:
86+
logger.exception(
87+
"The key `%s` could not be found in the DRBD configuration. The JSON format may have changed.", e
88+
)
89+
except Exception as e:
90+
logger.exception("Failed to parse DRBD configuration: `%s`. The JSON format may have changed.", e)
91+
92+
def _get_drbd_status(resource_name: str) -> Dict[str, Any]:
93+
try:
94+
stdout, _stderr, ret_code = run_command([
95+
_EXEC_PATH_DRBDSETUP, "status", resource_name, "--json"
96+
], simple=False)
97+
if ret_code != 0:
98+
return {}
99+
except Exception as e:
100+
logger.error("Failed to get DRBD status: `%s`.", e)
101+
return {}
102+
103+
try:
104+
status = json.loads(stdout)
105+
except Exception as e:
106+
logger.error("Failed to read DRBD status as JSON: `%s`.", e)
107+
return {}
108+
109+
with _handle_drbd_json_error():
110+
return status[0]
111+
return {}
112+
113+
# ------------------------------------------------------------------------------
114+
115+
def get_drbd_connection_address(resource_name: str, node_name: str) -> str:
116+
status = _get_drbd_status(resource_name)
117+
if not status:
118+
return ""
119+
120+
with _handle_drbd_json_error():
121+
for connection in status["connections"]:
122+
if connection["name"] == node_name:
123+
return connection["paths"][0]["remote_host"]["address"]
124+
return ""
125+
126+
def get_drbd_primary_address(resource_name: str) -> str:
127+
status = _get_drbd_status(resource_name)
128+
if not status:
129+
return ""
130+
131+
with _handle_drbd_json_error():
132+
if status["role"] == "Primary":
133+
return status["connections"][0]["paths"][0]["this_host"]["address"]
134+
135+
for connection in status["connections"]:
136+
if connection["peer-role"] == "Primary":
137+
return connection["paths"][0]["remote_host"]["address"]
138+
139+
return ""
140+
141+
# ------------------------------------------------------------------------------
142+
143+
@dataclass(frozen=True)
144+
class DrbdOpener:
145+
pid: int
146+
process_name: str
147+
cmdline: List[str]
148+
# The duration is expressed in milliseconds.
149+
open_duration: int
150+
151+
def get_drbd_local_openers(resource_name: str, volume_number: int) -> List[DrbdOpener]:
152+
assert resource_name, "Cannot get DRBD openers without resource name."
153+
154+
path = Path(f"/sys/kernel/debug/drbd/resources/{resource_name}/volumes/{volume_number}/openers")
155+
try:
156+
lines = path.read_text().splitlines()
157+
except Exception as e:
158+
# The resource is probably available not on this node.
159+
logger.info("Unable to get DRBD openers of volume `%s/%d`: `%s`.", resource_name, volume_number, e)
160+
return []
161+
162+
drbd_openers = []
163+
for line in lines:
164+
match = _REGEX_DRBD_OPENER_LINE.fullmatch(line)
165+
if not match:
166+
logger.warning("Unable to parse DRBD opener line with: `%s`.", line)
167+
continue
168+
169+
groups = match.groups()
170+
pid = int(groups[1])
171+
drbd_openers.append(DrbdOpener(
172+
pid=pid,
173+
process_name=groups[0],
174+
# Note: `cmdline` is empty for `mount` calls. That's correct because `mount` process is dead.
175+
cmdline=get_process_cmdline(pid),
176+
open_duration=int(groups[2])
177+
))
178+
179+
return drbd_openers
180+
181+
# ------------------------------------------------------------------------------
182+
183+
def demote_drbd(resource_name: str) -> bool:
184+
error_message = ""
185+
try:
186+
_stdout, stderr, ret_code = run_command([_EXEC_PATH_DRBDSETUP, "secondary", resource_name], simple=False)
187+
if not ret_code:
188+
return True
189+
error_message = stderr
190+
except Exception as e:
191+
error_message = str(e)
192+
logger.error("Failed to demote DRBD resource `%s`: `%s`.", resource_name, error_message)
193+
return False

‎src/xcp_storage/rpc/api.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,12 @@
1212
# You should have received a copy of the GNU General Public License
1313
# along with this program. If not, see <https://www.gnu.org/licenses/>.
1414

15+
import xcp_storage.rpc.modules.drbd as drbd
1516
import xcp_storage.rpc.modules.echo as echo
1617

1718
# ==============================================================================
1819

1920
__all__ = [
21+
"drbd",
2022
"echo"
2123
]
Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
# Copyright (C) 2026 Vates SAS
2+
#
3+
# This program is free software: you can redistribute it and/or modify
4+
# it under the terms of the GNU General Public License as published by
5+
# the Free Software Foundation, either version 3 of the License, or
6+
# (at your option) any later version.
7+
# This program is distributed in the hope that it will be useful,
8+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10+
# GNU General Public License for more details.
11+
#
12+
# You should have received a copy of the GNU General Public License
13+
# along with this program. If not, see <https://www.gnu.org/licenses/>.
14+
15+
from dataclasses import asdict
16+
17+
from xcp_storage.backends.drbd import get_drbd_local_openers
18+
from xcp_storage.rpc.dispatcher import ApiDispatcher
19+
from xcp_storage.utils.json import JsonDict
20+
21+
from xcp_storage.typing import (
22+
assert_type,
23+
cast,
24+
Dict,
25+
List,
26+
MYPY,
27+
PYREFLY,
28+
Union,
29+
)
30+
31+
# ==============================================================================
32+
33+
@ApiDispatcher.method
34+
def get_openers(resource_name: str, volume_number: int) -> List[JsonDict]:
35+
openers = [asdict(opener) for opener in get_drbd_local_openers(resource_name, volume_number)]
36+
# Some linters, such as pyrefly, may struggle to convert a type to a recursive type.
37+
# We therefore perform a static check (so that it's validated by linters) and we cast it explicitly.
38+
if PYREFLY and not MYPY:
39+
assert_type(openers, List[Dict[str, Union[int, str, List[str]]]])
40+
return cast(List[JsonDict], openers)

‎src/xcp_storage/typing/__init__.py‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,3 +63,13 @@ def __call__(self, *_args, **_kwargs): # type: ignore # noqa: ANN002, ANN003, A
6363

6464
ParamSpec = _SubscriptableListMock() # type: ignore
6565
Concatenate = _SubscriptableListMock() # type: ignore
66+
67+
if not hasattr(typing, "assert_type"):
68+
_T = TypeVar("T") # noqa: F405
69+
def assert_type(value: _T, expected_type: Type[_T]) -> None: # noqa: F405
70+
pass
71+
72+
# ------------------------------------------------------------------------------
73+
74+
MYPY: Final = False # noqa: F405
75+
PYREFLY: Final = False # noqa: F405

‎src/xcp_storage/utils/process.py‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
# You should have received a copy of the GNU General Public License
1313
# along with this program. If not, see <https://www.gnu.org/licenses/>.
1414

15+
from pathlib import Path
1516
import subprocess
1617

1718
import xcp_storage.log as log
@@ -157,3 +158,13 @@ def run_command(
157158
ret_code_callback=ret_code_callback,
158159
quiet=quiet
159160
)
161+
162+
# ------------------------------------------------------------------------------
163+
164+
def get_process_cmdline(pid: int) -> List[str]:
165+
path = Path(f"/proc/{pid}/cmdline")
166+
try:
167+
return [arg.decode() for arg in path.read_bytes().split(b"\0") if arg]
168+
except Exception as e:
169+
logger.info("Unable to get command line of PID `%d`: `%s`.", pid, e)
170+
return []

0 commit comments

Comments
 (0)