|
@@ -3,14 +3,12 @@ import time
|
|
|
import json
|
|
import json
|
|
|
import threading
|
|
import threading
|
|
|
import random
|
|
import random
|
|
|
-from datetime import datetime
|
|
|
|
|
-from typing import List, Dict, Callable, Any
|
|
|
|
|
|
|
+from typing import List, Dict, Callable
|
|
|
|
|
|
|
|
from vs_types import GroupConfig, VSPlgConfig, Task, VSQueryResult, AppointmentType, AvailabilityStatus
|
|
from vs_types import GroupConfig, VSPlgConfig, Task, VSQueryResult, AppointmentType, AvailabilityStatus
|
|
|
from vs_plg_factory import VSPlgFactory
|
|
from vs_plg_factory import VSPlgFactory
|
|
|
from toolkit.thread_pool import ThreadPool
|
|
from toolkit.thread_pool import ThreadPool
|
|
|
from toolkit.vs_cloud_api import VSCloudApi
|
|
from toolkit.vs_cloud_api import VSCloudApi
|
|
|
-from toolkit.backoff import ExponentialBackoff
|
|
|
|
|
from utils.safe_redis_cli import SafeRedisClient
|
|
from utils.safe_redis_cli import SafeRedisClient
|
|
|
|
|
|
|
|
|
|
|
|
@@ -28,17 +26,9 @@ class OrderBookerGCO:
|
|
|
self.m_tasks: List[Task] = []
|
|
self.m_tasks: List[Task] = []
|
|
|
self.m_lock = threading.RLock()
|
|
self.m_lock = threading.RLock()
|
|
|
self.m_stop_event = threading.Event()
|
|
self.m_stop_event = threading.Event()
|
|
|
-
|
|
|
|
|
self.redis_client = SafeRedisClient(redis_conf, self.m_logger)
|
|
self.redis_client = SafeRedisClient(redis_conf, self.m_logger)
|
|
|
-
|
|
|
|
|
- self.m_pending_order_by_queue: Dict[str, int] = {}
|
|
|
|
|
- self.m_last_spawn_times: Dict[str, float] = {}
|
|
|
|
|
self.m_task_data_cache: Dict[str, dict] = {}
|
|
self.m_task_data_cache: Dict[str, dict] = {}
|
|
|
-
|
|
|
|
|
self.m_tracker_key = f"vs:worker:tasks_tracker:{self.m_cfg.identifier}"
|
|
self.m_tracker_key = f"vs:worker:tasks_tracker:{self.m_cfg.identifier}"
|
|
|
- self.queue_backoff = ExponentialBackoff(base_delay=1*60.0, max_delay=10*60.0, factor=2.0)
|
|
|
|
|
- self.account_backoff = ExponentialBackoff(base_delay=5*60.0, max_delay=2*60*60.0, factor=2.0)
|
|
|
|
|
- self.task_backoff = ExponentialBackoff(base_delay=10, max_delay=30*60.0, factor=2.0)
|
|
|
|
|
self.heartbeat_ttl = 2*60.0
|
|
self.heartbeat_ttl = 2*60.0
|
|
|
|
|
|
|
|
def _log(self, message):
|
|
def _log(self, message):
|
|
@@ -131,25 +121,20 @@ class OrderBookerGCO:
|
|
|
t.instance.keep_alive()
|
|
t.instance.keep_alive()
|
|
|
if t.instance.health_check():
|
|
if t.instance.health_check():
|
|
|
healthy_tasks.append(t)
|
|
healthy_tasks.append(t)
|
|
|
- next_delay = random.randint(55, 65)
|
|
|
|
|
- t.next_remote_ping = now + next_delay
|
|
|
|
|
- self._log(f"🛡️ Task={t.task_ref} keep-alive success. Next ping in {next_delay}s.")
|
|
|
|
|
|
|
+ t.next_remote_ping = now + random.gauss(self.m_cfg.booker.keep_alive, 5)
|
|
|
else:
|
|
else:
|
|
|
dead_tasks.append(t)
|
|
dead_tasks.append(t)
|
|
|
- self._log(f"♻️ Instance for task={t.task_ref} unhealthy.")
|
|
|
|
|
else:
|
|
else:
|
|
|
healthy_tasks.append(t)
|
|
healthy_tasks.append(t)
|
|
|
|
|
|
|
|
if healthy_tasks:
|
|
if healthy_tasks:
|
|
|
new_deadline = time.time() + self.heartbeat_ttl
|
|
new_deadline = time.time() + self.heartbeat_ttl
|
|
|
- mapping = {str(t.task_ref): new_deadline for t in healthy_tasks if t.task_ref is not None}
|
|
|
|
|
|
|
+ mapping = {str(t.task_ref): new_deadline for t in healthy_tasks}
|
|
|
self.redis_client.bulk_zadd(self.m_tracker_key, mapping)
|
|
self.redis_client.bulk_zadd(self.m_tracker_key, mapping)
|
|
|
- # self._log(f"💓 Heartbeat sent. Renewed {len(healthy_tasks)} tasks.")
|
|
|
|
|
|
|
|
|
|
if dead_tasks:
|
|
if dead_tasks:
|
|
|
- mapping = {str(t.task_ref): 0 for t in dead_tasks if t.task_ref is not None}
|
|
|
|
|
|
|
+ mapping = {str(t.task_ref): 0 for t in dead_tasks}
|
|
|
self.redis_client.bulk_zadd(self.m_tracker_key, mapping)
|
|
self.redis_client.bulk_zadd(self.m_tracker_key, mapping)
|
|
|
- self._log(f"🗑️ Handed over {len(dead_tasks)} dead tasks to Sweeper.")
|
|
|
|
|
|
|
|
|
|
if dead_tasks:
|
|
if dead_tasks:
|
|
|
with self.m_lock:
|
|
with self.m_lock:
|
|
@@ -203,10 +188,6 @@ class OrderBookerGCO:
|
|
|
task_id = task.task_ref
|
|
task_id = task.task_ref
|
|
|
task_data = self.m_task_data_cache.get(str(task_id), {})
|
|
task_data = self.m_task_data_cache.get(str(task_id), {})
|
|
|
user_input = task_data.get('user_inputs', {})
|
|
user_input = task_data.get('user_inputs', {})
|
|
|
- expected_start_date = (
|
|
|
|
|
- user_input.get('expected_start_date')
|
|
|
|
|
- or '2000-01-01'
|
|
|
|
|
- )
|
|
|
|
|
expected_end_date = (
|
|
expected_end_date = (
|
|
|
user_input.get('expected_end_date')
|
|
user_input.get('expected_end_date')
|
|
|
or '2100-01-01'
|
|
or '2100-01-01'
|
|
@@ -307,7 +288,6 @@ class OrderBookerGCO:
|
|
|
self._log(f"Failed to update success state to cloud: {e}")
|
|
self._log(f"Failed to update success state to cloud: {e}")
|
|
|
|
|
|
|
|
ThreadPool.getInstance().enqueue(_update_cloud_success)
|
|
ThreadPool.getInstance().enqueue(_update_cloud_success)
|
|
|
-
|
|
|
|
|
self.redis_client.zrem(self.m_tracker_key, task_id)
|
|
self.redis_client.zrem(self.m_tracker_key, task_id)
|
|
|
self._remove_task(task, "booking success")
|
|
self._remove_task(task, "booking success")
|
|
|
else:
|
|
else:
|
|
@@ -322,159 +302,89 @@ class OrderBookerGCO:
|
|
|
]
|
|
]
|
|
|
if any(rate_limited_indicators):
|
|
if any(rate_limited_indicators):
|
|
|
self._remove_task(task, "booking rate limited")
|
|
self._remove_task(task, "booking rate limited")
|
|
|
- if task_data and task_id is not None:
|
|
|
|
|
- task_meta = task_data.get('meta', {})
|
|
|
|
|
- t_fails = task_meta.get('booking_failures', 0) + 1
|
|
|
|
|
- task_meta['booking_failures'] = t_fails
|
|
|
|
|
-
|
|
|
|
|
- def _update_cloud_meta():
|
|
|
|
|
- try:
|
|
|
|
|
- VSCloudApi.Instance().update_vas_task(str(task_id), {"meta": task_meta})
|
|
|
|
|
- except Exception as cloud_err:
|
|
|
|
|
- self._log(f"Failed to update task meta: {cloud_err}")
|
|
|
|
|
- ThreadPool.getInstance().enqueue(_update_cloud_meta)
|
|
|
|
|
-
|
|
|
|
|
- t_cd = self.task_backoff.calculate(t_fails)
|
|
|
|
|
- self._log(f"⏳ Task={task_id} (Booking Attempt {t_fails}) suspended for {t_cd:.1f}s.")
|
|
|
|
|
- self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})
|
|
|
|
|
|
|
|
|
|
def _creator_loop(self):
|
|
def _creator_loop(self):
|
|
|
self._log("Creator loop started.")
|
|
self._log("Creator loop started.")
|
|
|
- spawn_interval = 10.0
|
|
|
|
|
while not self.m_stop_event.is_set():
|
|
while not self.m_stop_event.is_set():
|
|
|
try:
|
|
try:
|
|
|
time.sleep(1)
|
|
time.sleep(1)
|
|
|
- now = time.time()
|
|
|
|
|
for apt in self.m_cfg.appointment_types:
|
|
for apt in self.m_cfg.appointment_types:
|
|
|
r_key = apt.routing_key
|
|
r_key = apt.routing_key
|
|
|
- queue_cd_key = f"vs:queue:cooldown:{r_key}"
|
|
|
|
|
-
|
|
|
|
|
- if self.redis_client.exists(queue_cd_key):
|
|
|
|
|
- continue
|
|
|
|
|
-
|
|
|
|
|
with self.m_lock:
|
|
with self.m_lock:
|
|
|
active = sum(1 for t in self.m_tasks if t.source_queue == r_key)
|
|
active = sum(1 for t in self.m_tasks if t.source_queue == r_key)
|
|
|
- pending = self.m_pending_order_by_queue.get(r_key, 0)
|
|
|
|
|
target = self.m_cfg.booker.target_instances
|
|
target = self.m_cfg.booker.target_instances
|
|
|
-
|
|
|
|
|
- if (active + pending) < target:
|
|
|
|
|
- last_spawn = self.m_last_spawn_times.get(r_key, 0.0)
|
|
|
|
|
- if now - last_spawn >= spawn_interval:
|
|
|
|
|
- self.m_last_spawn_times[r_key] = now
|
|
|
|
|
- self._spawn_worker(r_key)
|
|
|
|
|
|
|
+ if active < target:
|
|
|
|
|
+ self._spawn_worker(r_key)
|
|
|
except Exception as e:
|
|
except Exception as e:
|
|
|
self._log(f'Creator loop exception:{e}')
|
|
self._log(f'Creator loop exception:{e}')
|
|
|
|
|
|
|
|
def _spawn_worker(self, target_routing_key: str):
|
|
def _spawn_worker(self, target_routing_key: str):
|
|
|
- with self.m_lock:
|
|
|
|
|
- self.m_pending_order_by_queue[target_routing_key] = self.m_pending_order_by_queue.get(target_routing_key, 0) + 1
|
|
|
|
|
|
|
+ instance = None
|
|
|
|
|
+ success = False
|
|
|
|
|
+ task_id = None
|
|
|
|
|
+ try:
|
|
|
|
|
+ queue_name = f"auto.{target_routing_key}"
|
|
|
|
|
+ task_data = VSCloudApi.Instance().get_vas_task_pop(queue_name)
|
|
|
|
|
+ if not task_data:
|
|
|
|
|
+ return
|
|
|
|
|
|
|
|
- def _job():
|
|
|
|
|
- success = False
|
|
|
|
|
- task_id = None
|
|
|
|
|
- is_rate_limited = False
|
|
|
|
|
|
|
+ task_id = task_data['id']
|
|
|
|
|
|
|
|
- try:
|
|
|
|
|
- queue_name = f"auto.{target_routing_key}"
|
|
|
|
|
- task_data = VSCloudApi.Instance().get_vas_task_pop(queue_name)
|
|
|
|
|
- if not task_data:
|
|
|
|
|
- return
|
|
|
|
|
-
|
|
|
|
|
- task_id = task_data['id']
|
|
|
|
|
-
|
|
|
|
|
- with self.m_lock:
|
|
|
|
|
- self.m_task_data_cache[str(task_id)] = task_data
|
|
|
|
|
-
|
|
|
|
|
- self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + 5*60.0})
|
|
|
|
|
- user_inputs = task_data.get('user_inputs', {})
|
|
|
|
|
-
|
|
|
|
|
- plg_cfg = VSPlgConfig()
|
|
|
|
|
- plg_cfg.debug = self.m_cfg.debug
|
|
|
|
|
- plg_cfg.free_config = self.m_cfg.free_config
|
|
|
|
|
- plg_cfg.session_max_life = self.m_cfg.session_max_life
|
|
|
|
|
- plg_cfg.account.username = user_inputs.get("username", "")
|
|
|
|
|
- plg_cfg.account.password = user_inputs.get("password", "")
|
|
|
|
|
- if not plg_cfg.account.username:
|
|
|
|
|
- return
|
|
|
|
|
-
|
|
|
|
|
- acceptable_keys = [target_routing_key]
|
|
|
|
|
- if self.m_cfg.need_proxy:
|
|
|
|
|
- proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, self.m_cfg.proxy_cd)
|
|
|
|
|
- plg_cfg.proxy = type(plg_cfg.proxy)(**proxy)
|
|
|
|
|
|
|
+ with self.m_lock:
|
|
|
|
|
+ self.m_task_data_cache[str(task_id)] = task_data
|
|
|
|
|
+
|
|
|
|
|
+ self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + 8*60.0})
|
|
|
|
|
+ user_inputs = task_data.get('user_inputs', {})
|
|
|
|
|
+
|
|
|
|
|
+ plg_cfg = VSPlgConfig()
|
|
|
|
|
+ plg_cfg.debug = self.m_cfg.debug
|
|
|
|
|
+ plg_cfg.free_config = self.m_cfg.free_config
|
|
|
|
|
+ plg_cfg.session_max_life = self.m_cfg.session_max_life
|
|
|
|
|
+ plg_cfg.account.username = user_inputs.get("username", "")
|
|
|
|
|
+ plg_cfg.account.password = user_inputs.get("password", "")
|
|
|
|
|
+ if not plg_cfg.account.username:
|
|
|
|
|
+ return
|
|
|
|
|
+
|
|
|
|
|
+ acceptable_keys = [target_routing_key]
|
|
|
|
|
+ if self.m_cfg.need_proxy:
|
|
|
|
|
+ proxy = VSCloudApi.Instance().get_next_proxy(self.m_cfg.proxy_pool, self.m_cfg.proxy_cd)
|
|
|
|
|
+ plg_cfg.proxy = type(plg_cfg.proxy)(**proxy)
|
|
|
|
|
|
|
|
- instance = self.m_factory.create(self.m_cfg.identifier, self.m_cfg.plugin_config.plugin_name)
|
|
|
|
|
- instance.set_log(self.m_logger)
|
|
|
|
|
- instance.set_config(plg_cfg)
|
|
|
|
|
- instance.create_session()
|
|
|
|
|
-
|
|
|
|
|
- with self.m_lock:
|
|
|
|
|
- self.m_tasks.append(
|
|
|
|
|
- Task(
|
|
|
|
|
- instance=instance,
|
|
|
|
|
- qw_cfg=self.m_cfg.query_wait,
|
|
|
|
|
- next_run=time.time(),
|
|
|
|
|
- task_ref=task_id,
|
|
|
|
|
- acceptable_routing_keys=acceptable_keys,
|
|
|
|
|
- source_queue=target_routing_key,
|
|
|
|
|
- book_allowed=True,
|
|
|
|
|
- next_remote_ping=time.time() + random.randint(55, 65)
|
|
|
|
|
- )
|
|
|
|
|
|
|
+ instance = self.m_factory.create(self.m_cfg.identifier, self.m_cfg.plugin_config.plugin_name)
|
|
|
|
|
+ instance.set_log(self.m_logger)
|
|
|
|
|
+ instance.set_config(plg_cfg)
|
|
|
|
|
+ instance.create_session()
|
|
|
|
|
+
|
|
|
|
|
+ with self.m_lock:
|
|
|
|
|
+ self.m_tasks.append(
|
|
|
|
|
+ Task(
|
|
|
|
|
+ instance=instance,
|
|
|
|
|
+ next_run=time.time(),
|
|
|
|
|
+ task_ref=task_id,
|
|
|
|
|
+ acceptable_routing_keys=acceptable_keys,
|
|
|
|
|
+ source_queue=target_routing_key,
|
|
|
|
|
+ book_allowed=True,
|
|
|
|
|
+ next_remote_ping=time.time() + random.gauss(self.m_cfg.booker.keep_alive, 5)
|
|
|
)
|
|
)
|
|
|
-
|
|
|
|
|
- success = True
|
|
|
|
|
- queue_fail_key = f"vs:queue:failures:{target_routing_key}"
|
|
|
|
|
- self.redis_client.delete(queue_fail_key)
|
|
|
|
|
- self._log(f"+++ Order Booker spawned: {plg_cfg.account.username} (Target: {acceptable_keys})")
|
|
|
|
|
- except Exception as e:
|
|
|
|
|
- err_str = str(e)
|
|
|
|
|
- resource_not_found_indicators = [
|
|
|
|
|
- "40401" in err_str,
|
|
|
|
|
- "Account not found" in err_str,
|
|
|
|
|
- "Proxy not found" in err_str
|
|
|
|
|
- ]
|
|
|
|
|
- if any(resource_not_found_indicators):
|
|
|
|
|
- return
|
|
|
|
|
-
|
|
|
|
|
- self._log(f"Order Booker spawn failed: {e}")
|
|
|
|
|
-
|
|
|
|
|
- rate_limited_indicators = [
|
|
|
|
|
- "42901" in err_str,
|
|
|
|
|
- "Rate limited" in err_str
|
|
|
|
|
- ]
|
|
|
|
|
- if any(rate_limited_indicators):
|
|
|
|
|
- is_rate_limited = True
|
|
|
|
|
- queue_fail_key = f"vs:queue:failures:{target_routing_key}"
|
|
|
|
|
- queue_cd_key = f"vs:queue:cooldown:{target_routing_key}"
|
|
|
|
|
-
|
|
|
|
|
- q_fails = self.redis_client.incr(queue_fail_key)
|
|
|
|
|
- q_cd = self.queue_backoff.calculate(q_fails)
|
|
|
|
|
- self.redis_client.set(queue_cd_key, "1", ex=int(q_cd))
|
|
|
|
|
- self._log(f"📉 [Rate Limited] Queue '{target_routing_key}' failed {q_fails} times. Global Backoff: {q_cd:.1f}s.")
|
|
|
|
|
-
|
|
|
|
|
- if task_id is not None:
|
|
|
|
|
- task_meta = task_data.get('meta') or {}
|
|
|
|
|
- t_fails = task_meta.get('spawn_failures', 0) + 1
|
|
|
|
|
- task_meta['spawn_failures'] = t_fails
|
|
|
|
|
-
|
|
|
|
|
- def _update_cloud_meta():
|
|
|
|
|
- try:
|
|
|
|
|
- VSCloudApi.Instance().update_vas_task(str(task_id), {"meta": task_meta})
|
|
|
|
|
- except Exception as cloud_err:
|
|
|
|
|
- self._log(f"Failed to update task meta: {cloud_err}")
|
|
|
|
|
- ThreadPool.getInstance().enqueue(_update_cloud_meta)
|
|
|
|
|
-
|
|
|
|
|
- t_cd = self.account_backoff.calculate(t_fails)
|
|
|
|
|
- self._log(f"⏳ Task={task_id} (Attempt {t_fails}) suspended for {t_cd:.1f}s.")
|
|
|
|
|
- self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + t_cd})
|
|
|
|
|
-
|
|
|
|
|
- finally:
|
|
|
|
|
- with self.m_lock:
|
|
|
|
|
- self.m_pending_order_by_queue[target_routing_key] = max(0, self.m_pending_order_by_queue[target_routing_key] - 1)
|
|
|
|
|
-
|
|
|
|
|
- if not success and task_id is not None and not is_rate_limited:
|
|
|
|
|
- self.redis_client.zadd(self.m_tracker_key, {str(task_id): 0})
|
|
|
|
|
- self._log(f"♻️ Task={task_id} failed normal spawn. Instantly handed over to Sweeper.")
|
|
|
|
|
|
|
+ )
|
|
|
|
|
+ success = True
|
|
|
|
|
+ self._log(f"+++ Order Booker spawned: {plg_cfg.account.username} (Target: {acceptable_keys})")
|
|
|
|
|
+ except Exception as e:
|
|
|
|
|
+ err_str = str(e)
|
|
|
|
|
+ self._log(f"Order Booker spawn failed: {err_str}")
|
|
|
|
|
+ rate_limited_indicators = [
|
|
|
|
|
+ "42901" in err_str,
|
|
|
|
|
+ "Rate limited" in err_str
|
|
|
|
|
+ ]
|
|
|
|
|
+ if any(rate_limited_indicators):
|
|
|
|
|
+ if task_id is not None:
|
|
|
|
|
+ self.redis_client.zadd(self.m_tracker_key, {str(task_id): time.time() + self.m_cfg.login_backoff})
|
|
|
|
|
+
|
|
|
|
|
+ finally:
|
|
|
|
|
+ if not success:
|
|
|
|
|
+ if task_id:
|
|
|
with self.m_lock:
|
|
with self.m_lock:
|
|
|
self.m_task_data_cache.pop(str(task_id), None)
|
|
self.m_task_data_cache.pop(str(task_id), None)
|
|
|
-
|
|
|
|
|
- ThreadPool.getInstance().enqueue(_job)
|
|
|
|
|
|
|
+ if instance:
|
|
|
|
|
+ instance.cleanup()
|
|
|
|
|
+
|