import asyncio
import atexit
import logging
import signal
import sys
import threading
from olc.alg.alg_refresh_manager import OlcAlgRefreshManager
from olc.bean.olc_control_request import OlcControlRequest
from olc.collector.collector_manager import OlcCollectorManager
from olc.config.properties_loader import OlcConfigManager
from olc.control.context.context import OlcContext
from olc.control.handler_chain.box_handler_chain_builder import OlcHandlerChain
from olc.limit.node.cluster_limiter_refresh import OlcClusterLimiterRefresh
from olc.rule.olc_rule_convert import OlcRuleConvert
from olc.task_schedule.task_schedule import OlcScheduleManager
logger = logging.getLogger(__name__)
class OLC:
_instance = None
_lock = threading.Lock()
def __init__(self):
logger.info("OLC init")
logger.info("Olc OlcConfigManager init")
OlcConfigManager.get_instance()
logger.info("Olc OlcConfigManager init success")
logger.info("Olc OlcScheduleManager init")
OlcScheduleManager.get_instance()
logger.info("Olc OlcScheduleManager init success")
logger.info("Olc OlcCollectorManager init")
OlcCollectorManager.get_instance().init()
logger.info("Olc OlcCollectorManager init success")
logger.info("olc OlcRuleConvert init")
OlcRuleConvert.get_instance().init()
logger.info("olc OlcRuleConvert init success")
logger.info("Olc OlcAlgRefreshManager init")
OlcAlgRefreshManager.get_instance().start()
logger.info("Olc OlcAlgRefreshManager init success")
logger.info("Olc cluster refresh init")
OlcClusterLimiterRefresh.start()
logger.info("Olc cluster refresh init success")
self._register_shutdown_hooks()
logger.info("olc init success")
@classmethod
def get_instance(cls) -> "OLC":
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = cls()
return cls._instance
@classmethod
async def _async_instance(cls) -> None:
"""用 run_in_executor 避免 init 阻塞事件循环"""
if cls._instance is not None:
return
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, cls.get_instance)
def _register_shutdown_hooks(self):
atexit.register(self._graceful_shutdown)
try:
if threading.current_thread() is threading.main_thread():
if sys.platform != "win32":
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
else:
signal.signal(signal.SIGINT, self._signal_handler)
except ValueError:
logger.warning("Cannot register signal handlers in non-main thread")
def _signal_handler(self, signum: int, frame):
logger.info(f"Received signal {signum}, initiating graceful shutdown")
self._graceful_shutdown()
@staticmethod
def _graceful_shutdown():
try:
logger.info("Starting graceful shutdown")
OlcScheduleManager.get_instance().close()
logger.info("Graceful shutdown completed")
except Exception as e:
logger.error(f"Error during graceful shutdown: {e}", exc_info=True)
@staticmethod
def process(olc_request: OlcControlRequest) -> OlcContext:
OLC.get_instance()
context = OlcContext(olc_request)
try:
if not OlcConfigManager.get_instance().get_sdk_config().switch:
return context
logger.info(f"into olc process request: {olc_request}")
OlcHandlerChain.chain.in_coming(context, None)
except Exception as e:
logger.error(f"Error during process: {e}", exc_info=True)
return context
@staticmethod
def clean(context: OlcContext):
if context is None:
return
OLC.get_instance()
try:
if not OlcConfigManager.get_instance().get_sdk_config().switch:
return
logger.info(f"into olc clean process context: {context}")
OlcHandlerChain.chain.out_coming(context)
except Exception as e:
logger.error(f"Error during clean: {e}", exc_info=True)
@staticmethod
async def async_process(olc_request: OlcControlRequest) -> OlcContext:
await OLC._async_instance()
context = OlcContext(olc_request)
try:
if not OlcConfigManager.get_instance().get_sdk_config().switch:
return context
logger.info(f"into olc async_process request: {olc_request}")
await OlcHandlerChain.chain.async_in_coming(context, None)
except Exception as e:
logger.error(f"Error during async_process: {e}", exc_info=True)
return context
@staticmethod
async def async_clean(context: OlcContext):
if context is None:
return
await OLC._async_instance()
try:
if not OlcConfigManager.get_instance().get_sdk_config().switch:
return
logger.info(f"into olc async_clean context: {context}")
await OlcHandlerChain.chain.async_out_coming(context)
except Exception as e:
logger.error(f"Error during async_clean: {e}", exc_info=True)