import os import sys import time import json import argparse from typing import Dict, List, Any from vs_types import BookerStandaloneConfig from gco_wrapper import GCOWrapper from logger_setup import setup_app_logger from booker_standalone import BookerStandalone from configure import REDIS_CFG from utils.cloud_config import load_remote_or_cache, compute_config_hash, fetch_cloud_config, save_local_cache def sync_standalone_groups( groups_conf: Any, active_wrappers: Dict[str, GCOWrapper], app_logger ): """ 根据最新 groups_conf 对比并管理各个 Standalone Booker 实例的生命周期。 支持 JSON List(包含多个 group 数组)或单个 Dict。 """ if isinstance(groups_conf, dict): conf_list = [groups_conf] elif isinstance(groups_conf, list): conf_list = groups_conf else: conf_list = [] new_groups_by_id: Dict[str, BookerStandaloneConfig] = {} for idx, item in enumerate(conf_list): cfg = BookerStandaloneConfig.from_json(item) gid = cfg.identifier or f"standalone_{idx}" new_groups_by_id[gid] = cfg # 1. 停止已被移除或被禁用的组 current_ids = list(active_wrappers.keys()) for gid in current_ids: wrapper = active_wrappers[gid] new_cfg = new_groups_by_id.get(gid) if not new_cfg or not new_cfg.enable: app_logger.info(f"Stopping disabled/removed Standalone group [{gid}]...") try: wrapper.stop() except Exception as e: app_logger.error(f"Error stopping group [{gid}]: {e}") del active_wrappers[gid] # 2. 动态更新现有组,或启动新增组 for gid, cfg in new_groups_by_id.items(): if not cfg.enable: continue if gid in active_wrappers: app_logger.info(f"Updating configuration for active Standalone group [{gid}]...") try: active_wrappers[gid].update_config(cfg) except Exception as e: app_logger.error(f"Error updating config for Standalone group [{gid}]: {e}") else: app_logger.info(f"Starting new wrapper for Standalone group [{gid}]...") try: wrapper = GCOWrapper( gco_class=BookerStandalone, gco_cfg=cfg, redis_conf=REDIS_CFG ) wrapper.load() wrapper.start() active_wrappers[gid] = wrapper except Exception as e: app_logger.error(f"Failed to start Standalone group [{gid}]: {e}") def main(): parser = argparse.ArgumentParser(description="Standalone Booker Runner") parser.add_argument( "-c", "--config", type=str, required=False, default=None, help="Path to local config.json. If specified, disables cloud config and runs with local file." ) parser.add_argument( "--poll-interval", type=float, required=False, default=30.0, help="Interval in seconds to poll cloud config (default: 30.0s)" ) args = parser.parse_args() # 逻辑 1:默认开启云端配置,除非命令行指定了 -c/--config 本地文件路径 if args.config is not None: config_path = args.config use_cloud = False else: config_path = "config/config_standalone.json" use_cloud = True # 逻辑 2:直接从环境变量获取 node_id,默认 node01 node_id = os.getenv("NODE_ID") or "node01" config_key = f"standalone_config:{node_id}" poll_interval = args.poll_interval app_logger = setup_app_logger("Booker") app_logger.info(f"Booker Logger is ready! (Node ID: '{node_id}', Cloud Mode: {use_cloud})") groups_conf, current_hash, loaded_from_remote = load_remote_or_cache( config_key=config_key, local_cache_path=config_path, use_cloud=use_cloud, logger=app_logger.info ) active_wrappers: Dict[str, GCOWrapper] = {} sync_standalone_groups(groups_conf, active_wrappers, app_logger) app_logger.info(f"Successfully initialized {len(active_wrappers)} Standalone Booker group(s).") if use_cloud: app_logger.info(f"Cloud config active for key '{config_key}'. Polling every {poll_interval}s.") last_poll_time = time.time() try: while True: time.sleep(1) now = time.time() if use_cloud and (now - last_poll_time >= poll_interval): last_poll_time = now try: new_conf = fetch_cloud_config(config_key) if new_conf is not None: new_hash = compute_config_hash(new_conf) if new_hash != current_hash: app_logger.info( f"Cloud config change detected (hash: {current_hash[:8]} -> {new_hash[:8]}). Auto-reloading..." ) save_local_cache(config_path, new_conf) sync_standalone_groups(new_conf, active_wrappers, app_logger) current_hash = new_hash app_logger.info(f"Auto-reload completed. Currently {len(active_wrappers)} Standalone group(s) running.") except Exception as e: app_logger.warning(f"Error polling cloud config key '{config_key}': {e}. Retrying next cycle.") except KeyboardInterrupt: app_logger.info("Shutting down Bookers...") for gid, wrapper in list(active_wrappers.items()): try: wrapper.stop() except Exception as e: app_logger.error(f"Error stopping Standalone group [{gid}]: {e}") if __name__ == "__main__": main()