Use "transitions" as the workflow handler

.gitignore

@@ -86,7 +86,7 @@ dmypy.json
 cython_debug/
 
 # Project specific
-data.yaml
+data.pickle
 config.yaml
 .ceph
 .k8s

requirements.txt

@@ -4,6 +4,7 @@ certifi==2024.2.2
 cffi==1.16.0
 charset-normalizer==3.3.2
 cryptography==42.0.4
+dill==0.3.8
 decorator==5.1.1
 Deprecated==1.2.14
 fabric==3.2.2
@@ -26,6 +27,7 @@ requests-oauthlib==1.3.1
 rsa==4.9
 six==1.16.0
 structlog==24.1.0
+transitions==0.9.0
 urllib3==2.2.1
 yamale==5.1.0
 websocket-client==1.7.0

src/config.example.yaml

@@ -1,5 +1,5 @@
 general:
-  module_data_file: data.yaml
+  machine_pickle_file: data.pickle
 
 logging:
   level: INFO # level at which logging should start

src/rookify/__main__.py

@@ -1,10 +1,9 @@
 # -*- coding: utf-8 -*-
 
-import os
-import rookify.modules
-from types import MappingProxyType
+from .modules import load_modules
+from .modules.machine import Machine
 from .logger import configure_logging, get_logger
-from .yaml import load_config, load_module_data, save_module_data
+from .yaml import load_config
 
 
 def main() -> None:
@@ -23,56 +22,7 @@ def main() -> None:
     log = get_logger()
     log.debug("Executing Rookify")
 
-    preflight_modules, migration_modules = rookify.modules.load_modules(
-        config["migration_modules"]
-    )
+    machine = Machine(config["general"].get("machine_pickle_file"))
+    load_modules(machine, config)
 
-    module_data = dict()
-
-    try:
-        module_data.update(load_module_data(config["general"]["module_data_file"]))
-    except FileNotFoundError:
-        pass
-
-    # Get absolute path of the rookify instance
-    rookify_path = os.path.dirname(__file__)
-
-    # Run preflight requirement modules
-    for preflight_module in preflight_modules:
-        module_path = os.path.join(
-            rookify_path, "modules", preflight_module.MODULE_NAME
-        )
-        handler = preflight_module.HANDLER_CLASS(
-            config=MappingProxyType(config),
-            data=MappingProxyType(module_data),
-            module_path=module_path,
-        )
-        result = handler.run()
-        module_data[preflight_module.MODULE_NAME] = result
-
-    # Run preflight and append handlers to list
-    handlers = list()
-    for migration_module in migration_modules:
-        module_path = os.path.join(
-            rookify_path, "modules", migration_module.MODULE_NAME
-        )
-        handler = migration_module.HANDLER_CLASS(
-            config=MappingProxyType(config),
-            data=MappingProxyType(module_data),
-            module_path=module_path,
-        )
-        handler.preflight()
-        handlers.append((migration_module, handler))
-
-    # Run migration modules
-    for migration_module, handler in handlers:
-        result = handler.run()
-        module_data[migration_module.MODULE_NAME] = result
-
-    save_module_data(config["general"]["module_data_file"], module_data)
-
-    log.info("Data was updated to module_data_file.")
-
-
-if __name__ == "__main__":
-    main()
+    machine.execute()

src/rookify/config.schema.yaml

@@ -1,5 +1,5 @@
 general:
-  module_data_file: str()
+  machine_pickle_file: str(required=False)
 
 logging:
   level: str()

src/rookify/modules/__init__.py

@@ -1,11 +1,8 @@
 # -*- coding: utf-8 -*-
 
 import importlib
-import types
-from typing import List
-
-from collections import OrderedDict
-from .module import ModuleHandler
+from typing import Any, Dict
+from .machine import Machine
 
 
 class ModuleLoadException(Exception):
@@ -24,107 +21,40 @@ def __init__(self, module_name: str, message: str):
         self.message = message
 
 
-def load_modules(
-    module_names: List[str],
-) -> tuple[List[types.ModuleType], List[types.ModuleType]]:
+def _load_module(machine: Machine, config: Dict[str, Any], module_name: str) -> None:
+    """
+    Dynamically loads a module from the 'rookify.modules' package.
+
+    :param module_names: The module names to load
+    :return: returns tuple of preflight_modules, modules
+    """
+
+    module = importlib.import_module("rookify.modules.{0}".format(module_name))
+    additional_modules = []
+
+    if not hasattr(module, "ModuleHandler") or not callable(
+        getattr(module.ModuleHandler, "register_states")
+    ):
+        raise ModuleLoadException(module_name, "Module structure is invalid")
+
+    if hasattr(module.ModuleHandler, "REQUIRES"):
+        assert isinstance(module.ModuleHandler.REQUIRES, list)
+        additional_modules = module.ModuleHandler.REQUIRES
+
+    for module_name in additional_modules:
+        _load_module(machine, config, module_name)
+
+    module.ModuleHandler.register_states(machine, config)
+
+
+def load_modules(machine: Machine, config: Dict[str, Any]) -> None:
     """
     Dynamically loads modules from the 'modules' package.
 
     :param module_names: The module names to load
     :return: returns tuple of preflight_modules, modules
     """
 
-    # Sanity checks for modules
-    def check_module_sanity(module_name: str, module: types.ModuleType) -> None:
-        for attr_type, attr_name in (
-            (ModuleHandler, "HANDLER_CLASS"),
-            (str, "MODULE_NAME"),
-            (list, "REQUIRES"),
-            (list, "AFTER"),
-            (list, "PREFLIGHT_REQUIRES"),
-        ):
-            if not hasattr(module, attr_name):
-                raise ModuleLoadException(
-                    module_name, f"Module has no attribute {attr_name}"
-                )
-
-            attr = getattr(module, attr_name)
-            if not isinstance(attr, attr_type) and not issubclass(attr, attr_type):
-                raise ModuleLoadException(
-                    module_name, f"Attribute {attr_name} is not type {attr_type}"
-                )
-
-    # Load the modules in the given list and recursivley load required modules
-    required_modules: OrderedDict[str, types.ModuleType] = OrderedDict()
-
-    def load_required_modules(
-        modules_out: OrderedDict[str, types.ModuleType], module_names: List[str]
-    ) -> None:
-        for module_name in module_names:
-            if module_name in modules_out:
-                continue
-
-            module = importlib.import_module(f".{module_name}", "rookify.modules")
-            check_module_sanity(module_name, module)
-
-            load_required_modules(modules_out, module.REQUIRES)
-            module.AFTER.extend(module.REQUIRES)
-
-            modules_out[module_name] = module
-
-    load_required_modules(required_modules, module_names)
-
-    # Recursively load the modules in the PREFLIGHT_REQUIRES attribute of the given modules
-    preflight_modules: OrderedDict[str, types.ModuleType] = OrderedDict()
-
-    def load_preflight_modules(
-        modules_in: OrderedDict[str, types.ModuleType],
-        modules_out: OrderedDict[str, types.ModuleType],
-        module_names: List[str],
-    ) -> None:
-        for module_name in module_names:
-            if module_name in modules_out:
-                continue
-
-            module = importlib.import_module(f".{module_name}", "rookify.modules")
-            check_module_sanity(module_name, module)
-
-            # We have to check, if the preflight_requires list is already loaded as migration requirement
-            for preflight_requirement in module.PREFLIGHT_REQUIRES:
-                if preflight_requirement in modules_in:
-                    raise ModuleLoadException(
-                        module_name,
-                        f"Module {preflight_requirement} is already loaded as migration requirement",
-                    )
-
-            load_preflight_modules(modules_in, modules_out, module.PREFLIGHT_REQUIRES)
-            if module_name not in modules_in:
-                modules_out[module_name] = module
-
-    load_preflight_modules(
-        required_modules, preflight_modules, list(required_modules.keys())
-    )
-
-    # Sort the modules by the AFTER keyword
-    modules: OrderedDict[str, types.ModuleType] = OrderedDict()
-
-    def sort_modules(
-        modules_in: OrderedDict[str, types.ModuleType],
-        modules_out: OrderedDict[str, types.ModuleType],
-        module_names: List[str],
-    ) -> None:
-        for module_name in module_names:
-            if module_name not in modules_in:
-                continue
-
-            if module_name in modules_out:
-                continue
-
-            after_modules_name = modules_in[module_name].AFTER
-            sort_modules(modules_in, modules_out, after_modules_name)
-
-            modules_out[module_name] = modules_in[module_name]
-
-    sort_modules(required_modules, modules, list(required_modules.keys()))
-
-    return list(preflight_modules.values()), list(modules.values())
+    for entry in importlib.resources.files("rookify.modules").iterdir():
+        if entry.is_dir() and entry.name in config["migration_modules"]:
+            _load_module(machine, config, entry.name)

src/rookify/modules/analyze_ceph/__init__.py

@@ -1,10 +1,3 @@
 # -*- coding: utf-8 -*-
-# type: ignore
 
-from .main import AnalyzeCephHandler
-
-MODULE_NAME = "analyze_ceph"
-HANDLER_CLASS = AnalyzeCephHandler
-REQUIRES = []
-AFTER = []
-PREFLIGHT_REQUIRES = []
+from .main import AnalyzeCephHandler as ModuleHandler  # noqa

src/rookify/modules/analyze_ceph/main.py

@@ -1,29 +1,42 @@
 # -*- coding: utf-8 -*-
 
-from ..module import ModuleHandler
 from typing import Any, Dict
+from ..machine import Machine
+from ..module import ModuleHandler
 
 
 class AnalyzeCephHandler(ModuleHandler):
-    def run(self) -> Any:
+    def preflight(self) -> Any:
         commands = ["mon dump", "osd dump", "device ls", "fs dump", "node ls"]
 
-        results: Dict[str, Any] = dict()
+        state = self.machine.get_preflight_state("AnalyzeCephHandler")
+        state.data: Dict[str, Any] = {}  # type: ignore
+
         for command in commands:
             parts = command.split(" ")
-            leaf = results
+            leaf = state.data
             for idx, part in enumerate(parts):
                 if idx < len(parts) - 1:
-                    leaf[part] = dict()
+                    leaf[part] = {}
                 else:
                     leaf[part] = self.ceph.mon_command(command)
                 leaf = leaf[part]
 
-        self.logger.info("Dictionary created")
-        results["ssh"] = dict()
-        results["ssh"]["osd"] = dict()
-        for node, values in results["node"]["ls"]["osd"].items():
+        self.logger.debug("AnalyzeCephHandler commands executed")
+
+        state.data["ssh"] = {}
+        state.data["ssh"]["osd"] = {}
+
+        for node, values in state.data["node"]["ls"]["osd"].items():
             devices = self.ssh.command(node, "find /dev/ceph-*/*").stdout.splitlines()
-            results["ssh"]["osd"][node] = {"devices": devices}
+            state.data["ssh"]["osd"][node] = {"devices": devices}
+
         self.logger.info("AnalyzeCephHandler ran successfully.")
-        return results
+
+    @staticmethod
+    def register_preflight_state(
+        machine: Machine, state_name: str, handler: ModuleHandler, **kwargs: Any
+    ) -> None:
+        ModuleHandler.register_preflight_state(
+            machine, state_name, handler, tags=["data"]
+        )

src/rookify/modules/cephx_auth_config/__init__.py

@@ -1,10 +1,3 @@
 # -*- coding: utf-8 -*-
-# type: ignore
 
-from .main import CephXAuthHandler
-
-MODULE_NAME = "cephx_auth_config"
-HANDLER_CLASS = CephXAuthHandler
-REQUIRES = []
-AFTER = []
-PREFLIGHT_REQUIRES = []
+from .main import CephXAuthHandler as ModuleHandler  # noqa

src/rookify/modules/cephx_auth_config/main.py

@@ -1,16 +1,37 @@
 # -*- coding: utf-8 -*-
 
-from ..module import ModuleHandler
 from typing import Any
+from ..machine import Machine
+from ..module import ModuleException, ModuleHandler
 
 
 class CephXAuthHandler(ModuleHandler):
-    def run(self) -> Any:
-        self.logger.debug("Reconfiguring Ceph to expect cephx auth")
+    def preflight(self) -> Any:
+        if not self.is_cephx_set(self.ceph.conf_get("auth_cluster_required")):
+            raise ModuleException(
+                "Ceph config value auth_cluster_required does not contain cephx"
+            )
 
-        self.ceph.conf_set("auth_cluster_required", "cephx")
-        self.ceph.conf_set("auth_service_required", "cephx")
-        self.ceph.conf_set("auth_client_required", "cephx")
+        if not self.is_cephx_set(self.ceph.conf_get("auth_service_required")):
+            raise ModuleException(
+                "Ceph config value auth_service_required does not contain cephx"
+            )
 
-        self.logger.info("Reconfigured Ceph to expect cephx auth")
-        return {"reconfigured": True}
+        if not self.is_cephx_set(self.ceph.conf_get("auth_client_required")):
+            raise ModuleException(
+                "Ceph config value auth_client_required does not contain cephx"
+            )
+
+        self.machine.get_preflight_state("CephXAuthHandler").verified = True
+        self.logger.info("Validated Ceph to expect cephx auth")
+
+    def is_cephx_set(self, values: str) -> Any:
+        return "cephx" in [value.strip() for value in values.split(",")]
+
+    @staticmethod
+    def register_preflight_state(
+        machine: Machine, state_name: str, handler: ModuleHandler, **kwargs: Any
+    ) -> None:
+        ModuleHandler.register_preflight_state(
+            machine, state_name, handler, tags=["verified"]
+        )