2922 lines
137 KiB
Python
Executable File
2922 lines
137 KiB
Python
Executable File
import redis
|
||
from redis import ConnectionPool
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
import traceback
|
||
from datetime import datetime
|
||
from typing import Optional, Dict, Any, List, Tuple
|
||
from config import (
|
||
ANALYTICS_ARTIFACT_WAIT_SECONDS,
|
||
ANALYTICS_TRAFFIC_ROOT,
|
||
ANALYTICS_TRAFFIC_ROOT_BLOCK,
|
||
REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_MAX_CONNECTIONS_DISPATCHER,
|
||
WECHAT_TOKENS, WECOM_TOKENS,
|
||
ALERT_WINDOW_SECONDS, ALERT_THRESHOLD, ALERT_COOLDOWN_SECONDS,
|
||
WORKER_REINIT_TIMEOUT, WORKER_INIT_MONITOR_WINDOW,
|
||
RUN_PIPELINE_DEFAULT_OPTIONS,
|
||
AUTO_REBOOT_RECOVERY_DELAY_SECONDS,
|
||
WORKER_INVENTORY,
|
||
MANAGED_WORKER_IDS, TASK_ROUTING_RULES,
|
||
ONLY_MANAGED_WORKERS_CAN_DISPATCH, WORKER_ONLINE_TIMEOUT_SECONDS,
|
||
normalize_worker_id,
|
||
APK_DOWNLOAD_MODE,
|
||
APK_PREFETCH_POLL_INTERVAL,
|
||
APK_DOWNLOAD_QUEUE_INTERVAL,
|
||
)
|
||
from analytics import AnalyticsService, _make_task_key
|
||
from log_manager import logger
|
||
from monitoring import MonitoringService
|
||
from notify_manager import Notifier
|
||
from remote_worker_controller import RemoteWorkerController
|
||
from result_codes import ErrorInfo, ErrorCategory, AppError, BusinessError, InfraError, DownloadError
|
||
|
||
class RedisTaskDispatcher:
|
||
HOT_APP_RETRY_DOWNLOADS_THRESHOLD = 1_000_000
|
||
INFRA_REBOOT_RECOVERY_DELAY_SECONDS = 300
|
||
TASK_QUEUE_NAMES = ("task:queue:high", "task:queue:default", "task:queue:low")
|
||
TASK_QUEUE_CLEANUP_NAMES = (*TASK_QUEUE_NAMES, "task:queue")
|
||
DIRECT_APK_SOURCES = ["google_play", "apkpure"]
|
||
APK_CACHE_SOURCES = ["google_play", "local", "apkpure"]
|
||
# 类级别的连接池（所有实例共享）
|
||
_connection_pool: Optional[ConnectionPool] = None
|
||
_connection_pool_config: Optional[Tuple[str, int, int, int]] = None
|
||
|
||
def __init__(
|
||
self,
|
||
redis_host=REDIS_HOST,
|
||
redis_port=REDIS_PORT,
|
||
redis_db=REDIS_DB,
|
||
max_connections=REDIS_MAX_CONNECTIONS_DISPATCHER,
|
||
):
|
||
"""初始化任务分发器
|
||
|
||
Args:
|
||
redis_host: Redis服务器地址
|
||
redis_port: Redis端口
|
||
redis_db: Redis数据库编号
|
||
max_connections: 连接池最大连接数（需要支持30+台PC并发）
|
||
"""
|
||
# 使用连接池提高性能
|
||
pool_config = (redis_host, redis_port, redis_db, max_connections)
|
||
if (
|
||
RedisTaskDispatcher._connection_pool is None
|
||
or RedisTaskDispatcher._connection_pool_config != pool_config
|
||
):
|
||
RedisTaskDispatcher._connection_pool = ConnectionPool(
|
||
host=redis_host,
|
||
port=redis_port,
|
||
db=redis_db,
|
||
decode_responses=True,
|
||
max_connections=max_connections,
|
||
socket_timeout=30, # 连接超时时间
|
||
socket_connect_timeout=10, # 连接建立超时
|
||
retry_on_timeout=True # 超时时自动重试
|
||
)
|
||
RedisTaskDispatcher._connection_pool_config = pool_config
|
||
self.redis = redis.Redis(connection_pool=RedisTaskDispatcher._connection_pool)
|
||
self.worker_inventory = {item["worker_id"]: dict(item) for item in WORKER_INVENTORY}
|
||
self.managed_worker_ids = set(MANAGED_WORKER_IDS)
|
||
self.only_managed_workers_can_dispatch = ONLY_MANAGED_WORKERS_CAN_DISPATCH
|
||
self.worker_online_timeout = WORKER_ONLINE_TIMEOUT_SECONDS
|
||
self.task_routing_rules = self._normalize_routing_rules(TASK_ROUTING_RULES)
|
||
self.remote_controller = RemoteWorkerController(WORKER_INVENTORY)
|
||
|
||
# 初始化告警推送器
|
||
self.notifier = Notifier(WECHAT_TOKENS, WECOM_TOKENS)
|
||
self.monitor = MonitoringService(self.redis, managed_worker_ids=MANAGED_WORKER_IDS)
|
||
self.analytics = AnalyticsService(
|
||
db_path=self.monitor.repo.db_path,
|
||
artifact_wait_seconds=ANALYTICS_ARTIFACT_WAIT_SECONDS,
|
||
on_snapshot_callback=self._handle_analytics_snapshot,
|
||
traffic_root_block=ANALYTICS_TRAFFIC_ROOT_BLOCK,
|
||
)
|
||
self._apk_registry = None # lazy init
|
||
self._minio_storage = None # lazy init
|
||
|
||
def _normalize_worker_ids(self, worker_ids: Optional[List[str]]) -> List[str]:
|
||
normalized = []
|
||
seen = set()
|
||
for worker_id in worker_ids or []:
|
||
candidate = normalize_worker_id(worker_id)
|
||
if candidate and candidate not in seen:
|
||
seen.add(candidate)
|
||
normalized.append(candidate)
|
||
return normalized
|
||
|
||
def _migrate_worker_runtime_id(self, old_worker_id: str, worker_id: str) -> None:
|
||
if not old_worker_id or not worker_id or old_worker_id == worker_id:
|
||
return
|
||
old_info_json = self.redis.hget("workers:info", old_worker_id)
|
||
if old_info_json and not self.redis.hget("workers:info", worker_id):
|
||
try:
|
||
info = json.loads(old_info_json)
|
||
except (TypeError, ValueError):
|
||
info = {}
|
||
info["worker_id"] = worker_id
|
||
info.setdefault("ip_address", worker_id)
|
||
self.redis.hset("workers:info", worker_id, json.dumps(info))
|
||
if old_info_json:
|
||
self.redis.hdel("workers:info", old_worker_id)
|
||
for set_key in ("workers:idle", "workers:busy"):
|
||
if self.redis.sismember(set_key, old_worker_id):
|
||
self.redis.srem(set_key, old_worker_id)
|
||
self.redis.sadd(set_key, worker_id)
|
||
old_task = self.redis.hget("worker:tasks", old_worker_id)
|
||
if old_task and not self.redis.hget("worker:tasks", worker_id):
|
||
self.redis.hset("worker:tasks", worker_id, old_task)
|
||
if old_task:
|
||
self.redis.hdel("worker:tasks", old_worker_id)
|
||
|
||
@staticmethod
|
||
def _normalize_device_type(raw_device_type: Any) -> str:
|
||
candidate = str(raw_device_type or '').strip().lower()
|
||
if candidate in {'emulator', 'physical'}:
|
||
return candidate
|
||
return ''
|
||
|
||
def _normalize_task_device_type(self, raw_device_type: Any) -> str:
|
||
candidate = str(raw_device_type or '').strip().lower()
|
||
if candidate in {'emulator', 'physical', 'any'}:
|
||
return candidate
|
||
return 'emulator'
|
||
|
||
def _normalize_routing_rules(self, rules: Optional[Dict[str, Dict[str, List[str]]]]) -> Dict[str, Dict[str, List[str]]]:
|
||
normalized = {
|
||
'package_name': {},
|
||
'task_key': {},
|
||
}
|
||
for rule_type in normalized:
|
||
for key, worker_ids in (rules or {}).get(rule_type, {}).items():
|
||
normalized[rule_type][str(key).strip()] = self._normalize_worker_ids(worker_ids)
|
||
return normalized
|
||
|
||
def _resolve_task_targets(self, task_key: str, package_name: str) -> Tuple[List[str], str]:
|
||
task_targets = self.task_routing_rules.get('task_key', {}).get(task_key, [])
|
||
if task_targets:
|
||
return task_targets, 'task_key'
|
||
|
||
package_targets = self.task_routing_rules.get('package_name', {}).get(package_name, [])
|
||
if package_targets:
|
||
return package_targets, 'package_name'
|
||
|
||
return [], 'public'
|
||
|
||
def _normalize_country_codes(self, raw_country_code: Any) -> List[str]:
|
||
if isinstance(raw_country_code, (list, tuple)):
|
||
raw_items = raw_country_code
|
||
else:
|
||
raw_items = str(raw_country_code or '').split(',')
|
||
country_codes = ['US']
|
||
seen = {'US'}
|
||
for item in raw_items:
|
||
candidate = str(item or '').strip().upper()
|
||
if not candidate or candidate in seen:
|
||
continue
|
||
seen.add(candidate)
|
||
country_codes.append(candidate)
|
||
return country_codes
|
||
|
||
def _build_task_info_from_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
||
normalized_payload = dict(payload or {})
|
||
app_name = str(normalized_payload.get('app_name') or '').strip()
|
||
package_name = str(normalized_payload.get('package_name') or '').strip()
|
||
task_key = str(normalized_payload.get('task_key') or _make_task_key(app_name, package_name)).strip()
|
||
target_worker_ids, routing_mode = self._resolve_task_targets(task_key, package_name)
|
||
country_code = str(normalized_payload.get('country_code', '') or '').strip()
|
||
country_codes = normalized_payload.get('country_codes') or country_code
|
||
original_row = normalized_payload.get('original_row') or {}
|
||
original_app_magic_label = original_row.get('app_magic_label') if isinstance(original_row, dict) else ''
|
||
original_last_updated = original_row.get('last_updated') if isinstance(original_row, dict) else ''
|
||
return {
|
||
'app_name': app_name,
|
||
'package_name': package_name,
|
||
'country_code': country_code,
|
||
'country_codes': self._normalize_country_codes(country_codes),
|
||
'app_magic_label': str(normalized_payload.get('app_magic_label') or original_app_magic_label or '').strip(),
|
||
'last_updated': str(normalized_payload.get('last_updated') or original_last_updated or '').strip(),
|
||
'collection_task_type': str(normalized_payload.get('collection_task_type') or 'new_app').strip() or 'new_app',
|
||
'last_update_interval_days': int(normalized_payload.get('last_update_interval_days') or 0),
|
||
'downloads': int(normalized_payload.get('downloads') or 0),
|
||
'device_type': self._normalize_task_device_type(normalized_payload.get('device_type', '')),
|
||
'available_sources': normalized_payload.get('available_sources') or ['google_play', 'local'],
|
||
'original_row': original_row,
|
||
'target_worker_ids': target_worker_ids,
|
||
'routing_mode': routing_mode,
|
||
}
|
||
|
||
@staticmethod
|
||
def _normalize_task_queue_label(raw_queue: Any, default: str = "default") -> str:
|
||
candidate = str(raw_queue or default).strip().lower()
|
||
if candidate in {"high", "default", "low"}:
|
||
return candidate
|
||
return default
|
||
|
||
def _should_auto_reroute_to_physical(self, task_details: Dict[str, Any], error_info: ErrorInfo) -> bool:
|
||
current_device_type = self._normalize_task_device_type(task_details.get('device_type', ''))
|
||
return (
|
||
current_device_type != 'physical'
|
||
and error_info.category == ErrorCategory.APP_ERROR
|
||
and error_info.code in {AppError.CRASH, AppError.ROOT_MODE_UNSUPPORTED}
|
||
)
|
||
|
||
def _build_physical_reroute_task_details(self, task_details: Dict[str, Any]) -> Dict[str, Any]:
|
||
rerouted_task_details = dict(task_details or {})
|
||
rerouted_task_details['device_type'] = 'physical'
|
||
original_row = rerouted_task_details.get('original_row')
|
||
if isinstance(original_row, dict):
|
||
original_row_copy = dict(original_row)
|
||
original_row_copy['device_type'] = 'physical'
|
||
rerouted_task_details['original_row'] = original_row_copy
|
||
return rerouted_task_details
|
||
|
||
def _is_task_queued(self, task_key: str) -> bool:
|
||
try:
|
||
for queue_name in self.TASK_QUEUE_NAMES:
|
||
if self.redis.lpos(queue_name, task_key) is not None:
|
||
return True
|
||
return False
|
||
except Exception:
|
||
for queue_name in self.TASK_QUEUE_NAMES:
|
||
if task_key in set(self.redis.lrange(queue_name, 0, -1)):
|
||
return True
|
||
return False
|
||
|
||
def _remove_task_from_queue(self, task_key: str) -> int:
|
||
removed = 0
|
||
try:
|
||
for queue_name in self.TASK_QUEUE_CLEANUP_NAMES:
|
||
removed += int(self.redis.lrem(queue_name, 0, task_key) or 0)
|
||
except Exception:
|
||
pass
|
||
return removed
|
||
|
||
def _get_queued_task_keys(self) -> List[str]:
|
||
task_keys = []
|
||
seen = set()
|
||
for queue_name in self.TASK_QUEUE_NAMES:
|
||
for task_key in reversed(self.redis.lrange(queue_name, 0, -1)):
|
||
if task_key in seen:
|
||
continue
|
||
seen.add(task_key)
|
||
task_keys.append(task_key)
|
||
return task_keys
|
||
|
||
def get_pending_queue_length(self) -> int:
|
||
return sum(int(self.redis.llen(queue_name) or 0) for queue_name in self.TASK_QUEUE_NAMES)
|
||
|
||
def _is_task_running(self, task_key: str) -> bool:
|
||
status_json = self.redis.hget("task:status", task_key)
|
||
if status_json:
|
||
try:
|
||
status = json.loads(status_json)
|
||
if status.get("status") == "running":
|
||
return True
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
return task_key in set(self.redis.hvals("worker:tasks"))
|
||
|
||
def _mark_task_pending_for_queue(self, task_key: str) -> None:
|
||
status_json = self.redis.hget("task:status", task_key)
|
||
retry_count = 0
|
||
if status_json:
|
||
try:
|
||
retry_count = int((json.loads(status_json) or {}).get("retry_count") or 0)
|
||
except (json.JSONDecodeError, TypeError, ValueError):
|
||
retry_count = 0
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "pending",
|
||
"end_time": time.time(),
|
||
"retry_count": retry_count,
|
||
}))
|
||
|
||
def _is_hot_new_app_task(self, task_details: Dict[str, Any]) -> bool:
|
||
if str(task_details.get('collection_task_type') or 'new_app').strip() != 'new_app':
|
||
return False
|
||
try:
|
||
downloads = int(task_details.get('downloads') or 0)
|
||
except (TypeError, ValueError):
|
||
downloads = 0
|
||
return downloads >= self.HOT_APP_RETRY_DOWNLOADS_THRESHOLD
|
||
|
||
def _mark_retry_avoid_worker(self, task_key: str, task_details: Dict[str, Any], worker_id: str) -> Dict[str, Any]:
|
||
next_details = dict(task_details or {})
|
||
excluded = [
|
||
str(item or '').strip()
|
||
for item in (next_details.get('excluded_worker_ids') or [])
|
||
if str(item or '').strip()
|
||
]
|
||
if worker_id and worker_id not in excluded:
|
||
excluded.append(worker_id)
|
||
next_details['excluded_worker_ids'] = excluded[-8:]
|
||
next_details['last_failed_worker_id'] = worker_id
|
||
self.redis.hset("task:details", task_key, json.dumps(next_details))
|
||
return next_details
|
||
|
||
def _queue_retry_task(self, task_key: str, task_details: Dict[str, Any], *, prioritize: bool = False):
|
||
self._remove_task_from_queue(task_key)
|
||
queue_name = self._resolve_task_queue(task_details)
|
||
if prioritize:
|
||
self.redis.rpush(queue_name, task_key)
|
||
else:
|
||
self.redis.lpush(queue_name, task_key)
|
||
|
||
def _resolve_task_queue(self, task_details: Dict[str, Any]) -> str:
|
||
task_queue = self._normalize_task_queue_label(task_details.get("task_queue"))
|
||
if task_queue == "low":
|
||
return "task:queue:low"
|
||
if task_queue == "high":
|
||
return "task:queue:high"
|
||
return "task:queue:default"
|
||
|
||
def _handle_analytics_snapshot(self, package_name: str, summary: Dict[str, Any]) -> None:
|
||
payload = dict(summary.get("task_payload") or {})
|
||
app_name = str(payload.get("app_name") or summary.get("app_name") or package_name).strip() or package_name
|
||
task_key = str(summary.get("latest_task_key") or payload.get("task_key") or _make_task_key(app_name, package_name)).strip()
|
||
if not task_key:
|
||
return
|
||
existing_task_details_json = self.redis.hget("task:details", task_key)
|
||
existing_task_details = json.loads(existing_task_details_json) if existing_task_details_json else {}
|
||
if existing_task_details.get("is_block_task") or task_key.endswith("_block"):
|
||
return
|
||
if existing_task_details.get("task_queue") == "low":
|
||
if summary.get("model_eligible"):
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "pending",
|
||
"end_time": time.time(),
|
||
"retry_count": 0,
|
||
}))
|
||
self.redis.lpush("task:queue:low", task_key)
|
||
logger.info(f"[模型流量] 任务 {task_key} 继续留在低优队列 (model_flow={summary.get('model_flow_count', 0)})")
|
||
else:
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.sadd("task:completed", task_key)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "completed",
|
||
"end_time": time.time(),
|
||
}))
|
||
logger.info(f"[模型流量] 任务 {task_key} 已完成，退出低优队列 (model_flow={summary.get('model_flow_count', 0)})")
|
||
return
|
||
collection_status = str(summary.get("collection_status") or "pending").strip() or "pending"
|
||
task_info = self._build_task_info_from_payload(
|
||
payload or {
|
||
"app_name": app_name,
|
||
"package_name": package_name,
|
||
"country_code": summary.get("country_code", ""),
|
||
"device_type": summary.get("device_type", ""),
|
||
}
|
||
)
|
||
if existing_task_details.get("excluded_worker_ids"):
|
||
task_info["excluded_worker_ids"] = existing_task_details.get("excluded_worker_ids") or []
|
||
task_info["last_failed_worker_id"] = existing_task_details.get("last_failed_worker_id", "")
|
||
existing_queue = self._normalize_task_queue_label(existing_task_details.get("task_queue"))
|
||
task_info["task_queue"] = existing_queue if existing_queue in {"high", "default"} else "default"
|
||
self.redis.hset("task:details", task_key, json.dumps(task_info))
|
||
prev_status_json = self.redis.hget("task:status", task_key)
|
||
task_status = json.loads(prev_status_json) if prev_status_json else {}
|
||
retry_count = int(task_status.get("retry_count") or 0)
|
||
if collection_status == "qualified":
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
self.redis.sadd("task:completed", task_key)
|
||
task_status.update({
|
||
"status": "qualified",
|
||
"end_time": time.time(),
|
||
"retry_count": retry_count,
|
||
})
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
return
|
||
if collection_status == "failed_terminal":
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.sadd("task:failed", task_key)
|
||
task_status.update({
|
||
"status": "failed",
|
||
"end_time": time.time(),
|
||
"retry_count": retry_count,
|
||
})
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
return
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
if task_status.get("status") != "running":
|
||
task_status.update({
|
||
"status": "pending",
|
||
"end_time": task_status.get("end_time", time.time()),
|
||
"retry_count": retry_count,
|
||
})
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
if not self._is_task_queued(task_key):
|
||
self._queue_retry_task(task_key, task_info, prioritize=self._is_hot_new_app_task(task_info))
|
||
|
||
def _task_to_payload(self, task_key: str, task_details: Dict[str, Any]) -> Dict[str, Any]:
|
||
task_details = self._enrich_task_with_apk(task_details)
|
||
payload = {
|
||
'task_key': task_key,
|
||
'app_name': task_details['app_name'],
|
||
'package_name': task_details['package_name'],
|
||
'country_code': task_details.get('country_code', ''),
|
||
'country_codes': self._normalize_country_codes(task_details.get('country_codes') or task_details.get('country_code', '')),
|
||
'app_magic_label': str(task_details.get('app_magic_label') or '').strip(),
|
||
'last_updated': str(task_details.get('last_updated') or '').strip(),
|
||
'collection_task_type': str(task_details.get('collection_task_type') or 'new_app').strip() or 'new_app',
|
||
'last_update_interval_days': int(task_details.get('last_update_interval_days') or 0),
|
||
'device_type': self._normalize_task_device_type(task_details.get('device_type', '')),
|
||
'available_sources': task_details.get('available_sources') or self._default_available_sources(),
|
||
'local_apk_dir': task_details.get('local_apk_dir', ''),
|
||
'local_apk_files': task_details.get('local_apk_files', []),
|
||
'target_worker_ids': task_details.get('target_worker_ids', []),
|
||
'routing_mode': task_details.get('routing_mode', 'public'),
|
||
'keep_app_installed': bool(task_details.get('keep_app_installed')),
|
||
'blocked': bool(task_details.get('is_block_task', False)),
|
||
'traffic_root': ANALYTICS_TRAFFIC_ROOT_BLOCK if task_details.get('is_block_task') else ANALYTICS_TRAFFIC_ROOT,
|
||
}
|
||
if payload.get('local_apk_dir'):
|
||
logger.info("[下发] 任务 %s 附带本地APK: dir=%s files=%s",
|
||
task_key, payload['local_apk_dir'], ', '.join(f.get('filename', '') for f in payload['local_apk_files']))
|
||
elif payload.get('available_sources') and 'local' in payload['available_sources']:
|
||
logger.warning("[下发] 任务 %s 标记了local源但无本地APK路径 (available_sources=%s)",
|
||
task_key, payload['available_sources'])
|
||
return payload
|
||
|
||
@staticmethod
|
||
def _apk_download_mode() -> str:
|
||
return str(APK_DOWNLOAD_MODE or "direct").strip().lower()
|
||
|
||
@classmethod
|
||
def _uses_central_apk_cache(cls) -> bool:
|
||
return cls._apk_download_mode() in {"minio", "smb"}
|
||
|
||
@classmethod
|
||
def _default_available_sources(cls) -> List[str]:
|
||
if cls._uses_central_apk_cache():
|
||
return list(cls.APK_CACHE_SOURCES)
|
||
return list(cls.DIRECT_APK_SOURCES)
|
||
|
||
def _get_worker_info(self, worker_id: str) -> Optional[Dict[str, Any]]:
|
||
worker_id = normalize_worker_id(worker_id)
|
||
worker_info_json = self.redis.hget("workers:info", worker_id)
|
||
if not worker_info_json:
|
||
return None
|
||
return json.loads(worker_info_json)
|
||
|
||
def _get_worker_ip(self, worker_id: str) -> str:
|
||
worker_id = normalize_worker_id(worker_id)
|
||
worker_info = self._get_worker_info(worker_id)
|
||
if worker_info:
|
||
return worker_info.get('ip_address', '')
|
||
inventory = self.worker_inventory.get(worker_id)
|
||
if inventory:
|
||
return inventory["ssh_target"].split("@")[-1]
|
||
return worker_id.split('_', 1)[0] if '_' in worker_id else ''
|
||
|
||
def _workers_meta_key(self) -> str:
|
||
return "workers:meta"
|
||
|
||
def _worker_recovery_observe_key(self, worker_id: str) -> str:
|
||
return f"worker:recovery_observe:{worker_id}"
|
||
|
||
def _worker_recovery_pending_key(self, worker_id: str) -> str:
|
||
return f"worker:recovery_pending:{worker_id}"
|
||
|
||
def _default_worker_meta(self, worker_id: str) -> Dict[str, Any]:
|
||
allow_dispatch = (not self.only_managed_workers_can_dispatch) or self._is_managed_worker_id(worker_id)
|
||
return {
|
||
'policy': 'enabled' if allow_dispatch else 'disabled',
|
||
'auto_recover_used': False,
|
||
'last_action': '',
|
||
'last_action_status': '',
|
||
'last_action_message': '',
|
||
'last_action_at': 0.0,
|
||
}
|
||
|
||
def _get_worker_meta(self, worker_id: str) -> Dict[str, Any]:
|
||
payload = self.redis.hget(self._workers_meta_key(), worker_id)
|
||
if not payload:
|
||
return self._default_worker_meta(worker_id)
|
||
try:
|
||
meta = json.loads(payload)
|
||
except (TypeError, ValueError):
|
||
return self._default_worker_meta(worker_id)
|
||
if not isinstance(meta, dict):
|
||
return self._default_worker_meta(worker_id)
|
||
normalized = self._default_worker_meta(worker_id)
|
||
normalized.update(meta)
|
||
normalized['auto_recover_used'] = bool(normalized.get('auto_recover_used'))
|
||
return normalized
|
||
|
||
def _set_worker_meta(self, worker_id: str, meta: Dict[str, Any]) -> Dict[str, Any]:
|
||
normalized = self._default_worker_meta(worker_id)
|
||
normalized.update(meta)
|
||
self.redis.hset(self._workers_meta_key(), worker_id, json.dumps(normalized))
|
||
return normalized
|
||
|
||
def _update_worker_meta(self, worker_id: str, **updates: Any) -> Dict[str, Any]:
|
||
meta = self._get_worker_meta(worker_id)
|
||
meta.update(updates)
|
||
return self._set_worker_meta(worker_id, meta)
|
||
|
||
def _mark_worker_action(self, worker_id: str, action: str, result: Dict[str, Any]) -> Dict[str, Any]:
|
||
return self._update_worker_meta(
|
||
worker_id,
|
||
last_action=action,
|
||
last_action_status='success' if result.get('ok') else 'failed',
|
||
last_action_message=result.get('message', ''),
|
||
last_action_at=time.time(),
|
||
)
|
||
|
||
def _set_worker_policy(self, worker_id: str, policy: str) -> Dict[str, Any]:
|
||
return self._update_worker_meta(worker_id, policy=policy)
|
||
|
||
@staticmethod
|
||
def _is_worker_recovery_action(action: str) -> bool:
|
||
return action in {'recover_and_start', 'run_pipeline'}
|
||
|
||
def _schedule_post_reboot_pipeline(self, worker_id: str, delay_seconds: int = AUTO_REBOOT_RECOVERY_DELAY_SECONDS) -> None:
|
||
thread = threading.Thread(
|
||
target=self._run_post_reboot_pipeline_after_delay,
|
||
args=(worker_id, delay_seconds),
|
||
name=f"post-reboot-pipeline-{worker_id}",
|
||
daemon=True,
|
||
)
|
||
thread.start()
|
||
|
||
def _run_post_reboot_pipeline_after_delay(self, worker_id: str, delay_seconds: int = AUTO_REBOOT_RECOVERY_DELAY_SECONDS) -> None:
|
||
pending_key = self._worker_recovery_pending_key(worker_id)
|
||
observe_key = self._worker_recovery_observe_key(worker_id)
|
||
try:
|
||
if delay_seconds > 0:
|
||
logger.info(
|
||
"[Worker自动恢复] worker=%s 主机重启后等待%s秒，再执行默认恢复 pipeline",
|
||
worker_id,
|
||
delay_seconds,
|
||
)
|
||
time.sleep(delay_seconds)
|
||
|
||
self.redis.delete(observe_key)
|
||
self.redis.delete(pending_key)
|
||
pipeline_result = self.run_dashboard_action([worker_id], 'run_pipeline')[0]
|
||
self._mark_worker_action(worker_id, 'run_pipeline', pipeline_result)
|
||
|
||
if pipeline_result.get('ok'):
|
||
self.redis.setex(pending_key, WORKER_REINIT_TIMEOUT, str(time.time()))
|
||
self._update_worker_meta(worker_id, auto_recover_used=True)
|
||
logger.warning(f"[Worker自动恢复] Worker {worker_id} 主机重启后已执行默认恢复 pipeline")
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker主机重启后自动恢复】\nWorker: {worker_id}\n已在主机重启后执行 dashboard 默认恢复链路，等待 Worker 重新 init"
|
||
)
|
||
return
|
||
|
||
self.redis.delete(pending_key)
|
||
self._update_worker_meta(worker_id, auto_recover_used=False)
|
||
logger.error(f"[Worker自动恢复失败] Worker {worker_id} 主机重启后执行默认恢复 pipeline 失败")
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker主机重启后自动恢复失败】\nWorker: {worker_id}\n恢复结果: {pipeline_result.get('message', '')}"
|
||
)
|
||
except Exception:
|
||
detail = traceback.format_exc().strip()
|
||
self.redis.delete(pending_key)
|
||
self._update_worker_meta(worker_id, auto_recover_used=False)
|
||
self._mark_worker_action(
|
||
worker_id,
|
||
'run_pipeline',
|
||
{
|
||
'ok': False,
|
||
'message': detail,
|
||
'stderr': detail,
|
||
},
|
||
)
|
||
logger.exception(f"[Worker自动恢复失败] Worker {worker_id} 主机重启后恢复线程异常")
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker主机重启后自动恢复异常】\nWorker: {worker_id}\n异常信息: {detail}"
|
||
)
|
||
|
||
def _get_control_display_status(self, worker_id: str, current_worker: Optional[Dict[str, Any]] = None) -> str:
|
||
meta = self._get_worker_meta(worker_id)
|
||
if meta.get('policy') == 'disabled':
|
||
return 'disabled'
|
||
if meta.get('policy') == 'drain':
|
||
return 'drain'
|
||
if current_worker:
|
||
if not current_worker.get('online'):
|
||
return 'offline'
|
||
if current_worker.get('state') == 'disabled':
|
||
return 'disabled'
|
||
if current_worker.get('status') == 'busy':
|
||
return 'running'
|
||
return 'idle'
|
||
return 'offline'
|
||
|
||
def _run_worker_action(
|
||
self,
|
||
worker_id: str,
|
||
action: str,
|
||
options: Optional[Dict[str, Any]] = None,
|
||
progress_callback=None,
|
||
) -> Dict[str, Any]:
|
||
logger.info(
|
||
"[DashboardAction] start worker=%s action=%s options=%s",
|
||
worker_id,
|
||
action,
|
||
json.dumps(options or {}, ensure_ascii=False, sort_keys=True),
|
||
)
|
||
result = self.remote_controller.run_action(worker_id, action, options=options, progress_callback=progress_callback)
|
||
self._mark_worker_action(worker_id, action, result)
|
||
logger.info(
|
||
"[DashboardAction] finish worker=%s action=%s ok=%s failed_step=%s stage=%s message=%s",
|
||
worker_id,
|
||
action,
|
||
result.get('ok'),
|
||
result.get('failed_step', ''),
|
||
result.get('stage', ''),
|
||
str(result.get('message', '')).replace('\r', ' ').replace('\n', ' | '),
|
||
)
|
||
return result
|
||
|
||
def get_worker_control_rows(
|
||
self,
|
||
date_str: Optional[str] = None,
|
||
worker_id: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
filter_worker_id = worker_id or ""
|
||
monitor_rows = {
|
||
row["worker_id"]: row
|
||
for row in self.monitor.get_worker_stats(date_str=date_str, dispatcher=self)
|
||
}
|
||
current_workers = {worker["worker_id"]: worker for worker in self.get_dashboard_workers()}
|
||
worker_ids = list(self.managed_worker_ids) or list(current_workers.keys())
|
||
rows: List[Dict[str, Any]] = []
|
||
for current_worker_id in sorted(set(worker_ids) | set(monitor_rows) | set(current_workers)):
|
||
if current_worker_id not in self.worker_inventory:
|
||
continue
|
||
if filter_worker_id and current_worker_id != filter_worker_id:
|
||
continue
|
||
inventory = self.worker_inventory.get(current_worker_id, {})
|
||
monitor_row = monitor_rows.get(current_worker_id, {})
|
||
current_worker = current_workers.get(current_worker_id, {})
|
||
meta = self._get_worker_meta(current_worker_id)
|
||
current_status = current_worker.get('state', 'offline')
|
||
current_status = {
|
||
'idle_waiting_task': 'idle',
|
||
'running_download': 'download',
|
||
'running_collect': 'collect',
|
||
}.get(current_status, current_status)
|
||
rows.append(
|
||
{
|
||
'worker_id': current_worker_id,
|
||
'ip_address': current_worker.get('ip_address') or monitor_row.get('ip_address') or self._get_worker_ip(current_worker_id),
|
||
'device_type': self._normalize_device_type(current_worker.get('device_type', '')),
|
||
'status': monitor_row.get('status', current_status),
|
||
'control_status': self._get_control_display_status(current_worker_id, current_worker),
|
||
'policy': meta.get('policy', 'enabled'),
|
||
'dispatch_enabled': self._worker_can_dispatch(current_worker_id),
|
||
'online': bool(current_worker.get('online')),
|
||
'task_count': monitor_row.get('task_count', 0),
|
||
'avg_download_duration_seconds': monitor_row.get('avg_download_duration_seconds', 0.0),
|
||
'avg_collect_duration_seconds': monitor_row.get('avg_collect_duration_seconds', 0.0),
|
||
'running_seconds': monitor_row.get('running_seconds', 0.0),
|
||
'idle_waiting_seconds': monitor_row.get('idle_waiting_seconds', 0.0),
|
||
'failed_count': monitor_row.get('failed_count', 0),
|
||
'last_task_at': monitor_row.get('last_task_at', '--'),
|
||
'current_task': current_worker.get('current_task'),
|
||
'last_action': meta.get('last_action', ''),
|
||
'last_action_status': meta.get('last_action_status', ''),
|
||
'last_action_message': meta.get('last_action_message', ''),
|
||
'last_action_at': meta.get('last_action_at', 0.0),
|
||
'tags': inventory.get('tags', []),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
def _is_physical_worker(self, worker_id: str) -> bool:
|
||
worker_info = self._get_worker_info(worker_id)
|
||
if not worker_info:
|
||
return False
|
||
return self._normalize_device_type(worker_info.get("device_type", "")) == "physical"
|
||
|
||
def run_dashboard_action(
|
||
self,
|
||
worker_ids: List[str],
|
||
action: str,
|
||
options: Optional[Dict[str, Any]] = None,
|
||
progress_callback=None,
|
||
) -> List[Dict[str, Any]]:
|
||
results = []
|
||
if action == 'force_reapk':
|
||
pkg = str((options or {}).get("package_name", "")).strip()
|
||
if not pkg:
|
||
return [{"worker_id": "", "action": "force_reapk", "ok": False, "message": "package_name is required"}]
|
||
if not self._uses_central_apk_cache():
|
||
return [{
|
||
"worker_id": "",
|
||
"action": "force_reapk",
|
||
"ok": False,
|
||
"message": "当前为 direct 模式，未启用中心 APK 缓存，无需 force_reapk",
|
||
}]
|
||
n = self.force_reapk(pkg)
|
||
return [{"worker_id": "", "action": "force_reapk", "ok": True, "message": f"已重置 {pkg} ({n} 个任务)，等待中心 APK 缓存刷新"}]
|
||
|
||
for worker_id in worker_ids:
|
||
try:
|
||
if action == 'enable':
|
||
self.redis.delete(self._worker_recovery_observe_key(worker_id))
|
||
self.redis.delete(self._worker_recovery_pending_key(worker_id))
|
||
meta = self._update_worker_meta(worker_id, policy='enabled', auto_recover_used=False)
|
||
worker_info = self._get_worker_info(worker_id)
|
||
if self._get_worker_online_state(worker_info):
|
||
self.monitor.set_worker_state(worker_id, 'idle_waiting_task', reason='manual_enable')
|
||
meta['ok'] = True
|
||
meta['message'] = 'worker enabled'
|
||
result = self._mark_worker_action(worker_id, 'enable', meta)
|
||
result.update({'worker_id': worker_id, 'action': 'enable', 'ok': True, 'message': 'worker enabled'})
|
||
elif action == 'disable':
|
||
self.redis.delete(self._worker_recovery_observe_key(worker_id))
|
||
self.redis.delete(self._worker_recovery_pending_key(worker_id))
|
||
self.monitor.mark_worker_disabled(worker_id, 'manual_disable')
|
||
meta = self._set_worker_policy(worker_id, 'disabled')
|
||
meta['ok'] = True
|
||
meta['message'] = 'worker disabled'
|
||
result = self._mark_worker_action(worker_id, 'disable', meta)
|
||
result.update({'worker_id': worker_id, 'action': 'disable', 'ok': True, 'message': 'worker disabled'})
|
||
elif action == 'drain':
|
||
meta = self._set_worker_policy(worker_id, 'drain')
|
||
meta['ok'] = True
|
||
meta['message'] = 'worker drain enabled'
|
||
result = self._mark_worker_action(worker_id, 'drain', meta)
|
||
result.update({'worker_id': worker_id, 'action': 'drain', 'ok': True, 'message': 'worker drain enabled'})
|
||
elif action == 'start':
|
||
self._set_worker_policy(worker_id, 'enabled')
|
||
pcap_result = self._run_worker_action(worker_id, 'pull_pcap_files', progress_callback=progress_callback)
|
||
if not pcap_result.get('ok'):
|
||
self._mark_worker_action(worker_id, 'start', pcap_result)
|
||
results.append(pcap_result)
|
||
continue
|
||
result = self._run_worker_action(worker_id, 'start_worker', progress_callback=progress_callback)
|
||
self._mark_worker_action(worker_id, 'start', result)
|
||
elif action == 'run_pipeline':
|
||
pipeline_options = dict(RUN_PIPELINE_DEFAULT_OPTIONS)
|
||
if options:
|
||
pipeline_options.update(options)
|
||
if self._is_physical_worker(worker_id):
|
||
pipeline_options.pop("recover_mumu", None)
|
||
pipeline_options.pop("restart_mumu", None)
|
||
if pipeline_options.get('start_worker'):
|
||
self._set_worker_policy(worker_id, 'enabled')
|
||
result = self._run_worker_action(worker_id, 'run_pipeline', options=pipeline_options, progress_callback=progress_callback)
|
||
elif action == 'stop_worker':
|
||
result = self._run_worker_action(worker_id, 'stop_worker', progress_callback=progress_callback)
|
||
elif action == 'recover_and_start':
|
||
self._set_worker_policy(worker_id, 'enabled')
|
||
self._run_worker_action(worker_id, 'stop_worker', progress_callback=progress_callback)
|
||
pcap_result = self._run_worker_action(worker_id, 'pull_pcap_files', progress_callback=progress_callback)
|
||
if not pcap_result.get('ok'):
|
||
self._mark_worker_action(worker_id, 'recover_and_start', pcap_result)
|
||
results.append(pcap_result)
|
||
continue
|
||
if self._is_physical_worker(worker_id):
|
||
result = self._run_worker_action(worker_id, 'start_worker', progress_callback=progress_callback)
|
||
self._mark_worker_action(worker_id, 'recover_and_start', result)
|
||
results.append(result)
|
||
continue
|
||
recover_result = self._run_worker_action(worker_id, 'recover_mumu', progress_callback=progress_callback)
|
||
if not recover_result.get('ok'):
|
||
self._mark_worker_action(worker_id, 'recover_and_start', recover_result)
|
||
results.append(recover_result)
|
||
continue
|
||
restart_result = self._run_worker_action(worker_id, 'restart_mumu', progress_callback=progress_callback)
|
||
if not restart_result.get('ok'):
|
||
self._mark_worker_action(worker_id, 'recover_and_start', restart_result)
|
||
results.append(restart_result)
|
||
continue
|
||
self._run_worker_action(worker_id, 'configure_mumu_network', progress_callback=progress_callback)
|
||
self._run_worker_action(worker_id, 'fix_adb_connection', progress_callback=progress_callback)
|
||
result = self._run_worker_action(worker_id, 'start_worker', progress_callback=progress_callback)
|
||
self._mark_worker_action(worker_id, 'recover_and_start', result)
|
||
elif action == 'clone':
|
||
result = self._run_worker_action(worker_id, 'clone', progress_callback=progress_callback)
|
||
elif action == 'pull':
|
||
result = self._run_worker_action(worker_id, 'pull', progress_callback=progress_callback)
|
||
elif action == 'pull_pcap_files':
|
||
result = self._run_worker_action(worker_id, 'pull_pcap_files', progress_callback=progress_callback)
|
||
elif action == 'setup':
|
||
result = self._run_worker_action(worker_id, 'setup', progress_callback=progress_callback)
|
||
elif action == 'restart_mumu':
|
||
if self._is_physical_worker(worker_id):
|
||
result = {"worker_id": worker_id, "action": "restart_mumu", "ok": False, "message": "真机设备，跳过重启MuMu"}
|
||
results.append(result)
|
||
continue
|
||
result = self._run_worker_action(worker_id, 'restart_mumu', progress_callback=progress_callback)
|
||
elif action == 'recover_mumu':
|
||
if self._is_physical_worker(worker_id):
|
||
result = {"worker_id": worker_id, "action": "recover_mumu", "ok": False, "message": "真机设备，跳过MuMu镜像恢复"}
|
||
results.append(result)
|
||
continue
|
||
result = self._run_worker_action(worker_id, 'recover_mumu', progress_callback=progress_callback)
|
||
elif action == 'reboot':
|
||
result = self._run_worker_action(worker_id, 'reboot', progress_callback=progress_callback)
|
||
elif action == 'status':
|
||
result = self._run_worker_action(worker_id, 'status', progress_callback=progress_callback)
|
||
elif action == 'execute_command':
|
||
result = self._run_worker_action(worker_id, 'execute_command', options=options, progress_callback=progress_callback)
|
||
elif action == 'fix_adb_connection':
|
||
result = self._run_worker_action(worker_id, 'fix_adb_connection', progress_callback=progress_callback)
|
||
elif action == 'configure_mumu_network':
|
||
result = self._run_worker_action(worker_id, 'configure_mumu_network', progress_callback=progress_callback)
|
||
elif action == 'recover_mumu_full':
|
||
if self._is_physical_worker(worker_id):
|
||
result = {"worker_id": worker_id, "action": "recover_mumu_full", "ok": False, "message": "真机设备，跳过MuMu恢复"}
|
||
results.append(result)
|
||
continue
|
||
result = self._run_worker_action(worker_id, 'recover_mumu_full', progress_callback=progress_callback)
|
||
else:
|
||
raise ValueError(f'unsupported action: {action}')
|
||
except Exception as e:
|
||
detail = traceback.format_exc().strip()
|
||
result = {
|
||
'worker_id': worker_id,
|
||
'action': action,
|
||
'ok': False,
|
||
'message': detail or str(e),
|
||
'stderr': detail,
|
||
}
|
||
self._mark_worker_action(worker_id, action, result)
|
||
results.append(result)
|
||
return results
|
||
|
||
def check_recovery_windows(self) -> List[str]:
|
||
reset_workers = []
|
||
for worker_id in self.managed_worker_ids:
|
||
meta = self._get_worker_meta(worker_id)
|
||
if not meta.get('auto_recover_used'):
|
||
continue
|
||
if self.redis.exists(self._worker_recovery_pending_key(worker_id)):
|
||
continue
|
||
observe_key = self._worker_recovery_observe_key(worker_id)
|
||
if self.redis.exists(observe_key):
|
||
continue
|
||
if self._is_worker_recovery_action(meta.get('last_action', '')) and meta.get('last_action_status') == 'success':
|
||
self._update_worker_meta(worker_id, auto_recover_used=False)
|
||
reset_workers.append(worker_id)
|
||
return reset_workers
|
||
|
||
def _is_managed_worker_id(self, worker_id: str) -> bool:
|
||
worker_id = normalize_worker_id(worker_id)
|
||
return not self.managed_worker_ids or worker_id in self.managed_worker_ids
|
||
|
||
def _worker_can_dispatch(self, worker_id: str) -> bool:
|
||
worker_id = normalize_worker_id(worker_id)
|
||
if not worker_id:
|
||
return False
|
||
meta = self._get_worker_meta(worker_id)
|
||
if meta.get('policy') != 'enabled':
|
||
return False
|
||
if self.only_managed_workers_can_dispatch:
|
||
return self._is_managed_worker_id(worker_id)
|
||
return True
|
||
|
||
def _worker_matches_task_device_type(
|
||
self,
|
||
task_details: Dict[str, Any],
|
||
worker_info: Optional[Dict[str, Any]],
|
||
) -> bool:
|
||
required_device_type = self._normalize_task_device_type(task_details.get('device_type', ''))
|
||
worker_device_type = self._normalize_device_type((worker_info or {}).get('device_type', ''))
|
||
if required_device_type == 'any':
|
||
return worker_device_type in {'emulator', 'physical'}
|
||
return worker_device_type == required_device_type
|
||
|
||
def _task_matches_worker(
|
||
self,
|
||
task_details: Dict[str, Any],
|
||
worker_id: str,
|
||
worker_info: Optional[Dict[str, Any]] = None,
|
||
) -> bool:
|
||
excluded_worker_ids = {
|
||
normalize_worker_id(item)
|
||
for item in (task_details.get('excluded_worker_ids') or [])
|
||
if normalize_worker_id(item)
|
||
}
|
||
if worker_id in excluded_worker_ids:
|
||
return False
|
||
target_worker_ids = self._normalize_worker_ids(task_details.get('target_worker_ids', []))
|
||
if target_worker_ids:
|
||
if worker_id not in target_worker_ids:
|
||
return False
|
||
elif not self._worker_can_dispatch(worker_id):
|
||
return False
|
||
return self._worker_matches_task_device_type(task_details, worker_info)
|
||
|
||
def _restore_unmatched_tasks(self, skipped: List[Tuple[str, str]]):
|
||
for task_key, queue_name in skipped:
|
||
self.redis.rpush(queue_name, task_key)
|
||
|
||
def _get_worker_online_state(self, worker_info: Optional[Dict[str, Any]], current_time: Optional[float] = None) -> bool:
|
||
if not worker_info:
|
||
return False
|
||
current_time = current_time or time.time()
|
||
last_update = worker_info.get('last_update', worker_info.get('register_time', 0))
|
||
return (current_time - last_update) <= self.worker_online_timeout
|
||
|
||
def _get_dashboard_worker_lookup(self) -> Dict[str, Dict[str, Any]]:
|
||
current_time = time.time()
|
||
worker_lookup: Dict[str, Dict[str, Any]] = {}
|
||
|
||
for worker in self.get_registered_workers():
|
||
worker_copy = dict(worker)
|
||
meta = self._get_worker_meta(worker_copy.get('worker_id', ''))
|
||
inventory = self.worker_inventory.get(worker_copy.get('worker_id', ''), {})
|
||
worker_copy['online'] = self._get_worker_online_state(worker_copy, current_time)
|
||
worker_copy['is_managed'] = self._is_managed_worker_id(worker_copy.get('worker_id', ''))
|
||
worker_copy['dispatch_enabled'] = self._worker_can_dispatch(worker_copy.get('worker_id', ''))
|
||
worker_copy['policy'] = meta.get('policy', 'enabled')
|
||
worker_copy['last_action'] = meta.get('last_action', '')
|
||
worker_copy['last_action_status'] = meta.get('last_action_status', '')
|
||
worker_copy['last_action_message'] = meta.get('last_action_message', '')
|
||
worker_copy['last_action_at'] = meta.get('last_action_at', 0.0)
|
||
worker_copy['device_type'] = self._normalize_device_type(worker_copy.get('device_type', ''))
|
||
worker_copy['tags'] = inventory.get('tags', [])
|
||
worker_copy['ssh_target'] = inventory.get('ssh_target', '')
|
||
worker_lookup[worker_copy['worker_id']] = worker_copy
|
||
|
||
return worker_lookup
|
||
|
||
def _build_pending_task_entry(self, task_key: str, worker_lookup: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
task_details = json.loads(task_details_json) if task_details_json else {}
|
||
target_worker_ids = task_details.get('target_worker_ids', [])
|
||
routing_state = 'public_pending'
|
||
|
||
if target_worker_ids:
|
||
online_workers = [
|
||
worker for worker in worker_lookup.values()
|
||
if worker.get('online') and worker.get('worker_id') in target_worker_ids
|
||
]
|
||
idle_workers = [worker for worker in online_workers if worker.get('status') == 'idle']
|
||
if idle_workers:
|
||
routing_state = 'routable_pending'
|
||
elif online_workers:
|
||
routing_state = 'waiting_target_busy'
|
||
else:
|
||
routing_state = 'waiting_target_offline'
|
||
|
||
status_json = self.redis.hget("task:status", task_key)
|
||
status = json.loads(status_json) if status_json else {}
|
||
return {
|
||
'task_key': task_key,
|
||
'app_name': task_details.get('app_name', ''),
|
||
'package_name': task_details.get('package_name', ''),
|
||
'status': status.get('status', 'pending'),
|
||
'retry_count': status.get('retry_count', 0),
|
||
'device_type': self._normalize_task_device_type(task_details.get('device_type', '')),
|
||
'target_worker_ids': target_worker_ids,
|
||
'routing_mode': task_details.get('routing_mode', 'public'),
|
||
'routing_state': routing_state,
|
||
}
|
||
|
||
def load_tasks_from_app_summary(self) -> int:
|
||
"""从应用全表加载任务到三个优先级队列。"""
|
||
for queue_name in self.TASK_QUEUE_CLEANUP_NAMES:
|
||
self.redis.delete(queue_name)
|
||
self.redis.delete("task:status")
|
||
self.redis.delete("task:details")
|
||
self.redis.delete("task:completed")
|
||
self.redis.delete("task:failed")
|
||
logger.info("[清空] 已清空任务队列和相关数据")
|
||
|
||
entries = self.analytics.list_pending_collection_tasks()
|
||
high_count = 0
|
||
default_count = 0
|
||
active_base_keys: set = set()
|
||
|
||
for entry in entries:
|
||
payload = entry.get("task_payload") or {}
|
||
package_name = str(payload.get("package_name") or entry.get("package_name") or "").strip()
|
||
app_name = str(payload.get("app_name") or entry.get("app_name") or "").strip()
|
||
if not package_name or not app_name:
|
||
continue
|
||
task_key = _make_task_key(app_name, package_name)
|
||
task_info = self._build_task_info_from_payload(payload)
|
||
unmanaged_targets = [
|
||
target_worker_id for target_worker_id in task_info.get('target_worker_ids', [])
|
||
if not self._is_managed_worker_id(target_worker_id)
|
||
]
|
||
if unmanaged_targets:
|
||
logger.warning(
|
||
f"[定向配置] 任务 {task_key} 配置了未纳入名单的目标Worker: {unmanaged_targets}"
|
||
)
|
||
task_queue = self._normalize_task_queue_label(entry.get("task_queue"), "default")
|
||
task_info["task_queue"] = "high" if task_queue == "high" else "default"
|
||
self.redis.hset("task:details", task_key, json.dumps(task_info))
|
||
active_base_keys.add(task_key)
|
||
|
||
if task_queue == "high":
|
||
self.redis.lpush("task:queue:high", task_key)
|
||
high_count += 1
|
||
else:
|
||
self.redis.lpush("task:queue:default", task_key)
|
||
default_count += 1
|
||
|
||
# 加载低优先级模型流量任务（使用 _model 后缀 task_key 避免与采集任务冲突）
|
||
model_entries = self.analytics.list_model_eligible_apps()
|
||
low_count = 0
|
||
for entry in model_entries:
|
||
payload = entry.get("task_payload") or {}
|
||
package_name = str(payload.get("package_name") or entry.get("package_name") or "").strip()
|
||
app_name = str(payload.get("app_name") or entry.get("app_name") or "").strip()
|
||
if not package_name or not app_name:
|
||
continue
|
||
base_task_key = _make_task_key(app_name, package_name)
|
||
if base_task_key in active_base_keys:
|
||
continue
|
||
task_key = f"{base_task_key}_model"
|
||
task_info = self._build_task_info_from_payload(payload)
|
||
task_info["keep_app_installed"] = True
|
||
task_info["task_queue"] = "low"
|
||
task_info["package_name"] = package_name
|
||
task_info["app_name"] = app_name
|
||
self.redis.hset("task:details", task_key, json.dumps(task_info))
|
||
self.redis.lpush("task:queue:low", task_key)
|
||
low_count += 1
|
||
|
||
logger.info(
|
||
f"[加载完成] 高优 {high_count} | 默认 {default_count} | 低优(模型) {low_count}"
|
||
)
|
||
return high_count + default_count + low_count
|
||
|
||
def refresh_tasks_from_app_summary(self) -> int:
|
||
"""增量刷新：从应用全表加载新增的 pending/eligible 任务到三个优先级队列。"""
|
||
entries = self.analytics.list_pending_collection_tasks()
|
||
added_count = 0
|
||
active_base_keys: set = set()
|
||
for entry in entries:
|
||
payload = entry.get("task_payload") or {}
|
||
package_name = str(payload.get("package_name") or entry.get("package_name") or "").strip()
|
||
app_name = str(payload.get("app_name") or entry.get("app_name") or "").strip()
|
||
if not package_name or not app_name:
|
||
continue
|
||
task_key = _make_task_key(app_name, package_name)
|
||
active_base_keys.add(task_key)
|
||
|
||
if self._is_task_queued(task_key):
|
||
continue
|
||
if self._is_task_running(task_key):
|
||
continue
|
||
|
||
task_info = self._build_task_info_from_payload(payload)
|
||
unmanaged_targets = [
|
||
target_worker_id for target_worker_id in task_info.get('target_worker_ids', [])
|
||
if not self._is_managed_worker_id(target_worker_id)
|
||
]
|
||
if unmanaged_targets:
|
||
logger.warning(
|
||
f"[定向配置] 任务 {task_key} 配置了未纳入名单的目标Worker: {unmanaged_targets}"
|
||
)
|
||
task_queue = self._normalize_task_queue_label(entry.get("task_queue"), "default")
|
||
task_info["task_queue"] = "high" if task_queue == "high" else "default"
|
||
self.redis.hset("task:details", task_key, json.dumps(task_info))
|
||
target_queue = "task:queue:high" if task_queue == "high" else "task:queue:default"
|
||
self._remove_task_from_queue(task_key)
|
||
self._mark_task_pending_for_queue(task_key)
|
||
self.redis.lpush(target_queue, task_key)
|
||
added_count += 1
|
||
|
||
model_entries = self.analytics.list_model_eligible_apps()
|
||
for entry in model_entries:
|
||
payload = entry.get("task_payload") or {}
|
||
package_name = str(payload.get("package_name") or entry.get("package_name") or "").strip()
|
||
app_name = str(payload.get("app_name") or entry.get("app_name") or "").strip()
|
||
if not package_name or not app_name:
|
||
continue
|
||
base_task_key = _make_task_key(app_name, package_name)
|
||
if base_task_key in active_base_keys:
|
||
continue
|
||
task_key = f"{base_task_key}_model"
|
||
if self._is_task_queued(task_key):
|
||
continue
|
||
if self._is_task_running(task_key):
|
||
continue
|
||
|
||
task_info = self._build_task_info_from_payload(payload)
|
||
task_info["keep_app_installed"] = True
|
||
task_info["task_queue"] = "low"
|
||
task_info["package_name"] = package_name
|
||
task_info["app_name"] = app_name
|
||
self.redis.hset("task:details", task_key, json.dumps(task_info))
|
||
self._remove_task_from_queue(task_key)
|
||
self._mark_task_pending_for_queue(task_key)
|
||
self.redis.lpush("task:queue:low", task_key)
|
||
added_count += 1
|
||
|
||
if added_count > 0:
|
||
logger.info(f"[增量刷新] 从应用全表新增任务 {added_count} 个")
|
||
return added_count
|
||
|
||
def _clear_block_tasks(self) -> int:
|
||
cleared = 0
|
||
all_details = dict(self.redis.hgetall("task:details") or {})
|
||
for task_key in all_details:
|
||
if not task_key.endswith("_block"):
|
||
continue
|
||
if self._is_task_running(task_key):
|
||
continue
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.hdel("task:details", task_key)
|
||
self.redis.hdel("task:status", task_key)
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
cleared += 1
|
||
return cleared
|
||
|
||
def load_block_tasks(self, packages: List[Dict[str, Any]]) -> int:
|
||
cleared = self._clear_block_tasks()
|
||
if cleared:
|
||
logger.info(f"[Block测试] 已清除 {cleared} 个旧的 block 任务")
|
||
count = 0
|
||
for entry in packages:
|
||
payload = entry.get("task_payload") or {}
|
||
package_name = str(payload.get("package_name") or entry.get("package_name") or "").strip()
|
||
app_name = str(payload.get("app_name") or entry.get("app_name") or "").strip()
|
||
if not package_name or not app_name:
|
||
continue
|
||
base_task_key = _make_task_key(app_name, package_name)
|
||
task_key = f"{base_task_key}_block"
|
||
if self._is_task_running(task_key):
|
||
continue
|
||
task_info = self._build_task_info_from_payload(payload)
|
||
task_info["is_block_task"] = True
|
||
task_info["task_queue"] = "high"
|
||
self.redis.hset("task:details", task_key, json.dumps(task_info))
|
||
self._remove_task_from_queue(task_key)
|
||
self._mark_task_pending_for_queue(task_key)
|
||
self.redis.lpush("task:queue:high", task_key)
|
||
count += 1
|
||
logger.info(f"[Block测试] 已加载 {count} 个 block 任务到高优队列")
|
||
return count
|
||
|
||
def load_all_qualified_block_tasks(self) -> int:
|
||
entries = self.analytics.repo.list_qualified_apps()
|
||
return self.load_block_tasks(entries)
|
||
|
||
def worker_init(self, worker_id, ip_address, mac_address, hostname, platform, device_type=''):
|
||
"""Worker初始化：注册Worker并领取第一个任务
|
||
|
||
Args:
|
||
worker_id: Worker标识符（IP_MAC）
|
||
ip_address: IP地址
|
||
mac_address: MAC地址
|
||
hostname: 主机名
|
||
platform: 平台
|
||
|
||
Returns:
|
||
dict: 包含第一个任务信息或None（无任务）
|
||
"""
|
||
raw_worker_id = str(worker_id or "").strip()
|
||
worker_id = normalize_worker_id(worker_id, ip_address)
|
||
self._migrate_worker_runtime_id(raw_worker_id, worker_id)
|
||
existing_worker = self.redis.hget("workers:info", worker_id)
|
||
previous_task = None
|
||
meta = self._get_worker_meta(worker_id)
|
||
|
||
if existing_worker:
|
||
# 获取之前分配的任务
|
||
previous_task_key = self.redis.hget("worker:tasks", worker_id)
|
||
if previous_task_key:
|
||
logger.warning(f"[重复初始化] Worker {worker_id} 再次初始化，将重新分配之前的任务: {previous_task_key}")
|
||
# 获取任务详情
|
||
task_details_json = self.redis.hget("task:details", previous_task_key)
|
||
if task_details_json:
|
||
task_details = json.loads(task_details_json)
|
||
previous_task = self._task_to_payload(previous_task_key, task_details)
|
||
else:
|
||
logger.warning(f"[重复初始化] Worker {worker_id} 再次初始化，之前无分配任务")
|
||
|
||
# 注册/更新Worker信息
|
||
worker_info = {
|
||
'worker_id': worker_id,
|
||
'ip_address': ip_address,
|
||
'mac_address': mac_address,
|
||
'hostname': hostname,
|
||
'platform': platform,
|
||
'device_type': self._normalize_device_type(device_type),
|
||
'register_time': time.time(),
|
||
'last_update': time.time(),
|
||
'status': 'idle',
|
||
'is_managed': self._is_managed_worker_id(worker_id),
|
||
'dispatch_enabled': self._worker_can_dispatch(worker_id),
|
||
}
|
||
|
||
self.redis.hset("workers:info", worker_id, json.dumps(worker_info))
|
||
self.redis.sadd("workers:idle", worker_id)
|
||
self.monitor.register_worker(worker_id, event_time=worker_info['last_update'])
|
||
if (
|
||
meta.get('auto_recover_used')
|
||
and self._is_worker_recovery_action(meta.get('last_action', ''))
|
||
and meta.get('last_action_status') == 'success'
|
||
and self.redis.exists(self._worker_recovery_pending_key(worker_id))
|
||
):
|
||
self.redis.delete(self._worker_recovery_pending_key(worker_id))
|
||
self.redis.setex(self._worker_recovery_observe_key(worker_id), WORKER_INIT_MONITOR_WINDOW, str(time.time()))
|
||
|
||
# 如果有之前的任务，直接返回该任务
|
||
if previous_task:
|
||
self._update_worker_status(worker_id, 'busy', previous_task['task_key'])
|
||
self.monitor.handle_task_assigned(worker_id, previous_task['task_key'])
|
||
logger.info(f"[初始化] Worker {worker_id} 重新领取之前的任务: {previous_task['app_name']}")
|
||
return previous_task
|
||
|
||
if not worker_info['dispatch_enabled']:
|
||
logger.info(f"[初始化] Worker {worker_id} 已注册，但当前未启用分发")
|
||
return None
|
||
|
||
# 否则领取新任务
|
||
task = self._assign_task(worker_id)
|
||
|
||
if task:
|
||
logger.info(f"[初始化] Worker {worker_id} 注册成功，领取任务: {task['app_name']}")
|
||
else:
|
||
logger.info(f"[初始化] Worker {worker_id} 注册成功，暂无任务")
|
||
|
||
return task
|
||
|
||
def worker_report(self, worker_id, previous_task_key, report_data: Optional[Dict[str, Any]] = None, **legacy_kwargs):
|
||
"""上报前一个任务的完成状态，并获取下一个任务
|
||
|
||
Args:
|
||
worker_id: Worker标识符
|
||
previous_task_key: 前一个任务的键（应用名称_包名）
|
||
report_data: 上报数据字典，结构如下:
|
||
{
|
||
"status": "success|failed|stop", # 必填
|
||
"error": { # 失败时必填
|
||
"category": "INFRA_ERROR|DOWNLOAD_ERROR|APP_ERROR|BUSINESS_ERROR",
|
||
"code": 1,
|
||
"reason": "错误原因描述",
|
||
"details": "详细错误信息（可选）",
|
||
"crashed_source": "google_play" # APP_CRASH时必填，闪退的下载源
|
||
},
|
||
"metrics": { # 可选
|
||
"login_count": 0,
|
||
"register_count": 0,
|
||
"stuck_reason_code": 0,
|
||
"guiagent_message": "GuiAgent消息",
|
||
"scenario_triggered": true
|
||
},
|
||
"statistics": { # 可选
|
||
"round": 1,
|
||
"exit_code": 0,
|
||
"error_reason": "...",
|
||
"droidbot_steps": 100,
|
||
"guiagent_steps": 50,
|
||
"total_steps": 150,
|
||
"duration_seconds": 120.5,
|
||
"is_retry": false,
|
||
"download_source": "google"
|
||
}
|
||
}
|
||
|
||
Returns:
|
||
dict: 包含下一个任务信息或None（无任务/stop状态/无效Worker）
|
||
"""
|
||
VALID_STATUS = ['success', 'failed', 'stop']
|
||
raw_worker_id = str(worker_id or "").strip()
|
||
worker_id = normalize_worker_id(worker_id)
|
||
self._migrate_worker_runtime_id(raw_worker_id, worker_id)
|
||
if not isinstance(report_data, dict):
|
||
report_data = {
|
||
'status': legacy_kwargs.get('status', 'failed'),
|
||
'error': {},
|
||
'metrics': {},
|
||
'statistics': {},
|
||
}
|
||
message = legacy_kwargs.get('message')
|
||
error_type = legacy_kwargs.get('error_type')
|
||
if message:
|
||
report_data['error']['reason'] = message
|
||
report_data['error']['details'] = message
|
||
if error_type:
|
||
category, _, code_text = str(error_type).partition('/')
|
||
report_data['error']['category'] = category or 'INFRA_ERROR'
|
||
try:
|
||
report_data['error']['code'] = int(code_text)
|
||
except (TypeError, ValueError):
|
||
report_data['error']['code'] = 0
|
||
if not report_data['error']:
|
||
report_data.pop('error')
|
||
|
||
status = report_data.get('status', 'failed')
|
||
error_data = report_data.get('error', {})
|
||
metrics = report_data.get('metrics', {})
|
||
statistics = report_data.get('statistics', {})
|
||
download_errors = report_data.get('download_errors', {})
|
||
|
||
error_info = ErrorInfo.from_report_dict(error_data)
|
||
crashed_source = error_data.get('crashed_source') if error_data else None
|
||
|
||
login_count = metrics.get('login_count', 0) if metrics else 0
|
||
register_count = metrics.get('register_count', 0) if metrics else 0
|
||
guiagent_message = metrics.get('guiagent_message', '') if metrics else ''
|
||
scenario_triggered = metrics.get('scenario_triggered', False) if metrics else False
|
||
|
||
message = error_info.reason
|
||
|
||
if download_errors:
|
||
source_msgs = [f"{k}: {v['message']}" for k, v in download_errors.items()]
|
||
message = f"下载失败: {', '.join(source_msgs)}"
|
||
elif guiagent_message:
|
||
if guiagent_message != error_info.reason:
|
||
message = guiagent_message
|
||
elif error_info.details and error_info.details != error_info.reason:
|
||
message = error_info.details
|
||
|
||
if status == 'success':
|
||
if login_count > 0 or register_count > 0:
|
||
message = f"成功 (登录{login_count}次, 注册{register_count}次)"
|
||
|
||
if not isinstance(status, str) or status not in VALID_STATUS:
|
||
logger.warning(f"Worker {worker_id} 上报了无效的status: {status} (类型: {type(status).__name__})")
|
||
logger.warning(f"预期status值: {VALID_STATUS}")
|
||
return None
|
||
|
||
worker_exists = self.redis.hget("workers:info", worker_id)
|
||
retry_count = 0
|
||
if previous_task_key:
|
||
previous_status_json = self.redis.hget("task:status", previous_task_key)
|
||
if previous_status_json:
|
||
previous_status = json.loads(previous_status_json)
|
||
retry_count = previous_status.get('retry_count', 0)
|
||
self.monitor.handle_worker_report(
|
||
worker_id=worker_id,
|
||
previous_task_key=previous_task_key,
|
||
report_data=report_data,
|
||
current_time=time.time(),
|
||
retry_count=retry_count,
|
||
resolved_message=message,
|
||
)
|
||
if not worker_exists:
|
||
logger.warning(f"[忽略] Worker {worker_id} 已不在Worker列表中，返回错误码")
|
||
return {
|
||
'task_key': 'ERROR',
|
||
'app_name': 'ERROR',
|
||
'package_name': '-1',
|
||
'country_code': '',
|
||
'device_type': '',
|
||
'error_code': -1,
|
||
'error_message': 'Worker已从列表中移除'
|
||
}
|
||
|
||
if previous_task_key:
|
||
if status in ['stop', 'failed'] and message:
|
||
logger.warning(f"[告警] Worker {worker_id} [{error_info.category.name}/{error_info.code}]: {message}")
|
||
|
||
if status in ['stop', 'failed']:
|
||
self._record_failed_task(
|
||
previous_task_key,
|
||
worker_id,
|
||
error_info,
|
||
message,
|
||
statistics,
|
||
crashed_source,
|
||
download_errors=download_errors,
|
||
)
|
||
if self.redis.hget("worker:tasks", worker_id) == previous_task_key:
|
||
self.redis.hdel("worker:tasks", worker_id)
|
||
self._update_worker_status(worker_id, 'idle', None)
|
||
self.monitor.set_worker_state(worker_id, 'idle_waiting_task', reason='waiting_task')
|
||
elif status == 'success':
|
||
self._complete_task(previous_task_key, worker_id, status, message, statistics)
|
||
task_details_json = self.redis.hget("task:details", previous_task_key)
|
||
package_name = ""
|
||
if task_details_json:
|
||
package_name = json.loads(task_details_json).get("package_name", "")
|
||
if package_name:
|
||
self.analytics.enqueue_incremental(
|
||
package_name=package_name,
|
||
task_key=previous_task_key,
|
||
worker_id=worker_id,
|
||
report_time=time.time(),
|
||
trigger_source="worker_report",
|
||
waiting_for_artifacts=True,
|
||
)
|
||
|
||
# stop状态：清除Worker并不再分配任务
|
||
if status == 'stop':
|
||
self.monitor.mark_worker_offline(worker_id, "worker_stopped")
|
||
self._remove_worker(worker_id)
|
||
logger.info(f"[停止] Worker {worker_id} 请求停止，已从Worker列表移除")
|
||
return None
|
||
|
||
# 获取下一个任务
|
||
next_task = self._assign_task(worker_id)
|
||
|
||
if next_task:
|
||
logger.info(f"[上报] Worker {worker_id} 上报任务 {previous_task_key}={status}，领取新任务: {next_task['app_name']}")
|
||
else:
|
||
self.monitor.set_worker_state(worker_id, 'idle_waiting_task', reason='waiting_task')
|
||
logger.info(f"[上报] Worker {worker_id} 上报任务 {previous_task_key}={status}，暂无新任务")
|
||
|
||
return next_task
|
||
|
||
def get_analytics_overview(
|
||
self,
|
||
*,
|
||
incremental_batch_tag: str = "",
|
||
top_n: int = 3000,
|
||
) -> Dict[str, Any]:
|
||
return self.analytics.get_overview(
|
||
incremental_batch_tag=incremental_batch_tag,
|
||
top_n=top_n,
|
||
)
|
||
|
||
def get_analytics_apps(
|
||
self,
|
||
*,
|
||
q: str = "",
|
||
latest_status: str = "",
|
||
artifact_status: str = "",
|
||
collection_status: str = "",
|
||
restriction_status: str = "",
|
||
retryability: str = "",
|
||
incremental_batch_tag: str = "",
|
||
top_n: int = 3000,
|
||
sort: str = "updated_at",
|
||
order: str = "desc",
|
||
page: int = 1,
|
||
page_size: int = 50,
|
||
) -> Dict[str, Any]:
|
||
return self.analytics.list_apps(
|
||
q=q,
|
||
latest_status=latest_status,
|
||
artifact_status=artifact_status,
|
||
collection_status=collection_status,
|
||
restriction_status=restriction_status,
|
||
retryability=retryability,
|
||
incremental_batch_tag=incremental_batch_tag,
|
||
top_n=top_n,
|
||
sort=sort,
|
||
order=order,
|
||
page=page,
|
||
page_size=page_size,
|
||
)
|
||
|
||
def get_analytics_app_detail(self, package_name: str) -> Optional[Dict[str, Any]]:
|
||
return self.analytics.get_app_detail(package_name)
|
||
|
||
def get_analytics_jobs(
|
||
self,
|
||
*,
|
||
job_type: str = "",
|
||
status: str = "",
|
||
package_name: str = "",
|
||
limit: int = 50,
|
||
) -> List[Dict[str, Any]]:
|
||
return self.analytics.list_jobs(
|
||
job_type=job_type or None,
|
||
status=status or None,
|
||
package_name=package_name or None,
|
||
limit=limit,
|
||
)
|
||
|
||
def enqueue_analytics_backfill(self, packages: Optional[List[str]] = None) -> Dict[str, Any]:
|
||
return self.analytics.enqueue_backfill(packages=packages)
|
||
|
||
def enqueue_analytics_rebuild(self, package_name: str) -> Dict[str, Any]:
|
||
return self.analytics.enqueue_manual_rebuild(package_name)
|
||
|
||
def handle_analytics_worker_event(self, payload: Dict[str, Any]) -> int:
|
||
return self.analytics.handle_worker_event(payload)
|
||
|
||
def close(self):
|
||
try:
|
||
self.analytics.close()
|
||
except Exception:
|
||
logger.exception("Failed to close analytics service")
|
||
|
||
def worker_retry(self, worker_id, current_task_key):
|
||
"""重试：告知该任务需重新计时
|
||
|
||
Args:
|
||
worker_id: Worker标识符
|
||
current_task_key: 当前任务的键（应用名称_包名）
|
||
|
||
Returns:
|
||
bool: 是否成功更新
|
||
"""
|
||
raw_worker_id = str(worker_id or "").strip()
|
||
worker_id = normalize_worker_id(worker_id)
|
||
self._migrate_worker_runtime_id(raw_worker_id, worker_id)
|
||
# 检查Worker是否已被清除
|
||
worker_exists = self.redis.hget("workers:info", worker_id)
|
||
if not worker_exists:
|
||
logger.warning(f"[忽略] Worker {worker_id} 已不在Worker列表中，忽略其retry")
|
||
return False
|
||
|
||
if not current_task_key:
|
||
return False
|
||
|
||
try:
|
||
# 获取任务状态
|
||
status_json = self.redis.hget("task:status", current_task_key)
|
||
if not status_json:
|
||
return False
|
||
|
||
status = json.loads(status_json)
|
||
if status['status'] != 'running' or status.get('worker_id') != worker_id:
|
||
return False
|
||
|
||
current_time = time.time()
|
||
# 更新重试时间（保留原始start_time，用于超时计算时以重试时间为准）
|
||
status['last_retry'] = current_time
|
||
self.redis.hset("task:status", current_task_key, json.dumps(status))
|
||
|
||
# 更新Worker最后更新时间
|
||
worker_info_json = self.redis.hget("workers:info", worker_id)
|
||
if worker_info_json:
|
||
worker_info = json.loads(worker_info_json)
|
||
worker_info['last_update'] = current_time
|
||
self.redis.hset("workers:info", worker_id, json.dumps(worker_info))
|
||
|
||
logger.info(f"[重试] Worker {worker_id} 任务 {current_task_key} 重新计时成功")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"重试更新失败: {e}")
|
||
return False
|
||
|
||
def _assign_task(self, worker_id) -> Optional[Dict[str, Any]]:
|
||
"""分配任务给Worker（内部方法），按优先级：高优 → 默认 → 低优。"""
|
||
lock_key = "lock:assign:queue"
|
||
lock = self.redis.lock(lock_key, timeout=5, blocking_timeout=1)
|
||
|
||
try:
|
||
with lock:
|
||
current_task = self.redis.hget("worker:tasks", worker_id)
|
||
if current_task:
|
||
return None
|
||
|
||
if not self._worker_can_dispatch(worker_id):
|
||
return None
|
||
|
||
worker_info = self._get_worker_info(worker_id)
|
||
|
||
task_key = None
|
||
task_details = None
|
||
skipped: List[Tuple[str, str]] = []
|
||
|
||
for queue_name in self.TASK_QUEUE_NAMES:
|
||
queue_length = self.redis.llen(queue_name)
|
||
for _ in range(queue_length):
|
||
candidate_task_key = self.redis.rpop(queue_name)
|
||
if not candidate_task_key:
|
||
break
|
||
|
||
task_details_json = self.redis.hget("task:details", candidate_task_key)
|
||
if not task_details_json:
|
||
logger.warning(f"[任务详情缺失] 任务 {candidate_task_key} 缺少详情，已跳过")
|
||
continue
|
||
|
||
candidate_task_details = json.loads(task_details_json)
|
||
if self._task_matches_worker(candidate_task_details, worker_id, worker_info=worker_info):
|
||
task_key = candidate_task_key
|
||
task_details = candidate_task_details
|
||
break
|
||
|
||
skipped.append((candidate_task_key, queue_name))
|
||
|
||
if task_key:
|
||
break
|
||
|
||
if skipped:
|
||
self._restore_unmatched_tasks(skipped)
|
||
|
||
if not task_key or not task_details:
|
||
return None
|
||
|
||
current_time = time.time()
|
||
if task_details.get('excluded_worker_ids'):
|
||
task_details = dict(task_details)
|
||
task_details.pop('excluded_worker_ids', None)
|
||
self.redis.hset("task:details", task_key, json.dumps(task_details))
|
||
|
||
# 获取之前的重试次数
|
||
prev_status_json = self.redis.hget("task:status", task_key)
|
||
retry_count = 0
|
||
if prev_status_json:
|
||
prev_status = json.loads(prev_status_json)
|
||
retry_count = prev_status.get('retry_count', 0)
|
||
|
||
# 更新任务状态
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
'status': 'running',
|
||
'worker_id': worker_id,
|
||
'start_time': current_time,
|
||
'last_retry': current_time,
|
||
'retry_count': retry_count
|
||
}))
|
||
|
||
# 更新Worker任务映射
|
||
self.redis.hset("worker:tasks", worker_id, task_key)
|
||
|
||
# 更新Worker状态
|
||
self._update_worker_status(worker_id, 'busy', task_key)
|
||
self.monitor.handle_task_assigned(worker_id, task_key, event_time=current_time)
|
||
|
||
# 发布任务分发事件
|
||
self.redis.publish(f"task:assign:{worker_id}", json.dumps({
|
||
'task_key': task_key,
|
||
'task_details': task_details
|
||
}))
|
||
|
||
logger.info(
|
||
f"[分配] 任务 {task_key} -> Worker {worker_id} "
|
||
f"(routing={task_details.get('routing_mode', 'public')}, "
|
||
f"device_type={task_details.get('device_type', '')}, "
|
||
f"targets={task_details.get('target_worker_ids', [])})"
|
||
)
|
||
|
||
return self._task_to_payload(task_key, task_details)
|
||
|
||
except redis.exceptions.LockNotOwnedError:
|
||
# 锁已过期或被其他进程获取
|
||
return None
|
||
except redis.exceptions.LockError:
|
||
# 无法获取锁
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"分配任务到 Worker {worker_id} 时出错: {e}")
|
||
return None
|
||
|
||
def _update_worker_status(self, worker_id: str, status: str, current_task: Optional[str] = None):
|
||
"""更新Worker状态的辅助方法"""
|
||
current_time = time.time()
|
||
|
||
if status == 'busy':
|
||
self.redis.sadd("workers:busy", worker_id)
|
||
self.redis.srem("workers:idle", worker_id)
|
||
else: # idle
|
||
self.redis.sadd("workers:idle", worker_id)
|
||
self.redis.srem("workers:busy", worker_id)
|
||
|
||
worker_info_json = self.redis.hget("workers:info", worker_id)
|
||
if worker_info_json:
|
||
worker_info = json.loads(worker_info_json)
|
||
worker_info['status'] = status
|
||
worker_info['current_task'] = current_task
|
||
worker_info['last_update'] = current_time
|
||
self.redis.hset("workers:info", worker_id, json.dumps(worker_info))
|
||
|
||
def _remove_worker(self, worker_id: str):
|
||
"""从Worker列表中移除Worker（内部方法）
|
||
|
||
Args:
|
||
worker_id: Worker标识符
|
||
"""
|
||
# 从idle/busy集合中移除
|
||
self.redis.srem("workers:idle", worker_id)
|
||
self.redis.srem("workers:busy", worker_id)
|
||
|
||
# 删除Worker信息
|
||
self.redis.hdel("workers:info", worker_id)
|
||
|
||
# 清除Worker的任务映射
|
||
self.redis.hdel("worker:tasks", worker_id)
|
||
|
||
def _complete_task(self, task_key, worker_id, status, message='', statistics=None):
|
||
"""任务完成（内部方法）"""
|
||
current_time = time.time()
|
||
|
||
prev_status_json = self.redis.hget("task:status", task_key)
|
||
retry_count = 0
|
||
if prev_status_json:
|
||
prev_status = json.loads(prev_status_json)
|
||
retry_count = prev_status.get('retry_count', 0)
|
||
|
||
task_status = {
|
||
'status': status,
|
||
'worker_id': worker_id,
|
||
'end_time': current_time,
|
||
'retry_count': retry_count
|
||
}
|
||
|
||
if statistics:
|
||
task_status['statistics'] = statistics
|
||
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
|
||
if self.redis.hget("worker:tasks", worker_id) == task_key:
|
||
self.redis.hdel("worker:tasks", worker_id)
|
||
self._update_worker_status(worker_id, 'idle', None)
|
||
self.monitor.set_worker_state(worker_id, 'idle_waiting_task', reason='waiting_task')
|
||
|
||
if status == 'success':
|
||
self.redis.sadd("task:completed", task_key)
|
||
else:
|
||
new_retry_count = retry_count + 1
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
task_details = json.loads(task_details_json) if task_details_json else {}
|
||
retry_task_details = self._mark_retry_avoid_worker(task_key, task_details, worker_id)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
'status': 'pending_retry',
|
||
'worker_id': worker_id,
|
||
'end_time': current_time,
|
||
'retry_count': new_retry_count
|
||
}))
|
||
self.redis.srem("task:failed", task_key)
|
||
self._queue_retry_task(
|
||
task_key,
|
||
retry_task_details,
|
||
prioritize=self._is_hot_new_app_task(retry_task_details),
|
||
)
|
||
logger.info(f"[重试] 任务 {task_key} 失败，已重新入队 (第{new_retry_count}次失败)")
|
||
|
||
self.redis.publish("task:completed", json.dumps({
|
||
'task_key': task_key,
|
||
'worker_id': worker_id,
|
||
'status': status,
|
||
'retry_count': retry_count,
|
||
'statistics': statistics
|
||
}))
|
||
|
||
def get_statistics(self):
|
||
"""获取统计信息
|
||
|
||
Returns:
|
||
dict: 包含各项统计数据
|
||
- pending: 待分发任务数（含重试中任务）
|
||
- running: 运行中任务数
|
||
- completed: 成功任务数
|
||
- failed: 永久失败任务数
|
||
"""
|
||
all_status = self.redis.hgetall("task:status")
|
||
running_count = 0
|
||
pending_entries = []
|
||
worker_lookup = self._get_dashboard_worker_lookup()
|
||
queue_task_keys = self._get_queued_task_keys()
|
||
|
||
for task_key, status_json in all_status.items():
|
||
status = json.loads(status_json)
|
||
if status.get('status') == 'running':
|
||
running_count += 1
|
||
|
||
for task_key in queue_task_keys:
|
||
pending_entries.append(self._build_pending_task_entry(task_key, worker_lookup))
|
||
|
||
return {
|
||
'pending': len(queue_task_keys),
|
||
'running': running_count,
|
||
'completed': self.redis.scard("task:completed"),
|
||
'failed': self.redis.scard("task:failed"),
|
||
'pending_targeted': sum(1 for item in pending_entries if item['target_worker_ids']),
|
||
'waiting_target_busy': sum(1 for item in pending_entries if item['routing_state'] == 'waiting_target_busy'),
|
||
'waiting_target_offline': sum(1 for item in pending_entries if item['routing_state'] == 'waiting_target_offline'),
|
||
}
|
||
|
||
def get_registered_workers(self):
|
||
"""获取所有已注册的Worker（以 worker_inventory.json 为准）"""
|
||
workers_info = self.redis.hgetall("workers:info")
|
||
workers = []
|
||
for worker_id, inventory_data in self.worker_inventory.items():
|
||
info_json = workers_info.get(worker_id)
|
||
if info_json:
|
||
info = json.loads(info_json)
|
||
else:
|
||
info = {
|
||
"worker_id": worker_id,
|
||
"ip_address": inventory_data.get("ssh_target", ""),
|
||
"hostname": "",
|
||
"platform": "",
|
||
"device_type": "",
|
||
"status": "offline",
|
||
"register_time": 0,
|
||
"last_update": 0,
|
||
"current_task": None
|
||
}
|
||
info.setdefault('dispatch_enabled', self._worker_can_dispatch(worker_id))
|
||
info.setdefault('is_managed', True)
|
||
workers.append(info)
|
||
return workers
|
||
|
||
def get_idle_workers(self):
|
||
"""获取空闲的Worker列表"""
|
||
idle_workers = self.redis.smembers("workers:idle")
|
||
workers_info = self.redis.hgetall("workers:info")
|
||
|
||
idle_worker_list = []
|
||
for worker_id in idle_workers:
|
||
if worker_id in self.worker_inventory and worker_id in workers_info:
|
||
info = json.loads(workers_info[worker_id])
|
||
idle_worker_list.append(info)
|
||
|
||
return idle_worker_list
|
||
|
||
def get_busy_workers(self):
|
||
"""获取忙碌的Worker列表"""
|
||
busy_workers = self.redis.smembers("workers:busy")
|
||
workers_info = self.redis.hgetall("workers:info")
|
||
|
||
busy_worker_list = []
|
||
for worker_id in busy_workers:
|
||
if worker_id in self.worker_inventory and worker_id in workers_info:
|
||
info = json.loads(workers_info[worker_id])
|
||
busy_worker_list.append(info)
|
||
|
||
return busy_worker_list
|
||
|
||
def get_dashboard_workers(self) -> List[Dict[str, Any]]:
|
||
current_time = time.time()
|
||
registered_by_id: Dict[str, Dict[str, Any]] = {}
|
||
extra_workers: List[Dict[str, Any]] = []
|
||
|
||
for worker in self.get_registered_workers():
|
||
worker_copy = dict(worker)
|
||
worker_copy['online'] = self._get_worker_online_state(worker_copy, current_time)
|
||
worker_copy['dispatch_enabled'] = self._worker_can_dispatch(worker_copy.get('worker_id', ''))
|
||
worker_copy['is_managed'] = self._is_managed_worker_id(worker_copy.get('worker_id', ''))
|
||
worker_copy['state'] = worker_copy.get('state', 'idle_waiting_task' if worker_copy['online'] else 'offline')
|
||
registered_by_id[worker_copy.get('worker_id', '')] = worker_copy
|
||
|
||
if not self.managed_worker_ids:
|
||
return sorted(
|
||
registered_by_id.values(),
|
||
key=lambda item: (item.get('ip_address', ''), item.get('worker_id', '')),
|
||
)
|
||
|
||
managed_workers: List[Dict[str, Any]] = []
|
||
for managed_worker_id in sorted(self.managed_worker_ids):
|
||
worker = registered_by_id.get(managed_worker_id, {})
|
||
meta = self._get_worker_meta(managed_worker_id)
|
||
inventory = self.worker_inventory.get(managed_worker_id, {})
|
||
managed_workers.append({
|
||
'worker_id': managed_worker_id,
|
||
'ip_address': worker.get('ip_address', self._get_worker_ip(managed_worker_id)),
|
||
'hostname': worker.get('hostname', ''),
|
||
'platform': worker.get('platform', ''),
|
||
'device_type': self._normalize_device_type(worker.get('device_type', '')),
|
||
'current_task': worker.get('current_task'),
|
||
'status': worker.get('status', 'offline') if worker else 'offline',
|
||
'state': worker.get('state', 'idle_waiting_task') if worker else 'offline',
|
||
'state_since': worker.get('state_since'),
|
||
'pause_reason': worker.get('pause_reason'),
|
||
'register_time': worker.get('register_time'),
|
||
'last_update': worker.get('last_update'),
|
||
'online': self._get_worker_online_state(worker, current_time) if worker else False,
|
||
'is_managed': True,
|
||
'dispatch_enabled': self._worker_can_dispatch(managed_worker_id),
|
||
'policy': meta.get('policy', 'enabled'),
|
||
'last_action': meta.get('last_action', ''),
|
||
'last_action_status': meta.get('last_action_status', ''),
|
||
'last_action_message': meta.get('last_action_message', ''),
|
||
'last_action_at': meta.get('last_action_at', 0.0),
|
||
'tags': inventory.get('tags', []),
|
||
'ssh_target': inventory.get('ssh_target', ''),
|
||
})
|
||
|
||
return managed_workers
|
||
|
||
def get_dashboard_tasks(
|
||
self,
|
||
pending_offset: int = 0,
|
||
pending_limit: int = 50,
|
||
recent_limit: int = 20,
|
||
) -> Dict[str, Any]:
|
||
worker_lookup = self._get_dashboard_worker_lookup()
|
||
running_tasks = []
|
||
failed_tasks = []
|
||
pending_tasks = []
|
||
|
||
all_status = self.redis.hgetall("task:status")
|
||
for task_key, status_json in all_status.items():
|
||
status = json.loads(status_json)
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
task_details = json.loads(task_details_json) if task_details_json else {}
|
||
task_entry = {
|
||
'task_key': task_key,
|
||
'app_name': task_details.get('app_name', ''),
|
||
'package_name': task_details.get('package_name', ''),
|
||
'worker_id': status.get('worker_id'),
|
||
'status': status.get('status', 'unknown'),
|
||
'retry_count': status.get('retry_count', 0),
|
||
'device_type': self._normalize_task_device_type(task_details.get('device_type', '')),
|
||
'target_worker_ids': task_details.get('target_worker_ids', []),
|
||
'routing_mode': task_details.get('routing_mode', 'public'),
|
||
'last_fail_type': status.get('last_fail_type'),
|
||
'last_fail_message': status.get('last_fail_message'),
|
||
'start_time': status.get('start_time'),
|
||
'end_time': status.get('end_time'),
|
||
'last_fail_time': status.get('last_fail_time'),
|
||
}
|
||
if status.get('status') == 'running':
|
||
running_tasks.append(task_entry)
|
||
elif status.get('status') == 'failed':
|
||
failed_tasks.append(task_entry)
|
||
|
||
queue_task_keys = self._get_queued_task_keys()
|
||
pending_total = len(queue_task_keys)
|
||
safe_offset = max(pending_offset, 0)
|
||
safe_limit = max(pending_limit, 1)
|
||
pending_window = queue_task_keys[safe_offset:safe_offset + safe_limit]
|
||
for task_key in pending_window:
|
||
pending_tasks.append(self._build_pending_task_entry(task_key, worker_lookup))
|
||
|
||
running_tasks.sort(key=lambda item: item.get('start_time') or 0, reverse=True)
|
||
failed_tasks.sort(key=lambda item: item.get('last_fail_time') or 0, reverse=True)
|
||
|
||
return {
|
||
'running': running_tasks[:recent_limit],
|
||
'pending': pending_tasks,
|
||
'failed': failed_tasks[:recent_limit],
|
||
'pending_total': pending_total,
|
||
'pending_offset': safe_offset,
|
||
'pending_limit': safe_limit,
|
||
'pending_window_end': min(safe_offset + safe_limit, pending_total),
|
||
}
|
||
|
||
def get_dashboard_summary(self) -> Dict[str, Any]:
|
||
stats = self.get_statistics()
|
||
workers = self.get_dashboard_workers()
|
||
total_tasks = stats['pending'] + stats['running'] + stats['completed'] + stats['failed']
|
||
managed_workers = [worker for worker in workers if worker.get('is_managed')]
|
||
online_workers = [worker for worker in managed_workers if worker.get('online')]
|
||
idle_workers = [worker for worker in online_workers if worker.get('status') == 'idle']
|
||
busy_workers = [worker for worker in online_workers if worker.get('status') == 'busy']
|
||
|
||
return {
|
||
**stats,
|
||
'managed_workers_total': len(managed_workers),
|
||
'managed_workers_online': len(online_workers),
|
||
'managed_workers_idle': len(idle_workers),
|
||
'managed_workers_busy': len(busy_workers),
|
||
'completion_rate': round((stats['completed'] / total_tasks) * 100, 2) if total_tasks else 0.0,
|
||
'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
}
|
||
|
||
def get_dashboard_snapshot(
|
||
self,
|
||
pending_offset: int = 0,
|
||
pending_limit: int = 50,
|
||
recent_limit: int = 20,
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
'summary': self.get_dashboard_summary(),
|
||
'workers': self.get_dashboard_workers(),
|
||
'tasks': self.get_dashboard_tasks(
|
||
pending_offset=pending_offset,
|
||
pending_limit=pending_limit,
|
||
recent_limit=recent_limit,
|
||
),
|
||
}
|
||
|
||
def get_monitoring_overview(self, date_str: Optional[str] = None, worker_id: Optional[str] = None) -> Dict[str, Any]:
|
||
return self.monitor.get_overview(date_str=date_str, dispatcher=self, worker_id=worker_id)
|
||
|
||
def get_monitoring_timeline(
|
||
self,
|
||
date_str: Optional[str] = None,
|
||
bucket_minutes: int = 15,
|
||
worker_id: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
return self.monitor.get_timeline(
|
||
date_str=date_str,
|
||
dispatcher=self,
|
||
bucket_minutes=bucket_minutes,
|
||
worker_id=worker_id,
|
||
)
|
||
|
||
def get_monitoring_distributions(self, date_str: Optional[str] = None, worker_id: Optional[str] = None) -> Dict[str, Any]:
|
||
return self.monitor.get_distributions(date_str=date_str, worker_id=worker_id, dispatcher=self)
|
||
|
||
def get_monitoring_workers(self, date_str: Optional[str] = None) -> List[Dict[str, Any]]:
|
||
return self.monitor.get_worker_stats(date_str=date_str, dispatcher=self)
|
||
|
||
def get_monitoring_tasks(
|
||
self,
|
||
date_str: Optional[str] = None,
|
||
worker_id: Optional[str] = None,
|
||
limit: int = 200,
|
||
) -> List[Dict[str, Any]]:
|
||
return self.monitor.get_task_rows(date_str=date_str, worker_id=worker_id, limit=limit)
|
||
|
||
def reset_monitoring_dashboard(self) -> Dict[str, Any]:
|
||
return self.monitor.reset_monitoring_data(worker_snapshots=self.get_dashboard_workers())
|
||
|
||
def sync_worker_registry(self):
|
||
"""启动时同步：以 worker_inventory.json 为准清理 Redis 中的残留 Worker 数据"""
|
||
valid_ids = set(self.worker_inventory.keys())
|
||
|
||
# 清理 workers:info 中不在 inventory 的历史 worker
|
||
all_info_ids = set(self.redis.hgetall("workers:info").keys())
|
||
stale_info_ids = all_info_ids - valid_ids
|
||
for stale_id in stale_info_ids:
|
||
self.redis.hdel("workers:info", stale_id)
|
||
|
||
# 清理 idle/busy 集合中不在 inventory 的条目
|
||
for set_key in ("workers:idle", "workers:busy"):
|
||
for member in self.redis.smembers(set_key):
|
||
if member not in valid_ids:
|
||
self.redis.srem(set_key, member)
|
||
|
||
# 清理 worker:tasks 映射中不在 inventory 的条目
|
||
all_task_mappings = self.redis.hgetall("worker:tasks")
|
||
for wid in list(all_task_mappings.keys()):
|
||
if wid not in valid_ids:
|
||
self.redis.hdel("worker:tasks", wid)
|
||
|
||
cleaned = len(stale_info_ids)
|
||
if cleaned:
|
||
logger.info(f"[启动同步] 已清理 {cleaned} 个不在 inventory 中的历史 Worker: {sorted(stale_info_ids)}")
|
||
else:
|
||
logger.info("[启动同步] Worker 注册表与 inventory 一致，无需清理")
|
||
|
||
def cleanup_stale_workers(self, timeout=7200):
|
||
"""清理超时的Worker
|
||
|
||
Args:
|
||
timeout: Worker超时时间(秒)，默认7200秒(120分钟)
|
||
"""
|
||
current_time = time.time()
|
||
workers_info = self.redis.hgetall("workers:info")
|
||
cleaned_count = 0
|
||
|
||
for worker_id, info_json in workers_info.items():
|
||
info = json.loads(info_json)
|
||
last_update = info.get('last_update', info.get('register_time', 0))
|
||
|
||
if current_time - last_update > timeout:
|
||
# Worker超时，清理
|
||
self.monitor.mark_worker_offline(worker_id, "heartbeat_timeout")
|
||
self.redis.srem("workers:idle", worker_id)
|
||
self.redis.srem("workers:busy", worker_id)
|
||
self.redis.hdel("workers:info", worker_id)
|
||
|
||
# 获取Worker的任务并标记为失败
|
||
current_task = self.redis.hget("worker:tasks", worker_id)
|
||
if current_task:
|
||
# 记录任务失败并更新运行态/分析态
|
||
worker_timeout_error = ErrorInfo.infra(InfraError.WORKER_TIMEOUT, "Worker heartbeat timeout")
|
||
self._record_failed_task(current_task, worker_id, worker_timeout_error, "Worker heartbeat timeout")
|
||
self.redis.hdel("worker:tasks", worker_id)
|
||
|
||
cleaned_count += 1
|
||
logger.info(f"[清理] Worker {worker_id} 超时，已清理")
|
||
|
||
return cleaned_count
|
||
|
||
def _handle_infra_error_recovery_alert(self, task_key: str, worker_id: str, error_type_str: str, message: str) -> bool:
|
||
alert_key = f"alert:{worker_id}:infra_errors"
|
||
pending_key = self._worker_recovery_pending_key(worker_id)
|
||
observe_key = self._worker_recovery_observe_key(worker_id)
|
||
current_time = time.time()
|
||
|
||
pipe = self.redis.pipeline()
|
||
pipe.zremrangebyscore(alert_key, 0, current_time - ALERT_WINDOW_SECONDS)
|
||
pipe.zadd(alert_key, {str(current_time): current_time})
|
||
pipe.zcard(alert_key)
|
||
pipe.expire(alert_key, ALERT_WINDOW_SECONDS + 60)
|
||
_, _, count, _ = pipe.execute()
|
||
|
||
if count < ALERT_THRESHOLD:
|
||
return False
|
||
|
||
self.redis.delete(alert_key)
|
||
self.redis.delete(observe_key)
|
||
self.redis.delete(pending_key)
|
||
logger.warning(
|
||
f"[Worker基础设施错误高频告警] Worker {worker_id} 高频基础设施错误，"
|
||
f"{count}次/{ALERT_WINDOW_SECONDS//60}分钟，自动重启已关闭，请手动处理"
|
||
)
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker基础设施错误高频告警】\nWorker: {worker_id}\n任务: {task_key}\n"
|
||
f"频率: {count}次/{ALERT_WINDOW_SECONDS//60}分钟\n最新错误: {error_type_str}\n消息: {message}\n"
|
||
f"自动重启已关闭，请手动处理"
|
||
)
|
||
return True
|
||
|
||
def _dispatch_alert(self, task_key, worker_id, error_info: ErrorInfo, message, retry_count, download_errors=None):
|
||
"""根据错误类型分发告警逻辑
|
||
|
||
Args:
|
||
task_key: 任务键
|
||
worker_id: Worker ID
|
||
error_info: 错误信息对象
|
||
message: 错误消息
|
||
retry_count: 当前重试次数
|
||
"""
|
||
error_type_str = f"{error_info.category.name}/{error_info.code}"
|
||
|
||
infra_recovery_triggered = False
|
||
if error_info.category == ErrorCategory.INFRA_ERROR:
|
||
infra_recovery_triggered = self._handle_infra_error_recovery_alert(
|
||
task_key,
|
||
worker_id,
|
||
error_type_str,
|
||
message,
|
||
)
|
||
|
||
if error_info.category == ErrorCategory.INFRA_ERROR and error_info.code == InfraError.WORKER_TIMEOUT:
|
||
logger.error(f"[告警推送] Worker超时: {worker_id}")
|
||
if not infra_recovery_triggered:
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker超时】\nWorker: {worker_id}\n任务: {task_key}\n消息: {message}"
|
||
)
|
||
return
|
||
|
||
if error_info.is_no_retry(download_errors=download_errors):
|
||
logger.info(f"[直接失败] 任务 {task_key} 不可重试错误: {error_type_str}, Worker: {worker_id}")
|
||
return
|
||
|
||
alert_key = f"alert:{worker_id}:all_errors"
|
||
observe_key = self._worker_recovery_observe_key(worker_id)
|
||
pending_key = self._worker_recovery_pending_key(worker_id)
|
||
current_time = time.time()
|
||
|
||
pipe = self.redis.pipeline()
|
||
pipe.zremrangebyscore(alert_key, 0, current_time - ALERT_WINDOW_SECONDS)
|
||
pipe.zadd(alert_key, {str(current_time): current_time})
|
||
pipe.zcard(alert_key)
|
||
pipe.expire(alert_key, ALERT_WINDOW_SECONDS + 60)
|
||
_, _, count, _ = pipe.execute()
|
||
|
||
if count < ALERT_THRESHOLD or infra_recovery_triggered:
|
||
return
|
||
|
||
meta = self._get_worker_meta(worker_id)
|
||
recovery_in_effect = bool(meta.get('auto_recover_used')) and (
|
||
self.redis.exists(pending_key) or self.redis.exists(observe_key)
|
||
)
|
||
self.redis.delete(alert_key)
|
||
|
||
if recovery_in_effect:
|
||
self.redis.delete(observe_key)
|
||
self.redis.delete(pending_key)
|
||
self._update_worker_meta(worker_id, auto_recover_used=False)
|
||
|
||
logger.warning(
|
||
f"[Worker高频错误告警] Worker {worker_id} 自动恢复后再次高频错误，"
|
||
f"自动重启已关闭，请手动处理"
|
||
)
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker高频错误告警】\nWorker: {worker_id}\n频率: {count}次/{ALERT_WINDOW_SECONDS//60}分钟\n最新错误: {error_type_str}\n消息: {message}\n自动恢复后再次触发连续错误，自动重启已关闭，请手动处理"
|
||
)
|
||
return
|
||
|
||
self.redis.delete(observe_key)
|
||
self.redis.delete(pending_key)
|
||
pipeline_result = self.run_dashboard_action([worker_id], 'run_pipeline')[0]
|
||
self._mark_worker_action(worker_id, 'run_pipeline', pipeline_result)
|
||
|
||
if pipeline_result.get('ok'):
|
||
self.redis.setex(pending_key, WORKER_REINIT_TIMEOUT, str(time.time()))
|
||
self._update_worker_meta(worker_id, auto_recover_used=True)
|
||
logger.warning(f"[Worker自动恢复] Worker {worker_id} 高频错误，已执行默认恢复 pipeline")
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker高频错误自动恢复】\nWorker: {worker_id}\n频率: {count}次/{ALERT_WINDOW_SECONDS//60}分钟\n最新错误: {error_type_str}\n消息: {message}\n已执行 dashboard 默认恢复链路"
|
||
)
|
||
return
|
||
|
||
self.redis.delete(pending_key)
|
||
self._update_worker_meta(worker_id, auto_recover_used=False)
|
||
logger.error(f"[Worker自动恢复失败] Worker {worker_id} 高频错误后执行默认恢复 pipeline 失败")
|
||
self.notifier.send_weCom_alert(
|
||
worker_id,
|
||
f"【Worker高频错误自动恢复失败】\nWorker: {worker_id}\n频率: {count}次/{ALERT_WINDOW_SECONDS//60}分钟\n最新错误: {error_type_str}\n消息: {message}\n恢复结果: {pipeline_result.get('message', '')}"
|
||
)
|
||
|
||
def _record_failed_task(self, task_key, worker_id, error_info: ErrorInfo, message='', statistics=None, crashed_source=None, download_errors=None):
|
||
"""记录失败任务，分流重试和最终失败
|
||
|
||
- APP_ERROR.CRASH / APP_ERROR.ROOT_MODE_UNSUPPORTED: 若当前任务不是 physical，则自动改派到真机并重新入队
|
||
- APP_ERROR / BUSINESS_ERROR: 默认直接标记为永久失败
|
||
- 满足终态下载组合条件的 DOWNLOAD_ERROR（Google Play 终态类错误 + local 任意失败）→ 直接标记为永久失败
|
||
- 可重试错误 → 等待应用画像决定是否重新入队
|
||
|
||
Args:
|
||
task_key: 任务键
|
||
worker_id: Worker标识符
|
||
error_info: 错误信息对象
|
||
message: 附加消息/告警信息
|
||
statistics: 统计数据字典
|
||
crashed_source: 闪退的下载源（已废弃，保留参数兼容）
|
||
"""
|
||
if error_info.category == ErrorCategory.SUCCESS and error_info.code == 0:
|
||
logger.warning(f"[异常上报] Worker {worker_id} 上报失败但错误码为 SUCCESS/0，按成功处理: {task_key}")
|
||
self._complete_task(task_key, worker_id, 'success', message or 'worker reported failure with SUCCESS/0, treated as success')
|
||
return
|
||
|
||
is_no_retry_error = error_info.is_no_retry(download_errors=download_errors)
|
||
is_app_crash = (error_info.category == ErrorCategory.APP_ERROR and error_info.code == AppError.CRASH)
|
||
is_terminal_local_fallback_failed = (
|
||
error_info.category == ErrorCategory.DOWNLOAD_ERROR
|
||
and error_info.code in {
|
||
DownloadError.REGION_RESTRICTED,
|
||
DownloadError.APP_NOT_FOUND,
|
||
DownloadError.INCOMPATIBLE,
|
||
DownloadError.COUNTRY_LOCKED,
|
||
}
|
||
and is_no_retry_error
|
||
)
|
||
|
||
prev_status_json = self.redis.hget("task:status", task_key)
|
||
retry_count = 0
|
||
if prev_status_json:
|
||
prev_status = json.loads(prev_status_json)
|
||
retry_count = prev_status.get('retry_count', 0)
|
||
|
||
new_retry_count = retry_count + 1
|
||
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
task_details = {}
|
||
app_name = ""
|
||
package_name = ""
|
||
if task_details_json:
|
||
task_details = json.loads(task_details_json)
|
||
app_name = task_details.get('app_name', '')
|
||
package_name = task_details.get('package_name', '')
|
||
app_magic_label = str(task_details.get('app_magic_label') or '').strip()
|
||
is_block_task = bool(task_details.get('is_block_task', False))
|
||
|
||
error_type_str = f"{error_info.category.name}/{error_info.code}"
|
||
task_status = {
|
||
'status': 'failed',
|
||
'worker_id': worker_id,
|
||
'retry_count': new_retry_count,
|
||
'last_fail_time': time.time(),
|
||
'last_fail_type': error_type_str,
|
||
'last_fail_message': message
|
||
}
|
||
|
||
if statistics:
|
||
task_status['statistics'] = statistics
|
||
|
||
is_download_error = (
|
||
error_info.category == ErrorCategory.DOWNLOAD_ERROR
|
||
or (error_info.category == ErrorCategory.INFRA_ERROR and error_info.code == InfraError.DOWNLOAD_FAILED)
|
||
)
|
||
|
||
is_local_install_failed = (
|
||
is_download_error
|
||
and message
|
||
and ("local: install failed" in message.lower()
|
||
or "local: Install" in message)
|
||
)
|
||
if is_local_install_failed and self._uses_central_apk_cache():
|
||
prev_status = json.loads(prev_status_json) if prev_status_json else {}
|
||
if prev_status.get("force_reapk_done"):
|
||
self._log_apk_install_failure(task_key, package_name, app_name, message)
|
||
logger.warning("[自动重下] %s 已重下过一次仍失败，不再重试: %s", task_key, message)
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
self.monitor.set_failure_bucket(task_key, "failed")
|
||
self.redis.sadd("task:failed", task_key)
|
||
return
|
||
logger.info("[自动重下] %s 本地APK安装失败，触发force_reapk后重试", task_key)
|
||
self.force_reapk(package_name)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "pending_retry",
|
||
"worker_id": worker_id,
|
||
"retry_count": new_retry_count,
|
||
"last_fail_time": time.time(),
|
||
"last_fail_type": error_type_str,
|
||
"last_fail_message": message,
|
||
"force_reapk_done": True,
|
||
}))
|
||
self.monitor.set_failure_bucket(task_key, "retry")
|
||
self.redis.srem("task:failed", task_key)
|
||
return
|
||
|
||
if is_block_task:
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
if is_download_error:
|
||
last_updated = str(task_details.get("last_updated") or "").strip()
|
||
registry = self._get_apk_registry() if self._uses_central_apk_cache() else None
|
||
if registry and registry.is_fresh_enough(package_name, last_updated):
|
||
self.monitor.set_failure_bucket(task_key, "retry")
|
||
self.redis.srem("task:failed", task_key)
|
||
retry_task_details = self._mark_retry_avoid_worker(task_key, task_details, worker_id)
|
||
retry_status = dict(task_status)
|
||
retry_status["status"] = "pending"
|
||
self.redis.hset("task:status", task_key, json.dumps(retry_status))
|
||
self._queue_retry_task(task_key, retry_task_details, prioritize=True)
|
||
logger.info(f"[Block测试] 任务 {task_key} 下载失败({error_type_str})，本地已有APK，立即重试")
|
||
else:
|
||
self._mark_download_error_awaiting_apk(package_name,
|
||
app_name=app_name,
|
||
last_updated=last_updated)
|
||
self.monitor.set_failure_bucket(task_key, "retry")
|
||
self.redis.srem("task:failed", task_key)
|
||
retry_task_details = self._mark_retry_avoid_worker(task_key, task_details, worker_id)
|
||
retry_status = dict(task_status)
|
||
retry_status["status"] = "pending_retry"
|
||
self.redis.hset("task:status", task_key, json.dumps(retry_status))
|
||
if self._uses_central_apk_cache():
|
||
logger.info(f"[Block测试] 任务 {task_key} 下载失败({error_type_str})，等待中心 APK 缓存刷新后重试一次")
|
||
else:
|
||
logger.info(f"[Block测试] 任务 {task_key} 下载失败({error_type_str})，direct 模式不等待中心 APK")
|
||
else:
|
||
self.monitor.set_failure_bucket(task_key, "failed")
|
||
self.redis.sadd("task:failed", task_key)
|
||
logger.info(f"[Block测试] 任务 {task_key} 因非下载错误({error_type_str})直接标记为永久失败")
|
||
self._dispatch_alert(task_key, worker_id, error_info, message, retry_count, download_errors=download_errors)
|
||
return
|
||
|
||
auto_physical_reroute = self._should_auto_reroute_to_physical(task_details, error_info)
|
||
|
||
self.redis.hset("task:status", task_key, json.dumps(task_status))
|
||
|
||
related_magic_label_collected = (
|
||
bool(app_magic_label)
|
||
and bool(package_name)
|
||
and self.analytics.repo.has_successful_related_magic_label(app_magic_label, package_name)
|
||
)
|
||
if related_magic_label_collected:
|
||
qualified_status = dict(task_status)
|
||
qualified_status['status'] = 'qualified'
|
||
qualified_status['end_time'] = time.time()
|
||
qualified_status['collection_status_reason'] = '关联应用已采集'
|
||
self.redis.hset("task:status", task_key, json.dumps(qualified_status))
|
||
self.redis.srem("task:failed", task_key)
|
||
self.redis.sadd("task:completed", task_key)
|
||
self.monitor.set_failure_bucket(task_key, "light_restricted")
|
||
logger.warning(
|
||
f"[轻度受限] 任务 {task_key} 测试失败，但 app_magic_label={app_magic_label} 的关联应用已采集成功，"
|
||
f"当前任务按轻度受限记录"
|
||
)
|
||
return
|
||
|
||
if auto_physical_reroute:
|
||
self.monitor.set_failure_bucket(task_key, "retry")
|
||
rerouted_task_details = self._build_physical_reroute_task_details(task_details)
|
||
rerouted_task_details['auto_physical_fallback_applied'] = True
|
||
rerouted_task_details['auto_physical_fallback_from_worker'] = worker_id
|
||
rerouted_task_details['auto_physical_fallback_error_type'] = error_type_str
|
||
self.redis.hset("task:details", task_key, json.dumps(rerouted_task_details))
|
||
|
||
reroute_status = dict(task_status)
|
||
reroute_status['status'] = 'pending'
|
||
reroute_status['retry_count'] = retry_count
|
||
self.redis.hset("task:status", task_key, json.dumps(reroute_status))
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
self._queue_retry_task(task_key, rerouted_task_details, prioritize=self._is_hot_new_app_task(rerouted_task_details))
|
||
logger.warning(
|
||
f"[真机改派] 任务 {task_key} 在 Worker {worker_id} 上报 {error_type_str}，"
|
||
f"已改为仅 physical 执行并重新放回待分发"
|
||
)
|
||
elif is_app_crash:
|
||
self.monitor.set_failure_bucket(task_key, "failed")
|
||
self.redis.sadd("task:failed", task_key)
|
||
logger.error(f"[永久失败] 任务 {task_key} 应用闪退，直接标记为永久失败")
|
||
elif is_no_retry_error and not is_terminal_local_fallback_failed:
|
||
# 非下载类的不可重试错误（APP_ERROR / BUSINESS_ERROR）→ 直接永久失败
|
||
self.monitor.set_failure_bucket(task_key, "failed")
|
||
self.redis.sadd("task:failed", task_key)
|
||
logger.error(f"[永久失败] 任务 {task_key} 因不可重试错误({error_type_str})，直接标记为永久失败")
|
||
else:
|
||
# 可重试的下载错误 或 终态下载错误（REGION_RESTRICTED等 + local失败）
|
||
# 中心 APK 缓存模式下等待 US/MinIO 下载；direct 模式下由 Worker 自行下载。
|
||
if self._uses_central_apk_cache():
|
||
if error_info.category in (ErrorCategory.DOWNLOAD_ERROR,):
|
||
self._mark_download_error_awaiting_apk(package_name)
|
||
elif error_info.category == ErrorCategory.INFRA_ERROR and error_info.code == InfraError.DOWNLOAD_FAILED:
|
||
self._mark_download_error_awaiting_apk(package_name)
|
||
|
||
if is_terminal_local_fallback_failed and self._uses_central_apk_cache():
|
||
logger.warning(
|
||
f"[等待中心APK] 任务 {task_key} Google Play 终态错误({error_type_str})且本地兜底失败，"
|
||
f"标记为等待中心 APK 下载服务（可能通过不同地区或 APKPure 成功下载）"
|
||
)
|
||
|
||
self.monitor.set_failure_bucket(task_key, "retry")
|
||
self.redis.srem("task:failed", task_key)
|
||
retry_task_details = self._mark_retry_avoid_worker(task_key, task_details, worker_id)
|
||
retry_status = dict(task_status)
|
||
retry_status["status"] = "pending_retry"
|
||
self.redis.hset("task:status", task_key, json.dumps(retry_status))
|
||
if self._is_hot_new_app_task(retry_task_details):
|
||
self._queue_retry_task(task_key, retry_task_details, prioritize=True)
|
||
logger.info(
|
||
f"[热门重试] 新增热门应用任务 {task_key} 第{new_retry_count}次失败，"
|
||
f"已优先重新入队并避开 Worker {worker_id}"
|
||
)
|
||
logger.info(f"[等待分析] 任务 {task_key} 已记录为可重试（第{new_retry_count}次失败），等待应用画像决定是否重新入队")
|
||
self._dispatch_alert(task_key, worker_id, error_info, message, retry_count, download_errors=download_errors)
|
||
|
||
def clear_all(self):
|
||
"""清空所有任务数据（谨慎使用）"""
|
||
keys_to_delete = [
|
||
*self.TASK_QUEUE_CLEANUP_NAMES,
|
||
"task:status",
|
||
"task:details",
|
||
"worker:tasks",
|
||
"task:completed",
|
||
"task:failed",
|
||
"workers:info",
|
||
"workers:idle",
|
||
"workers:busy",
|
||
"workers:meta",
|
||
]
|
||
for key in keys_to_delete:
|
||
self.redis.delete(key)
|
||
for key in self.redis.scan_iter(match="alert:*"):
|
||
self.redis.delete(key)
|
||
for key in self.redis.scan_iter(match="cooldown:*"):
|
||
self.redis.delete(key)
|
||
for key in self.redis.scan_iter(match="worker:recovery_observe:*"):
|
||
self.redis.delete(key)
|
||
for key in self.redis.scan_iter(match="worker:recovery_pending:*"):
|
||
self.redis.delete(key)
|
||
logger.info("已清空所有任务数据")
|
||
self._apk_registry = None
|
||
self._minio_storage = None
|
||
|
||
# ── APK prefetch integration ───────────────────────────────────
|
||
|
||
def _get_apk_registry(self):
|
||
if self._apk_registry is None:
|
||
from apk_cloud.registry import ApkRegistry
|
||
self._apk_registry = ApkRegistry()
|
||
return self._apk_registry
|
||
|
||
def _get_minio_storage(self):
|
||
"""获取 MinioStorage 实例，如果 MINIO_ENABLED=False 则返回 None"""
|
||
if self._apk_download_mode() != "minio":
|
||
return None
|
||
if self._minio_storage is None:
|
||
from config import MINIO_ENABLED
|
||
if not MINIO_ENABLED:
|
||
return None
|
||
from apk_cloud.storage import MinioStorage
|
||
self._minio_storage = MinioStorage.create() # 使用工厂方法
|
||
return self._minio_storage
|
||
|
||
def _resolve_apk_for_task(self, package_name: str, last_updated: str) -> dict:
|
||
if not self._uses_central_apk_cache():
|
||
return {}
|
||
registry = self._get_apk_registry()
|
||
if not registry.is_fresh_enough(package_name, last_updated):
|
||
return {}
|
||
smb_dir = registry.get_smb_apk_dir(package_name)
|
||
if not smb_dir:
|
||
return {}
|
||
entry = registry.get(package_name) or {}
|
||
apk_files = entry.get("apk_files", [])
|
||
return {
|
||
"local_apk_dir": smb_dir,
|
||
"local_apk_files": apk_files,
|
||
"apk_download_date": entry.get("download_date", ""),
|
||
}
|
||
|
||
def _enrich_task_with_apk(self, task_details: dict) -> dict:
|
||
package_name = str(task_details.get("package_name") or "").strip()
|
||
last_updated = str(task_details.get("last_updated") or "").strip()
|
||
if not package_name:
|
||
return task_details
|
||
apk_info = self._resolve_apk_for_task(package_name, last_updated)
|
||
if apk_info:
|
||
task_details = dict(task_details)
|
||
task_details["available_sources"] = ["local", "google_play", "apkpure"]
|
||
task_details["local_apk_dir"] = apk_info["local_apk_dir"]
|
||
task_details["local_apk_files"] = apk_info["local_apk_files"]
|
||
return task_details
|
||
|
||
def _mark_download_error_awaiting_apk(self, package_name: str, app_name: str = "", last_updated: str = "", force: bool = False) -> None:
|
||
if not package_name:
|
||
return
|
||
if not self._uses_central_apk_cache():
|
||
logger.info(
|
||
"[APK] 当前为 direct 模式，Worker 将自行下载，跳过中心 APK 等待队列: %s",
|
||
package_name,
|
||
)
|
||
return
|
||
registry = self._get_apk_registry()
|
||
if not force and registry.is_fresh_enough(package_name, last_updated):
|
||
logger.info("[APK] 本地已有新鲜APK，无需等待中心 APK 缓存刷新: %s", package_name)
|
||
return
|
||
registry.mark_pending_download(package_name)
|
||
logger.info("[APK] 下载失败，标记为等待APK: %s", package_name)
|
||
try:
|
||
storage = self._get_minio_storage()
|
||
if storage is None:
|
||
logger.info("[APK] Minio 未启用，跳过优先下载队列推送: %s", package_name)
|
||
return
|
||
storage.prepend_to_download_queue([{
|
||
"package_name": package_name,
|
||
"app_name": app_name or package_name,
|
||
"last_updated": last_updated or "",
|
||
"force": force,
|
||
}])
|
||
logger.info("[APK] 下载失败，任务已加入下载队列开头: %s force=%s", package_name, force)
|
||
except Exception as exc:
|
||
logger.warning("[APK] 下载失败，推送优先下载队列失败: %s", exc)
|
||
|
||
def _fail_block_tasks_waiting_for_apk(self, package_name: str) -> None:
|
||
task_keys = self._find_task_keys_by_package(package_name)
|
||
for task_key in task_keys:
|
||
if not task_key.endswith("_block"):
|
||
continue
|
||
status_json = self.redis.hget("task:status", task_key)
|
||
if not status_json:
|
||
continue
|
||
status_data = json.loads(status_json)
|
||
if status_data.get("status") != "pending_retry":
|
||
continue
|
||
status_data["status"] = "failed"
|
||
status_data["last_fail_message"] = "中心 APK 下载失败，block任务永久失败"
|
||
self.redis.hset("task:status", task_key, json.dumps(status_data))
|
||
self.redis.sadd("task:failed", task_key)
|
||
logger.info(f"[Block测试] 中心 APK 下载失败，block任务 {task_key} 标记为永久失败")
|
||
|
||
def _log_apk_install_failure(self, task_key: str, package_name: str, app_name: str, message: str) -> None:
|
||
import csv, io
|
||
from config import REPORT_DIR
|
||
import os as _os
|
||
_os.makedirs(REPORT_DIR, exist_ok=True)
|
||
report_path = _os.path.join(REPORT_DIR, "apk_install_failures.csv")
|
||
existed = _os.path.isfile(report_path)
|
||
with open(report_path, "a", newline="", encoding="utf-8") as f:
|
||
writer = csv.writer(f)
|
||
if not existed:
|
||
writer.writerow(["时间", "任务键", "包名", "应用名", "失败原因"])
|
||
writer.writerow([
|
||
time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
task_key,
|
||
package_name,
|
||
app_name,
|
||
message,
|
||
])
|
||
|
||
def force_reapk(self, package_name: str) -> int:
|
||
registry = self._get_apk_registry()
|
||
pkg_dir = registry.storage_dir.rstrip("/\\") + "/" + package_name
|
||
import shutil
|
||
if os.path.isdir(pkg_dir):
|
||
shutil.rmtree(pkg_dir, ignore_errors=True)
|
||
logger.info("[force_reapk] 已清除本地APK目录: %s", pkg_dir)
|
||
try:
|
||
with registry._connect() as conn:
|
||
conn.execute("DELETE FROM apk_registry WHERE package_name = ?", (package_name,))
|
||
except Exception:
|
||
pass
|
||
registry.clear_pending_download(package_name)
|
||
|
||
task_keys = self._find_task_keys_by_package(package_name)
|
||
task_info = None
|
||
for tk in task_keys:
|
||
d = self.redis.hget("task:details", tk)
|
||
if d:
|
||
task_info = json.loads(d)
|
||
break
|
||
app_name = ""
|
||
last_updated = ""
|
||
if task_info:
|
||
app_name = str(task_info.get("app_name") or "").strip()
|
||
last_updated = str(task_info.get("last_updated") or "").strip()
|
||
self._mark_download_error_awaiting_apk(package_name, app_name=app_name, last_updated=last_updated, force=True)
|
||
|
||
for task_key in task_keys:
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "pending_retry",
|
||
"end_time": time.time(),
|
||
"retry_count": 0,
|
||
"last_fail_message": "force_reapk: 等待重新下载",
|
||
}))
|
||
logger.info("[force_reapk] 已重置 %s (%d 个任务)，等待中心 APK 缓存刷新", package_name, len(task_keys))
|
||
return len(task_keys)
|
||
|
||
def _re_enqueue_apk_tasks_for_package(self, package_name: str) -> int:
|
||
task_keys = self._find_task_keys_by_package(package_name)
|
||
re_enqueued = 0
|
||
for task_key in task_keys:
|
||
status_json = self.redis.hget("task:status", task_key)
|
||
if not status_json:
|
||
continue
|
||
try:
|
||
status = json.loads(status_json)
|
||
except (json.JSONDecodeError, TypeError):
|
||
continue
|
||
if status.get("status") not in ("pending_retry", "failed", "pending"):
|
||
continue
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
task_details = json.loads(task_details_json) if task_details_json else {}
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "pending",
|
||
"end_time": time.time(),
|
||
"retry_count": int(status.get("retry_count") or 0),
|
||
}))
|
||
self._queue_retry_task(task_key, task_details)
|
||
re_enqueued += 1
|
||
logger.info("[APK] APK就绪，重新入队: %s (%s)", task_key, package_name)
|
||
registry = self._get_apk_registry()
|
||
registry.clear_pending_download(package_name)
|
||
return re_enqueued
|
||
|
||
def _try_re_enqueue_apk_ready_tasks(self) -> int:
|
||
registry = self._get_apk_registry()
|
||
ready_packages = registry.get_pending_that_are_available()
|
||
if not ready_packages:
|
||
return 0
|
||
|
||
re_enqueued = 0
|
||
for package_name in ready_packages:
|
||
task_keys = self._find_task_keys_by_package(package_name)
|
||
for task_key in task_keys:
|
||
status_json = self.redis.hget("task:status", task_key)
|
||
if not status_json:
|
||
continue
|
||
try:
|
||
status = json.loads(status_json)
|
||
except (json.JSONDecodeError, TypeError):
|
||
continue
|
||
if status.get("status") not in ("pending_retry", "failed", "pending"):
|
||
continue
|
||
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
task_details = json.loads(task_details_json) if task_details_json else {}
|
||
|
||
self._remove_task_from_queue(task_key)
|
||
self.redis.srem("task:completed", task_key)
|
||
self.redis.srem("task:failed", task_key)
|
||
self.redis.hset("task:status", task_key, json.dumps({
|
||
"status": "pending",
|
||
"end_time": time.time(),
|
||
"retry_count": int(status.get("retry_count") or 0),
|
||
}))
|
||
self._queue_retry_task(task_key, task_details)
|
||
re_enqueued += 1
|
||
logger.info("[APK] APK就绪，重新入队: %s (%s)", task_key, package_name)
|
||
|
||
registry.clear_pending_download(package_name)
|
||
return re_enqueued
|
||
|
||
def _find_task_keys_by_package(self, package_name: str) -> List[str]:
|
||
found = []
|
||
all_details = self.redis.hgetall("task:details")
|
||
for task_key, details_json in (all_details or {}).items():
|
||
try:
|
||
details = json.loads(details_json)
|
||
except (json.JSONDecodeError, TypeError):
|
||
continue
|
||
if str(details.get("package_name") or "").strip() == package_name:
|
||
found.append(task_key)
|
||
return found
|
||
|
||
def sync_download_queue(self) -> int:
|
||
"""同步下载队列（仅当 Minio 启用时）"""
|
||
try:
|
||
storage = self._get_minio_storage()
|
||
if storage is None:
|
||
# Minio 未启用，跳过队列同步
|
||
return 0
|
||
except Exception as exc:
|
||
logger.warning("[APK] MinIO不可达，跳过下载队列同步: %s", exc)
|
||
return 0
|
||
|
||
registry = self._get_apk_registry()
|
||
|
||
current_queue = storage.pull_download_queue()
|
||
filtered = []
|
||
removed_completed = 0
|
||
removed_failed = 0
|
||
for task in current_queue:
|
||
pkg = str(task.get("package_name") or "").strip()
|
||
last_updated = str(task.get("last_updated") or "").strip()
|
||
if not pkg:
|
||
continue
|
||
if registry.is_fresh_enough(pkg, last_updated):
|
||
removed_completed += 1
|
||
continue
|
||
if last_updated and registry.has_failed_with_same_version(pkg, last_updated):
|
||
removed_failed += 1
|
||
continue
|
||
filtered.append(task)
|
||
|
||
if removed_completed or removed_failed:
|
||
logger.info("[APK] 队列清理: 移除已完成 %d, 移除已失败 %d, 剩余 %d",
|
||
removed_completed, removed_failed, len(filtered))
|
||
|
||
pending_pkgs = registry.get_all_pending_downloads()
|
||
existing_pkgs = {t.get("package_name", "") for t in filtered}
|
||
new_tasks = []
|
||
for pkg in pending_pkgs:
|
||
if pkg in existing_pkgs:
|
||
continue
|
||
task_info = self._find_task_keys_by_package(pkg)
|
||
app_name = ""
|
||
last_updated = ""
|
||
for task_key in task_info:
|
||
details_json = self.redis.hget("task:details", task_key)
|
||
if details_json:
|
||
details = json.loads(details_json)
|
||
app_name = str(details.get("app_name") or app_name or "").strip() or app_name
|
||
last_updated = str(details.get("last_updated") or last_updated or "").strip() or last_updated
|
||
if app_name and last_updated:
|
||
break
|
||
if last_updated and registry.has_failed_with_same_version(pkg, last_updated):
|
||
continue
|
||
new_tasks.insert(0, {
|
||
"package_name": pkg,
|
||
"app_name": app_name or pkg,
|
||
"last_updated": last_updated or "",
|
||
})
|
||
|
||
queue = new_tasks + filtered
|
||
max_queue = 1000
|
||
if len(queue) > max_queue:
|
||
logger.info("[APK] 下载队列超过上限 %d，截断尾部 %d 条", max_queue, len(queue) - max_queue)
|
||
queue = queue[:max_queue]
|
||
try:
|
||
storage.push_download_queue(queue)
|
||
except Exception as exc:
|
||
logger.warning("[APK] 推送下载队列失败: %s", exc)
|
||
return 0
|
||
|
||
if new_tasks:
|
||
logger.info("[APK] 队列新增 %d 个优先任务, 队列总数 %d", len(new_tasks), len(queue))
|
||
return len(new_tasks)
|
||
|
||
def _get_stale_download_tasks(self, registry) -> List[Dict[str, str]]:
|
||
if not hasattr(self, '_last_stale_check'):
|
||
self._last_stale_check = 0.0
|
||
if time.time() - self._last_stale_check < APK_DOWNLOAD_QUEUE_INTERVAL:
|
||
return []
|
||
self._last_stale_check = time.time()
|
||
|
||
try:
|
||
stale_records = registry.get_stale_records()
|
||
except Exception as exc:
|
||
logger.warning("[APK] 查询过期APK记录失败: %s", exc)
|
||
return []
|
||
|
||
tasks = []
|
||
for record in stale_records:
|
||
package_name = str(record.get("package_name") or "").strip()
|
||
if not package_name:
|
||
continue
|
||
tasks.append({
|
||
"package_name": package_name,
|
||
"app_name": str(record.get("app_name") or "").strip(),
|
||
"last_updated": str(record.get("last_updated") or "").strip(),
|
||
})
|
||
if tasks:
|
||
logger.info("[APK] 检测到 %d 个过期本地APK，将重新下载", len(tasks))
|
||
return tasks
|
||
|
||
def poll_apk_results(self) -> int:
|
||
"""轮询 APK 下载结果（仅当 Minio 启用时）"""
|
||
try:
|
||
storage = self._get_minio_storage()
|
||
if storage is None:
|
||
# Minio 未启用，跳过结果轮询
|
||
return 0
|
||
except Exception as exc:
|
||
logger.warning("[APK] MinIO不可达，跳过结果轮询: %s", exc)
|
||
return 0
|
||
|
||
try:
|
||
results = storage.list_download_results()
|
||
except Exception as exc:
|
||
logger.warning("[APK] 轮询下载结果失败: %s", exc)
|
||
return 0
|
||
|
||
if not results:
|
||
return self._try_re_enqueue_apk_ready_tasks()
|
||
|
||
registry = self._get_apk_registry()
|
||
imported = 0
|
||
for package_name, result in results.items():
|
||
status = result.get("status", "")
|
||
|
||
try:
|
||
self._record_result_to_db(package_name, result)
|
||
except Exception as exc:
|
||
logger.warning("[APK] 记录下载结果到DB失败 %s: %s", package_name, exc)
|
||
|
||
if status != "ok":
|
||
if status == "failed":
|
||
storage.delete_download_result(package_name)
|
||
registry.clear_pending_download(package_name)
|
||
self._fail_block_tasks_waiting_for_apk(package_name)
|
||
continue
|
||
try:
|
||
local_paths = storage.download_apk(package_name, result, registry.storage_dir)
|
||
if not registry.verify_apk_versions(package_name):
|
||
logger.warning("[APK] 版本码不一致，丢弃: %s", package_name)
|
||
import shutil
|
||
shutil.rmtree(registry.storage_dir + "/" + package_name, ignore_errors=True)
|
||
continue
|
||
registry.import_from_download_result(package_name, result, local_dir=registry.storage_dir)
|
||
storage.delete_download_result(package_name)
|
||
imported += 1
|
||
re_enqueued = self._re_enqueue_apk_tasks_for_package(package_name)
|
||
logger.info("[APK] 下载完成: %s (%d files), 重新入队 %d 个任务",
|
||
package_name, len(local_paths), re_enqueued)
|
||
version_dir = f"{result.get('download_date', '')}_{result.get('version_code', '')}"
|
||
if version_dir.strip("_"):
|
||
storage.delete_apk_version(package_name, version_dir)
|
||
logger.info("[APK] 已清理MinIO源文件: %s/%s", package_name, version_dir)
|
||
storage.cleanup_old_versions(package_name, keep=0)
|
||
except Exception as exc:
|
||
logger.warning("[APK] 处理结果失败 %s: %s", package_name, exc)
|
||
|
||
if imported > 0:
|
||
logger.info("[APK] 本轮导入%d个新APK，刷新待分发任务...", imported)
|
||
self.refresh_tasks_from_app_summary()
|
||
self._enforce_apk_storage_limits()
|
||
|
||
re_enqueued = self._try_re_enqueue_apk_ready_tasks()
|
||
if re_enqueued:
|
||
logger.info("[APK] 重新入队 %d 个等待APK的任务", re_enqueued)
|
||
return imported + re_enqueued
|
||
|
||
def _record_result_to_db(self, package_name: str, result: dict) -> None:
|
||
registry = self._get_apk_registry()
|
||
package_name = str(package_name or "").strip()
|
||
if not package_name:
|
||
return
|
||
|
||
task_keys = self._find_task_keys_by_package(package_name)
|
||
if not task_keys:
|
||
registry.record_download_from_result(package_name, "", result)
|
||
logger.info("[APK] 记录下载结果(无对应任务): %s status=%s",
|
||
package_name, result.get("status"))
|
||
return
|
||
|
||
for task_key in task_keys:
|
||
task_details_json = self.redis.hget("task:details", task_key)
|
||
if not task_details_json:
|
||
continue
|
||
try:
|
||
task_details = json.loads(task_details_json)
|
||
except (json.JSONDecodeError, TypeError):
|
||
continue
|
||
country_code = str(task_details.get("country_code") or "").strip()
|
||
registry.record_download_from_result(package_name, country_code, result)
|
||
logger.info("[APK] 记录下载结果: %s/%s status=%s",
|
||
package_name, country_code, result.get("status"))
|
||
break
|
||
|
||
def _manage_stale_apks(self) -> int:
|
||
"""定期清理过期本地 APK 文件，并推送到下载队列重新下载。
|
||
|
||
返回清理的过期 APK 数量。
|
||
"""
|
||
registry = self._get_apk_registry()
|
||
try:
|
||
stale_records = registry.get_stale_records()
|
||
except Exception as exc:
|
||
logger.warning("[APK] 查询过期 APK 失败: %s", exc)
|
||
return 0
|
||
|
||
if not stale_records:
|
||
return 0
|
||
|
||
cleaned = 0
|
||
for record in stale_records:
|
||
package_name = str(record.get("package_name") or "").strip()
|
||
if not package_name:
|
||
continue
|
||
try:
|
||
registry.cleanup_stale_local(package_name)
|
||
cleaned += 1
|
||
logger.info("[APK] 已清理过期本地 APK: %s", package_name)
|
||
except Exception as exc:
|
||
logger.warning("[APK] 清理过期 APK 失败 %s: %s", package_name, exc)
|
||
|
||
if cleaned > 0:
|
||
logger.info("[APK] 过期 APK 清理: %d 个包已删除，将在下次队列同步时重新下载", cleaned)
|
||
return cleaned
|
||
|
||
def _enforce_apk_storage_limits(self) -> None:
|
||
"""强制执行 APK 存储限制（仅当 Minio 启用时）"""
|
||
try:
|
||
storage = self._get_minio_storage()
|
||
if storage is None:
|
||
# Minio 未启用，跳过存储限制检查
|
||
return
|
||
except Exception:
|
||
return
|
||
try:
|
||
minio_result = storage.enforce_storage_limit()
|
||
if minio_result.get("deleted_versions"):
|
||
logger.info(
|
||
"[APK] MinIO 超限清理: %.1fGB → %.1fGB (删除 %d 个版本)",
|
||
minio_result["before_bytes"] / (1024 ** 3),
|
||
minio_result["after_bytes"] / (1024 ** 3),
|
||
minio_result["deleted_versions"],
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("[APK] MinIO 存储限制检查失败: %s", exc)
|