feat(linstor): add LinstorManager class

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

Author: Wescoeur

xcp-storage/pyproject.toml

@@ -0,0 +1,72 @@
+[tool.mypy]
+mypy_path = "stubs"
+
+disallow_untyped_defs = true
+enable_error_code = ["explicit-override"]
+ignore_missing_imports = false
+implicit_optional = false
+
+[[tool.mypy.overrides]]
+module = "setuptools.*"
+ignore_missing_imports = true
+
+# To generate stubs, run the following commands in the linstor source folder:
+# ./setup.py build
+# stubgen --include-private build/lib/linstor/ -o <XCP_STORAGE_PATH>/stubs/
+# echo "from . import sharedconsts as consts" >> <XCP_STORAGE_PATH>/stubs/linstor/__init__.pyi
+[[tool.mypy.overrides]]
+module = "linstor.*"
+
+disallow_untyped_defs = false
+disable_error_code = [
+    "arg-type",
+    "attr-defined",
+    "explicit-override",
+    "name-defined",
+    "no-redef",
+    "override"
+]
+
+[tool.ruff]
+line-length = 120
+
+[tool.ruff.lint]
+exclude = ["stubs/*"]
+ignore = [
+    "UP006", # pyupgrade: non-pep585-annotation
+    "UP007", # pyupgrade: non-pep604-annotation-union
+    "UP045"  # pyupgrade: non-pep604-annotation-optional
+]
+select = [
+    "ARG",    # flake8-unused-arguments
+    "F",      # Pyflakes
+    "I",      # isort
+    "N",      # pep8-naming
+    "RUF013", # ruff: implicit-optional
+    "RUF059", # ruff: unused-unpacked-variable
+    "SIM",    # flake8-simplify
+    "SLF",    # flake8-self
+    "T20",    # flake8-print
+    "UP"      # pyupgrade
+]
+typing-modules = ["xcp_storage.typing"]
+
+[tool.ruff.lint.isort.sections]
+"typing" = ["xcp_storage.typing"]
+
+[tool.ruff.lint.isort]
+case-sensitive = false
+combine-as-imports = true
+force-sort-within-sections = true
+known-first-party = ["xcp_storage"]
+known-third-party = ["linstor"]
+lines-after-imports = 1
+order-by-type = false
+section-order = [
+    "future",
+    "standard-library",
+    "third-party",
+    "first-party",
+    "local-folder",
+    "typing"
+]

xcp-storage/setup.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+#
+# 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
+
+from setuptools import find_packages, setup
+
+setup(
+    name="xcp-storage",
+    version="1.0.0",
+    description="XCP storage layer, scripts and plugins",
+    author="Ronan Abhamon <ronan.abhamon@vates.tech>",
+    author_email="ronan.abhamon@vates.tech",
+    url="https://vates.tech",
+    license="GPLv3",
+    packages=find_packages(
+        where="src",
+    ),
+    python_requires=">=3.6",
+    package_dir={"": "src"},
+    scripts=[]
+)

xcp-storage/src/xcp_storage/__init__.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+#
+# 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
+import json
+
+import xcp_storage.log as log
+from xcp_storage.utils.process import run_command
+
+from xcp_storage.typing import (
+    Any,
+    Dict,
+    Iterator,
+)
+
+# ==============================================================================
+
+@contextlib.contextmanager
+def _handle_drbd_json_error() -> Iterator[None]:
+    try:
+        yield
+    except KeyError as e:
+        log.error(
+            f"The key `{e}` could not be found in the DRBD configuration. The JSON format may have changed.",
+            exc_info=True
+        )
+    except Exception as e:
+        log.error(f"Failed to parse DRBD configuration: `{e}`. The JSON format may have changed.", exc_info=True)
+
+def _get_drbd_status(resource_name: str) -> Dict[str, Any]:
+    stdout, _stderr, ret_code = run_command(["drbdsetup", "status", resource_name, "--json"], simple=False)
+    if ret_code != 0:
+        return {}
+
+    try:
+        status = json.loads(stdout)
+    except Exception as e:
+        log.error(f"Failed to read DRBD status as JSON: `{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 ""