mirror of
https://github.com/0O0o0oOoO00/Alas.git
synced 2026-05-15 03:09:25 +08:00
89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Dict
|
|
|
|
from module.config.config import AzurLaneConfig
|
|
from module.counter import MaxCounter
|
|
from module.notify import handle_notify
|
|
from module.webui.process_manager import ProcessManager
|
|
|
|
|
|
@dataclass
|
|
class InstanceSetting:
|
|
counter: MaxCounter
|
|
enable_notify: bool
|
|
notify_config: str
|
|
|
|
|
|
class InstanceWatcher:
|
|
instance = None
|
|
|
|
def __init__(self):
|
|
self.watcher: threading.Thread = None
|
|
self.instances: Dict[str, InstanceSetting] = dict()
|
|
|
|
def start(self):
|
|
if self.watcher is not None:
|
|
if self.watcher.is_alive():
|
|
return
|
|
self.watcher = threading.Thread(target=self.watcher_thread, daemon=True)
|
|
self.watcher.start()
|
|
|
|
def check_instances(self):
|
|
ins_has_triggered = []
|
|
for name, setting in self.instances.items():
|
|
ins = ProcessManager.get_manager(name)
|
|
if not ins.alive:
|
|
if ins.state != 3:
|
|
continue
|
|
if setting.counter.count_once(throw=False):
|
|
ins.start("alas")
|
|
if setting.enable_notify:
|
|
handle_notify(
|
|
setting.notify_config,
|
|
title=f"Alas <{ins.config_name}> instance auto restarted",
|
|
content=f"Critical error occurred, instance restarted",
|
|
)
|
|
else:
|
|
ins_has_triggered.append(name)
|
|
handle_notify(
|
|
setting.notify_config,
|
|
title=f"Alas <{ins.config_name}> instance restarted too many times",
|
|
content=f"Too many critical error occurred, instance restarted too many times",
|
|
)
|
|
|
|
for i in ins_has_triggered:
|
|
self.remove_instance(i)
|
|
|
|
def try_add_instance(self, name):
|
|
config = AzurLaneConfig(name)
|
|
full_config = config.full_config
|
|
if full_config.Restart_InstanceRestart_Enable:
|
|
setting = InstanceSetting(
|
|
counter=MaxCounter(full_config.Restart_InstanceRestart_MaxRetryTimes),
|
|
enable_notify=full_config.Restart_InstanceRestart_Notify,
|
|
notify_config=full_config.Alas_Error_OnePushConfig
|
|
)
|
|
self.instances[name] = setting
|
|
|
|
def remove_instance(self, name):
|
|
try:
|
|
self.instances.pop(name)
|
|
except Exception:
|
|
...
|
|
|
|
def watcher_thread(self):
|
|
while 1:
|
|
time.sleep(60)
|
|
try:
|
|
self.check_instances()
|
|
except Exception:
|
|
...
|
|
|
|
@staticmethod
|
|
def get_instance() -> "InstanceWatcher":
|
|
if InstanceWatcher.instance is None:
|
|
InstanceWatcher.instance = InstanceWatcher()
|
|
return InstanceWatcher.instance
|