feat(linstor): add LinstorManager class

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

Author: Wescoeur

scripts/linstor-repair

@@ -0,0 +1,385 @@
+#!/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_RESOURCE_DESTROY = 2
+    ERR_RESOURCE_TOGGLE = 3
+    ERR_RESOURCE_PROP_UPDATE = 4
+
+    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
+        def connect_impl():
+            nonlocal uri
+
+            if self._uri:
+                uri = self._uri
+            else:
+                uri = self.find_controller_uri(timeout=0)
+                if not uri:
+                    return 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):
+                pass
+            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_NETWORK
+            )
+
+        raise LinstorManagerError(
+            f"Unable to connect to LINSTOR with URI: `{uri}`.",
+            LinstorManagerError.ERR_NETWORK
+        )
+
+    @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):
+    class ResourceReplicas(object):
+        def __init__(self, name: str, node_names: List[str], expected_count: int) -> None:
+            self.name = name
+            self.node_names = node_names
+            self.expected_count = expected_count
+
+    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:
+    def resources_get_missing_replicas(self, resource_group_name: Optional[str] = None) -> List[ResourceReplicas]:
+        resource_replicas = []
+        group_name_to_replication_count = {
+            resource_group.name: resource_group.select_filter.place_count
+            for resource_group in self._fetch_resource_groups(resource_group_name)
+        }
+        if not group_name_to_replication_count:
+            return {}
+
+        name_to_resources = self._fetch_resources()
+        name_to_resource_definitions = self._fetch_resource_definitions()
+        for resource_definition in name_to_resource_definitions.values():
+            replication_count = group_name_to_replication_count.get(resource_definition.resource_group_name)
+            if replication_count is None:
+                continue
+
+            if resource_definition.resource_group_name not in group_name_to_replication_count:
+                continue
+
+            resources = name_to_resources.get(resource_definition.name)
+            if not resources:
+                continue
+
+            diskful_nodes = [
+                resource.node_name for resource in resources
+                if linstor.consts.FLAG_DISKLESS not in resource.flags
+            ]
+            if len(diskful_nodes) < replication_count:
+                resource_replicas.append(
+                    self.ResourceReplicas(resource_definition.name, diskful_nodes, replication_count)
+                )
+
+        return resource_replicas
+
+    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}`"
+                        f" on node `{resource.node_name}`: {e}"
+                    )
+
+    def _fetch_storage_pools(
+        self, node_filter: Optional[List[str]] = None, name_filter: Optional[List[str]] = None
+    ) -> List[linstor.responses.StoragePool]:
+        result = self._exec_query(
+            linstor.Linstor.storage_pool_list, filter_by_nodes=node_filter, filter_by_stor_pools=name_filter
+        )
+        if not result:
+            raise LinstorManagerError("Failed to fetch storage pools: no data.")
+        result = result[0]
+
+        if not isinstance(result, linstor.responses.StoragePoolListResponse):
+            raise LinstorManagerError(
+                f"Failed to fetch storage pools: `{result}`.",
+                LinstorManagerError.ERR_STORAGE_POOL_FETCH
+            )
+
+        return result.storage_pools
+
+    def _fetch_resource_groups(self, name_filter: Optional[List[str]] = None) -> List[linstor.responses.ResourceGroup]:
+        # There is no `resource_group_list` method, so find LINSTOR API call exception...
+        try:
+            return self._exec_query(
+                linstor.Linstor.resource_group_list_raise, filter_by_resource_groups=name_filter
+            ).resource_groups
+        except Exception as e:
+            cause = e.__cause__
+            if not cause:
+                raise
+            raise LinstorManagerError( # pylint: disable = W0707
+                f"Failed to fetch resource groups: `{cause}`.",
+                LinstorManagerError.ERR_RESOURCE_GROUP_FETCH
+            )
+
+    def _fetch_resource_definitions(
+        self, fetch_volumes = False, name_filter: Optional[List[str]] = None
+    ) -> Dict[str, List[linstor.responses.ResourceDefinition]]:
+        result = self._exec_query(
+            linstor.Linstor.resource_dfn_list,
+            query_volume_definitions=fetch_volumes,
+            filter_by_resource_definitions=name_filter
+        )
+        if not result:
+            raise LinstorManagerError(
+                "Failed to fetch resource definitions: no data.",
+                LinstorManagerError.ERR_RESOURCE_DEFINITION_FETCH
+            )
+        result = result[0]
+
+        if not isinstance(result, linstor.responses.ResourceDefinitionResponse):
+            raise LinstorManagerError(
+                f"Failed to fetch resource definitions: `{result}`.",
+                LinstorManagerError.ERR_RESOURCE_DEFINITION_FETCH
+            )
+
+        return {resource_definition.name: resource_definition for resource_definition in result.resource_definitions}
+
+    def _fetch_resources(self) -> Dict[str, List[linstor.responses.Resource]]:
+        result = self._exec_query(linstor.Linstor.resource_list)
+        if not result:
+            raise LinstorManagerError(
+                "Failed to fetch resources: no data.",
+                LinstorManagerError.ERR_RESOURCE_FETCH
+            )
+        result = result[0]
+
+        if not isinstance(result, linstor.responses.ResourceResponse):
+            raise LinstorManagerError(
+                f"Failed to fetch resources: `{result}`.",
+                LinstorManagerError.ERR_RESOURCE_FETCH
+            )
+
+        resources = defaultdict(list)
+        for resource in result.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")
+    parser.add_argument("--evacuate-resources", 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 args.evacuate_resources:
+        linstor_manager.resources_evacuate()
+
+if __name__ == "__main__":
+    main()