feat(core): add DRBD helpers (#11)

Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>

Author: Wescoeur

src/xcp_storage/backends/__init__.py

@@ -0,0 +1,187 @@
+# Copyright (C) 2026  Vates SAS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import contextlib
+from dataclasses import dataclass
+import json
+from pathlib import Path
+import re
+
+import xcp_storage.log as log
+from xcp_storage.utils.process import (
+    get_process_cmdline,
+    run_command,
+)
+
+from xcp_storage.typing import (
+    Any,
+    Dict,
+    Final,
+    Iterator,
+    List,
+)
+
+# ==============================================================================
+
+logger = log.get_logger() # Use default logger.
+
+# ------------------------------------------------------------------------------
+
+DRBD_BY_RES_PATH: Final = "/dev/drbd/by-res/"
+
+DRBD_PORT_RANGE: Final = (7000, 8000)
+
+# ------------------------------------------------------------------------------
+
+_EXEC_PATH_DRBDSETUP: Final = "/usr/sbin/drbdsetup"
+
+_REGEX_DRBD_OPENER_LINE: Final = re.compile(r"(.*)\s+(\d+)\s+(\d+)")
+
+# ------------------------------------------------------------------------------
+
+def build_drbd_path(resource_name: str, volume_number: int) -> str:
+    return f"{DRBD_BY_RES_PATH}{resource_name}/{volume_number}"
+
+def get_drbd_name_from_path(path: str) -> str:
+    # Assume that we have a path like this:
+    # - "/dev/drbd/by-res/xcp-volume-<UUID>/0"
+    # - "../xcp-volume-<UUID>/0"
+    if path.startswith(DRBD_BY_RES_PATH):
+        prefix_len = len(DRBD_BY_RES_PATH)
+    elif path.startswith("../"):
+        prefix_len = 3
+    else:
+        return ""
+
+    res_name_end = path.find("/", prefix_len)
+    if res_name_end == -1:
+        return ""
+
+    return path[prefix_len:res_name_end]
+
+# ------------------------------------------------------------------------------
+
+@contextlib.contextmanager
+def _handle_drbd_json_error() -> Iterator[None]:
+    try:
+        yield
+    except KeyError as e:
+        logger.exception(
+            "The key `%s` could not be found in the DRBD configuration. The JSON format may have changed.", e
+        )
+    except Exception as e:
+        logger.exception("Failed to parse DRBD configuration: `%s`. The JSON format may have changed.", e)
+
+def _get_drbd_status(resource_name: str) -> Dict[str, Any]:
+    try:
+        stdout, _stderr, ret_code = run_command([
+            _EXEC_PATH_DRBDSETUP, "status", resource_name, "--json"
+        ], simple=False)
+        if ret_code != 0:
+             return {}
+    except Exception as e:
+        logger.error("Failed to get DRBD status: `%s`.", e)
+        return {}
+
+    try:
+        status = json.loads(stdout)
+    except Exception as e:
+        logger.error("Failed to read DRBD status as JSON: `%s`.", e)
+        return {}
+
+    with _handle_drbd_json_error():
+        return status[0]
+    return {}
+
+# ------------------------------------------------------------------------------
+
+def get_drbd_connection_address(resource_name: str, node_name: str) -> str:
+    status = _get_drbd_status(resource_name)
+    if not status:
+        return ""
+
+    with _handle_drbd_json_error():
+        for connection in status["connections"]:
+            if connection["name"] == node_name:
+                return connection["paths"][0]["remote_host"]["address"]
+    return ""
+
+def get_drbd_primary_address(resource_name: str) -> str:
+    status = _get_drbd_status(resource_name)
+    if not status:
+        return ""
+
+    with _handle_drbd_json_error():
+        if status["role"] == "Primary":
+            return status["connections"][0]["paths"][0]["this_host"]["address"]
+
+        for connection in status["connections"]:
+            if connection["peer-role"] == "Primary":
+                return connection["paths"][0]["remote_host"]["address"]
+
+    return ""
+
+# ------------------------------------------------------------------------------
+
+@dataclass
+class DrbdOpener:
+    pid: int
+    process_name: str
+    cmdline: List[str]
+    # The duration is expressed in milliseconds.
+    open_duration: int
+
+def get_drbd_local_openers(resource_name: str, volume_number: int) -> List[DrbdOpener]:
+    assert resource_name, "Cannot get DRBD openers without resource name."
+
+    path = Path(f"/sys/kernel/debug/drbd/resources/{resource_name}/volumes/{volume_number}/openers")
+    try:
+        lines = path.read_text().splitlines()
+    except Exception as e:
+        # The resource is probably available not on this node.
+        logger.info("Unable to get DRBD openers of volume `%s/%d`: `%s`.", resource_name, volume_number, e)
+        return []
+
+    drbd_openers = []
+    for line in lines:
+        match = _REGEX_DRBD_OPENER_LINE.match(line)
+        if not match:
+            logger.warning("Unable to parse DRBD opener line with: `%s`.", line)
+            continue
+
+        groups = match.groups()
+        pid = int(groups[1])
+        drbd_openers.append(DrbdOpener(
+            pid=pid,
+            process_name=groups[0],
+            # Note: `cmdline`` is empty for `mount` calls. Logic the PID is dead.
+            cmdline=get_process_cmdline(pid),
+            open_duration=int(groups[2])
+        ))
+
+    return drbd_openers
+
+# ------------------------------------------------------------------------------
+
+def demote_drbd(resource_name: str) -> bool:
+    error_message = ""
+    try:
+        _stdout, stderr, ret_code = run_command([_EXEC_PATH_DRBDSETUP, "secondary", resource_name], simple=False)
+        if not ret_code:
+            return True
+        error_message = stderr
+    except Exception as e:
+        error_message = str(e)
+    logger.error("Failed to demote DRBD resource `%s`: `%s`.", resource_name, error_message)
+    return False

src/xcp_storage/backends/drbd/__init__.py

@@ -12,10 +12,12 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import xcp_storage.rpc.modules.drbd as drbd
 import xcp_storage.rpc.modules.echo as echo
 
 # ==============================================================================
 
 __all__ = [
+    "drbd",
     "echo"
 ]

src/xcp_storage/rpc/api.py

@@ -0,0 +1,40 @@
+# Copyright (C) 2026  Vates SAS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+from dataclasses import asdict
+
+from xcp_storage.backends.drbd import get_drbd_local_openers
+from xcp_storage.rpc.dispatcher import ApiDispatcher
+from xcp_storage.utils.json import JsonDict
+
+from xcp_storage.typing import (
+    assert_type,
+    cast,
+    Dict,
+    List,
+    MYPY,
+    PYREFLY,
+    Union,
+)
+
+# ==============================================================================
+
+@ApiDispatcher.method
+def get_openers(resource_name: str, volume_number: int) -> List[JsonDict]:
+    openers = [asdict(opener) for opener in get_drbd_local_openers(resource_name, volume_number)]
+    # Some linters, such as pyrefly, may struggle to convert a type to a recursive type.
+    # We therefore perform a static check (so that it's validated by linters) and we cast it explicitly.
+    if PYREFLY and not MYPY:
+        assert_type(openers, List[Dict[str, Union[int, str, List[str]]]])
+    return cast(List[JsonDict], openers)

src/xcp_storage/rpc/modules/drbd.py

@@ -63,3 +63,13 @@ def __call__(self, *_args, **_kwargs):  # type: ignore # noqa: ANN002, ANN003, A
 
         ParamSpec = _SubscriptableListMock()  # type: ignore
         Concatenate = _SubscriptableListMock()  # type: ignore
+
+    if not hasattr(typing, "assert_type"):
+        _T = TypeVar("T")  # noqa: F405
+        def assert_type(value: _T, expected_type: Type[_T]) -> None:  # noqa: F405
+            pass
+
+# ------------------------------------------------------------------------------
+
+MYPY: Final = False  # noqa: F405
+PYREFLY: Final = False  # noqa: F405

src/xcp_storage/typing/__init__.py

@@ -12,6 +12,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+from pathlib import Path
 import subprocess
 
 import xcp_storage.log as log
@@ -157,3 +158,13 @@ def run_command(
         ret_code_callback=ret_code_callback,
         quiet=quiet
     )
+
+# ------------------------------------------------------------------------------
+
+def get_process_cmdline(pid: int) -> List[str]:
+    path = Path(f"/proc/{pid}/cmdline")
+    try:
+        return [arg.decode() for arg in path.read_bytes().split(b"\0") if arg]
+    except Exception as e:
+        logger.info("Unable to get command line of PID `%d`: `%s`.", pid, e)
+        return []