feat(linstor): add LinstorManager class

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

Author: Wescoeur

scripts/linstor-repair

@@ -0,0 +1,268 @@
+#!/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/>.
+
+from typing import Any, Callable, Dict, List, Optional, TypeVar
+
+# TODO: Remove
+import sys
+sys.path[0] = '/opt/xensource/sm/'
+
+import argparse
+import linstor
+import time
+
+from collections import defaultdict
+
+# TODO: Remove
+from linstorvolumemanager import _get_controller_uri
+
+T = TypeVar('T')
+
+# ==============================================================================
+
+def wait_for_condition(function: Callable[[], T], timeout: int, interval: int) -> T:
+    if timeout <= 0:
+        return function()
+
+    start_time = time.time()
+    while True:
+        result = function()
+        if result:
+            return result
+        if time.time() - start_time >= timeout:
+            return None
+        time.sleep(interval)
+
+def logger(message: str) -> None:
+    print(message)
+
+# ==============================================================================
+
+class LinstorManagerError(Exception):
+    ERR_GENERIC = 0
+    ERR_NETWORK = 1
+    ERR_CONTROLLER_NOT_FOUND = 2
+    ERR_RESOURCE_DESTROY = 3
+    ERR_RESOURCE_TOGGLE = 4
+    ERR_RESOURCE_PROP_UPDATE = 5
+
+    def __init__(self, message: str, code=ERR_GENERIC) -> None:
+        super().__init__(message)
+        self._code = code
+
+    @property
+    def code(self) -> int:
+        return self._code
+
+# ------------------------------------------------------------------------------
+
+class LinstorManagerBase(object):
+    def __init__(self, uri = None) -> None:
+        self._linstor = None
+        self._uri = uri
+
+    def connect(self) -> None:
+        if self._linstor:
+            return
+
+        uri = None
+        network_issue = False
+        def connect_impl():
+            nonlocal uri
+            nonlocal network_issue
+
+            if self._uri:
+                uri = self._uri
+            else:
+                uri = self.find_controller_uri(timeout=0)
+                if not uri:
+                    return False
+
+            network_issue = False
+            try:
+                instance = linstor.Linstor(uri, timeout=5, keep_alive=True)
+                instance.connect()
+                self._linstor = instance
+                self._uri = uri
+                return True
+            except (linstor.errors.LinstorNetworkError, linstor.errors.LinstorTimeoutError):
+                network_issue = True
+            except Exception as e:
+                logger(f"Unable to connect to LINSTOR: `{e}`.")
+            return False
+
+        if wait_for_condition(connect_impl, timeout=120, interval=1):
+            return
+
+        self._uri = None
+        if not uri:
+            raise LinstorManagerError(
+                "Unable to find controller uri...",
+                LinstorManagerError.ERR_CONTROLLER_NOT_FOUND
+            )
+
+        raise LinstorManagerError(
+            f"Unable to connect to LINSTOR with URI: `{uri}`.",
+            LinstorManagerError.ERR_NETWORK if network_issue else LinstorManagerError.ERR_GENERIC
+        )
+
+    @staticmethod
+    def find_controller_uri(timeout: int = 30) -> Optional[str]:
+        return wait_for_condition(_get_controller_uri, timeout, interval=1)
+
+    def _exec_query(self, query: Callable[[linstor.Linstor, T], T], *args, **kwargs) -> T:
+        while True:
+            self.connect()
+            try:
+                return query(self._linstor, *args, **kwargs)
+            except (linstor.errors.LinstorNetworkError, linstor.errors.LinstorTimeoutError):
+                self._linstor = None
+            except Exception as e:
+                raise LinstorManagerError(f"LINSTOR query exception: `{e}`.") from e
+
+    @staticmethod
+    def _filter_errors(result: List[Any]) -> List[linstor.responses.ApiCallResponse]:
+        return [
+            err for err in result
+            if hasattr(err, 'is_error') and err.is_error()
+        ]
+
+    @classmethod
+    def _get_error_str(cls, result: List[Any]) -> str:
+        return ', '.join([
+            err.message for err in cls._filter_errors(result)
+        ])
+
+# ------------------------------------------------------------------------------
+
+class LinstorManager(LinstorManagerBase):
+    def resource_delete(self, resource_name: str, node_name: str) -> None:
+        result = self._exec_query(
+            linstor.Linstor.resource_delete,
+            node_name,
+            resource_name
+        )
+        errors = self._filter_errors(result)
+        if errors:
+            error_str = self._get_error_str(errors)
+            raise LinstorManagerError(
+                f"Failed to destroy resource `{resource_name}` on node `{node_name}`: `{error_str}`.",
+                LinstorManagerError.ERR_RESOURCE_DESTROY
+            )
+
+    def resource_toggle(self, resource_name: str, node_name: str, diskless: bool) -> None:
+        result = self._exec_query(
+            linstor.Linstor.resource_toggle_disk,
+            node_name,
+            resource_name,
+            diskless=diskless
+        )
+        errors = self._filter_errors(result)
+        if errors:
+            state = "Diskless" if diskless else "Diskful"
+            error_str = self._get_error_str(errors)
+            raise LinstorManagerError(
+                f"Could not toggle resource `{resource_name}` on node `{node_name}` to {state}: `{error_str}`.",
+                LinstorManagerError.ERR_RESOURCE_TOGGLE
+            )
+
+    def resource_remove_skip_disk_flag(self, resource_name: str, node_name: str) -> None:
+        return self._resource_delete_properties(resource_name, node_name, ["DrbdOptions/SkipDisk"])
+
+    def resources_create_missing_replicas(self) -> None:
+        pass # TODO
+
+    def resources_remove_skip_disks(self) -> None:
+        name_to_resources = self._fetch_resources()
+
+        for resources in name_to_resources.values():
+            skip_disk_resources = {
+                resource.node_name: resource
+                for resource in resources
+                if self._is_skip_disk_resource(resource)
+            }
+            # Skip disk deletion is only permitted if all remaining volumes are in a stable state.
+            if not skip_disk_resources or not all(
+                self._is_online_disk_state(volume.state.disk_state)
+                for resource in resources
+                if resource.node_name not in skip_disk_resources
+                for volume in resource.volumes
+            ):
+                continue
+
+            for resource in skip_disk_resources.values():
+                try:
+                    if linstor.consts.FLAG_DISKLESS not in resource.flags:
+                        self.resource_toggle(resource.name, resource.node_name, True)
+                    self.resource_remove_skip_disk_flag(resource.name, resource.node_name)
+                except LinstorManagerError as e:
+                    if e.code == LinstorManagerError.ERR_NETWORK:
+                        raise
+                    logger(f"Failed to delete skip disk on resource `{resource.name}` on node `{resource.node_name}`: {e}")
+
+    def _fetch_resources(self) -> Dict[str, List[linstor.responses.Resource]]:
+        resources = defaultdict(list)
+        for resource in self._exec_query(linstor.Linstor.resource_list_raise).resources:
+            resources[resource.name].append(resource)
+        return resources
+
+    def _resource_delete_properties(self, resource_name: str, node_name: str, properties: List[str]) -> None:
+        result = self._exec_query(
+            linstor.Linstor.resource_modify,
+            node_name,
+            resource_name,
+            property_dict={},
+            delete_props=properties
+        )
+        errors = self._filter_errors(result)
+        if errors:
+            error_str = self._get_error_str(errors)
+            raise LinstorManagerError(
+                f"Failed to destroy resource properties of `{resource_name}` on node `{node_name}`: `{error_str}`.",
+                LinstorManagerError.ERR_RESOURCE_PROP_UPDATE
+            )
+
+    @staticmethod
+    def _is_online_disk_state(disk_state: str) -> bool:
+        return disk_state in ("UpToDate", "Diskless") or disk_state.startswith("SyncTarget")
+
+    @classmethod
+    def _is_online_volume(cls, volume: linstor.responses.Volume) -> bool:
+        return cls._is_online_disk_state(volume.state.disk_state)
+
+    @staticmethod
+    def _is_skip_disk_resource(resource: linstor.responses.Resource) -> bool:
+        return resource.properties.get("DrbdOptions/SkipDisk") == "True"
+
+# ==============================================================================
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-u', '--uri', required=False)
+
+    parser.add_argument('--create-missing-replicas', action='store_true')
+    parser.add_argument('--remove-skip-disks', action='store_true')
+
+    args = parser.parse_args()
+
+    linstor_manager = LinstorManager(args.uri)
+    if args.remove_skip_disks:
+        linstor_manager.resources_remove_skip_disks()
+    if args.create_missing_replicas:
+        linstor_manager.resources_create_missing_replicas()
+
+if __name__ == '__main__':
+    main()