diff --git a/holmes/common/env_vars.py b/holmes/common/env_vars.py index bf71e7e9fe..e1c308d72b 100644 --- a/holmes/common/env_vars.py +++ b/holmes/common/env_vars.py @@ -129,3 +129,9 @@ def load_bool(env_var, default: Optional[bool]) -> Optional[bool]: KEEPALIVE_IDLE = int(os.environ.get("KEEPALIVE_IDLE", 2)) KEEPALIVE_INTVL = int(os.environ.get("KEEPALIVE_INTVL", 2)) KEEPALIVE_CNT = int(os.environ.get("KEEPALIVE_CNT", 5)) + +# Periodic refresh interval for toolset status in server mode (in seconds) +# Set to 0 to disable periodic refresh +TOOLSET_STATUS_REFRESH_INTERVAL_SECONDS = int( + os.environ.get("TOOLSET_STATUS_REFRESH_INTERVAL_SECONDS", 300) +) diff --git a/holmes/config.py b/holmes/config.py index daa47e6d11..64e86c42cb 100644 --- a/holmes/config.py +++ b/holmes/config.py @@ -283,6 +283,25 @@ def create_tool_executor(self, dal: Optional["SupabaseDal"]) -> ToolExecutor: return self._server_tool_executor + def refresh_server_tool_executor( + self, dal: Optional["SupabaseDal"] + ) -> list[tuple[str, str, str]]: + if not self._server_tool_executor: + self.create_tool_executor(dal) + return [] + + current_toolsets = self._server_tool_executor.toolsets + new_toolsets, changes = ( + self.toolset_manager.refresh_server_toolsets_and_get_changes( + current_toolsets, dal + ) + ) + + if changes: + self._server_tool_executor = ToolExecutor(new_toolsets) + + return [(name, old.value, new.value) for name, old, new in changes] + def create_console_toolcalling_llm( self, dal: Optional["SupabaseDal"] = None, diff --git a/holmes/core/tools.py b/holmes/core/tools.py index bb9b5d3792..c7f305d732 100644 --- a/holmes/core/tools.py +++ b/holmes/core/tools.py @@ -671,7 +671,7 @@ def interpolate_command(self, command: str) -> str: return interpolated_command - def check_prerequisites(self): + def check_prerequisites(self, silent: bool = False): self.status = ToolsetStatusEnum.ENABLED # Sort prerequisites by type to fail fast on missing env vars before @@ -741,11 +741,13 @@ def prereq_priority(prereq): self.status == ToolsetStatusEnum.DISABLED or self.status == ToolsetStatusEnum.FAILED ): - logger.info(f"❌ Toolset {self.name}: {self.error}") + if not silent: + logger.info(f"❌ Toolset {self.name}: {self.error}") # no point checking further prerequisites if one failed return - logger.info(f"✅ Toolset {self.name}") + if not silent: + logger.info(f"✅ Toolset {self.name}") @abstractmethod def get_example_config(self) -> Dict[str, Any]: diff --git a/holmes/core/toolset_manager.py b/holmes/core/toolset_manager.py index 1ec01745f7..d25c271d03 100644 --- a/holmes/core/toolset_manager.py +++ b/holmes/core/toolset_manager.py @@ -98,6 +98,7 @@ def _list_all_toolsets( check_prerequisites=True, enable_all_toolsets=False, toolset_tags: Optional[List[ToolsetTag]] = None, + silent: bool = False, ) -> List[Toolset]: """ List all built-in and custom toolsets. @@ -167,16 +168,16 @@ def _list_all_toolsets( enabled_toolsets.append(toolset) else: toolset.status = ToolsetStatusEnum.DISABLED - self.check_toolset_prerequisites(enabled_toolsets) + self.check_toolset_prerequisites(enabled_toolsets, silent=silent) return final_toolsets @classmethod - def check_toolset_prerequisites(cls, toolsets: list[Toolset]): + def check_toolset_prerequisites(cls, toolsets: list[Toolset], silent: bool = False): with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: futures = [] for toolset in toolsets: - futures.append(executor.submit(toolset.check_prerequisites)) + futures.append(executor.submit(toolset.check_prerequisites, silent)) for _ in concurrent.futures.as_completed(futures): pass @@ -365,7 +366,6 @@ def list_console_toolsets( ) return toolsets_with_status - # TODO(mainred): cache and refresh periodically toolset status for server if necessary def list_server_toolsets( self, dal: Optional[SupabaseDal] = None, refresh_status=True ) -> List[Toolset]: @@ -383,6 +383,31 @@ def list_server_toolsets( ) return toolsets_with_status + def refresh_server_toolsets_and_get_changes( + self, + current_toolsets: List[Toolset], + dal: Optional[SupabaseDal] = None, + ) -> tuple[List[Toolset], List[tuple[str, ToolsetStatusEnum, ToolsetStatusEnum]]]: + old_status_by_name: dict[str, ToolsetStatusEnum] = { + toolset.name: toolset.status for toolset in current_toolsets + } + + new_toolsets = self._list_all_toolsets( + dal, + check_prerequisites=True, + enable_all_toolsets=False, + toolset_tags=self.server_tool_tags, + silent=True, + ) + + changes: List[tuple[str, ToolsetStatusEnum, ToolsetStatusEnum]] = [] + for toolset in new_toolsets: + old_status = old_status_by_name.get(toolset.name) + if old_status is not None and old_status != toolset.status: + changes.append((toolset.name, old_status, toolset.status)) + + return new_toolsets, changes + def _load_toolsets_from_paths( self, toolset_paths: Optional[List[FilePath]], diff --git a/server.py b/server.py index 33c35340b1..b53c76524e 100644 --- a/server.py +++ b/server.py @@ -23,6 +23,7 @@ import logging import uvicorn import colorlog +import threading import time from litellm.exceptions import AuthenticationError @@ -38,6 +39,7 @@ ENABLE_TELEMETRY, DEVELOPMENT_MODE, SENTRY_TRACES_SAMPLE_RATE, + TOOLSET_STATUS_REFRESH_INTERVAL_SECONDS, ) from holmes.config import Config from holmes.core.conversations import ( @@ -110,6 +112,37 @@ def sync_before_server_start(): logging.error("Failed to synchronise holmes toolsets", exc_info=True) +def _toolset_status_refresh_loop(): + interval = TOOLSET_STATUS_REFRESH_INTERVAL_SECONDS + if interval <= 0: + logging.info("Periodic toolset status refresh is disabled") + return + + logging.info( + f"Starting periodic toolset status refresh (interval: {interval} seconds)" + ) + + def refresh_loop(): + while True: + time.sleep(interval) + try: + changes = config.refresh_server_tool_executor(dal) + if changes: + for toolset_name, old_status, new_status in changes: + logging.info( + f"Toolset '{toolset_name}' status changed: {old_status} -> {new_status}" + ) + else: + logging.debug("Periodic toolset status refresh: no changes detected") + except Exception: + logging.error( + "Error during periodic toolset status refresh", exc_info=True + ) + + thread = threading.Thread(target=refresh_loop, daemon=True, name="toolset-refresh") + thread.start() + + if ENABLE_TELEMETRY and SENTRY_DSN: # Initialize Sentry for official releases or when development mode is enabled if is_official_release() or DEVELOPMENT_MODE: @@ -452,4 +485,5 @@ def readiness_check(): "%(asctime)s %(levelname)-8s %(message)s" ) sync_before_server_start() + _toolset_status_refresh_loop() uvicorn.run(app, host=HOLMES_HOST, port=HOLMES_PORT, log_config=log_config)