Avoid initializing Blockbuster on Supervisor info call (#5901)

* Avoid initializing Blockbuster on Supervisor info call

Instead of creating an instance of Blockbuster to simply check if
Bluckbuster is enabled, use a global variable to store the instance
of Blockbuster and only initialize it when needed. This avoids
unnecessary initialization of Blockbuster when it is not required.

* Update supervisor/utils/blockbuster.py

Co-authored-by: Jan Čermák <sairon@users.noreply.github.com>

* Fix merge and rename singleton class to BlockBusterManager

* Fix pytest

---------

Co-authored-by: Jan Čermák <sairon@users.noreply.github.com>

Author: agners

supervisor/__main__.py

@@ -13,7 +13,7 @@
 
 # pylint: disable=wrong-import-position
 from supervisor import bootstrap  # noqa: E402
-from supervisor.utils.blockbuster import activate_blockbuster  # noqa: E402
+from supervisor.utils.blockbuster import BlockBusterManager  # noqa: E402
 from supervisor.utils.logging import activate_log_queue_handler  # noqa: E402
 
 # pylint: enable=wrong-import-position
@@ -55,7 +55,7 @@ def run_os_startup_check_cleanup() -> None:
     coresys = loop.run_until_complete(bootstrap.initialize_coresys())
     loop.set_debug(coresys.config.debug)
     if coresys.config.detect_blocking_io:
-        activate_blockbuster()
+        BlockBusterManager.activate()
     loop.run_until_complete(coresys.core.connect())
 
     loop.run_until_complete(bootstrap.supervisor_debugger(coresys))

supervisor/api/supervisor.py

@@ -49,11 +49,7 @@
 from ..coresys import CoreSysAttributes
 from ..exceptions import APIError
 from ..store.validate import repositories
-from ..utils.blockbuster import (
-    activate_blockbuster,
-    blockbuster_enabled,
-    deactivate_blockbuster,
-)
+from ..utils.blockbuster import BlockBusterManager
 from ..utils.sentry import close_sentry, init_sentry
 from ..utils.validate import validate_timezone
 from ..validate import version_tag, wait_boot
@@ -110,7 +106,7 @@ async def info(self, request: web.Request) -> dict[str, Any]:
             ATTR_DEBUG_BLOCK: self.sys_config.debug_block,
             ATTR_DIAGNOSTICS: self.sys_config.diagnostics,
             ATTR_AUTO_UPDATE: self.sys_updater.auto_update,
-            ATTR_DETECT_BLOCKING_IO: blockbuster_enabled(),
+            ATTR_DETECT_BLOCKING_IO: BlockBusterManager.is_enabled(),
             ATTR_COUNTRY: self.sys_config.country,
             # Depricated
             ATTR_WAIT_BOOT: self.sys_config.wait_boot,
@@ -180,10 +176,10 @@ async def options(self, request: web.Request) -> None:
                 detect_blocking_io = DetectBlockingIO.ON
 
             if detect_blocking_io == DetectBlockingIO.ON:
-                activate_blockbuster()
+                BlockBusterManager.activate()
             elif detect_blocking_io == DetectBlockingIO.OFF:
                 self.sys_config.detect_blocking_io = False
-                deactivate_blockbuster()
+                BlockBusterManager.deactivate()
 
         # Deprecated
         if ATTR_WAIT_BOOT in body:

supervisor/utils/blockbuster.py

@@ -1,35 +1,37 @@
 """Activate and deactivate blockbuster for finding blocking I/O."""
 
-from functools import cache
 import logging
 
 from blockbuster import BlockBuster
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
 
-@cache
-def _get_blockbuster() -> BlockBuster:
-    """Get blockbuster instance."""
-    return BlockBuster()
-
-
-def blockbuster_enabled() -> bool:
-    """Return true if blockbuster detection is enabled."""
-    blockbuster = _get_blockbuster()
-    # We activate all or none so just check the first one
-    for _, fn in blockbuster.functions.items():
-        return fn.activated
-    return False
-
-
-def activate_blockbuster() -> None:
-    """Activate blockbuster detection."""
-    _LOGGER.info("Activating BlockBuster blocking I/O detection")
-    _get_blockbuster().activate()
-
-
-def deactivate_blockbuster() -> None:
-    """Deactivate blockbuster detection."""
-    _LOGGER.info("Deactivating BlockBuster blocking I/O detection")
-    _get_blockbuster().deactivate()
+class BlockBusterManager:
+    """Manage BlockBuster instance."""
+
+    _instance: BlockBuster | None = None
+
+    @classmethod
+    def is_enabled(cls):
+        """Return true if blockbuster detection is enabled."""
+        if cls._instance is None:
+            return False
+        for _, fn in cls._instance.functions.items():
+            return fn.activated
+        return False
+
+    @classmethod
+    def activate(cls):
+        """Activate blockbuster detection."""
+        _LOGGER.info("Activating BlockBuster blocking I/O detection")
+        if cls._instance is None:
+            cls._instance = BlockBuster()
+        cls._instance.activate()
+
+    @classmethod
+    def deactivate(cls):
+        """Deactivate blockbuster detection."""
+        _LOGGER.info("Deactivating BlockBuster blocking I/O detection")
+        if cls._instance:
+            cls._instance.deactivate()