234 lines
8.1 KiB
Python
234 lines
8.1 KiB
Python
import time
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
from result_codes import AppError, BusinessError, DownloadError, InfraError
|
|
|
|
|
|
DOWNLOAD_NETWORK_ERROR_CODES = {
|
|
int(DownloadError.DOWNLOAD_TIMEOUT),
|
|
int(DownloadError.NETWORK_ERROR),
|
|
}
|
|
DOWNLOAD_NETWORK_MESSAGE_HINTS = (
|
|
"timeout",
|
|
"overtime",
|
|
"jump failed",
|
|
"app page load failed",
|
|
"page load failed",
|
|
"network error",
|
|
"connection error",
|
|
"超时",
|
|
"网络错误",
|
|
)
|
|
ERROR_ENUM_BY_CATEGORY = {
|
|
"DOWNLOAD_ERROR": DownloadError,
|
|
"INFRA_ERROR": InfraError,
|
|
"APP_ERROR": AppError,
|
|
"BUSINESS_ERROR": BusinessError,
|
|
}
|
|
|
|
|
|
def _parse_error_type(error_type: Optional[str]) -> Tuple[Optional[str], Optional[int]]:
|
|
if not error_type:
|
|
return None, None
|
|
try:
|
|
category, code = str(error_type).split("/", 1)
|
|
return category, int(code)
|
|
except (TypeError, ValueError):
|
|
return str(error_type), None
|
|
|
|
|
|
def _download_error_message_is_network(message: Optional[str]) -> bool:
|
|
if not message:
|
|
return False
|
|
lowered = str(message).lower()
|
|
return any(token in lowered for token in DOWNLOAD_NETWORK_MESSAGE_HINTS)
|
|
|
|
|
|
def _all_download_errors_are_network(download_errors: Any) -> bool:
|
|
if not isinstance(download_errors, dict) or not download_errors:
|
|
return False
|
|
for detail in download_errors.values():
|
|
if not isinstance(detail, dict):
|
|
return False
|
|
code = detail.get("code")
|
|
try:
|
|
if code is not None and int(code) in DOWNLOAD_NETWORK_ERROR_CODES:
|
|
continue
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if _download_error_message_is_network(detail.get("message")):
|
|
continue
|
|
return False
|
|
return True
|
|
|
|
|
|
def _failure_subtype_name(category: Optional[str], code: Optional[int]) -> str:
|
|
enum_cls = ERROR_ENUM_BY_CATEGORY.get(category)
|
|
if not enum_cls or code is None:
|
|
return category or "OTHER"
|
|
try:
|
|
return enum_cls(int(code)).name
|
|
except ValueError:
|
|
return f"{category}_{code}"
|
|
|
|
|
|
def classify_failure(
|
|
error_type: Optional[str],
|
|
error_message: Optional[str] = None,
|
|
download_errors: Any = None,
|
|
) -> Tuple[str, str]:
|
|
category, code = _parse_error_type(error_type)
|
|
subtype = _failure_subtype_name(category, code)
|
|
if category == "DOWNLOAD_ERROR":
|
|
if code in DOWNLOAD_NETWORK_ERROR_CODES:
|
|
return "download_network", subtype
|
|
if code == int(DownloadError.ALL_SOURCES_FAILED) and _all_download_errors_are_network(download_errors):
|
|
return "download_network", "ALL_SOURCES_NETWORK_FAILED"
|
|
if _download_error_message_is_network(error_message):
|
|
return "download_network", subtype
|
|
return "download_other", subtype
|
|
if category == "INFRA_ERROR":
|
|
return "infra", subtype
|
|
if category == "APP_ERROR":
|
|
return "app", subtype
|
|
if category == "BUSINESS_ERROR":
|
|
return "business", subtype
|
|
return "other", subtype
|
|
|
|
|
|
class TaskDiagnostics:
|
|
def __init__(self):
|
|
self.values: Dict[str, Any] = {}
|
|
|
|
def mark(self, key: str, value: Any = True):
|
|
if value is None:
|
|
return
|
|
self.values[key] = value
|
|
|
|
def build_payload(
|
|
self,
|
|
outcome: str,
|
|
error_type: Optional[str] = None,
|
|
error_message: Optional[str] = None,
|
|
download_errors: Any = None,
|
|
) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {"outcome": outcome}
|
|
if self.values:
|
|
payload["diagnostics"] = dict(sorted(self.values.items()))
|
|
if outcome == "success":
|
|
return payload
|
|
failure_domain, failure_subtype = classify_failure(
|
|
error_type,
|
|
error_message=error_message,
|
|
download_errors=download_errors,
|
|
)
|
|
payload["failure_domain"] = failure_domain
|
|
payload["failure_subtype"] = failure_subtype
|
|
payload["failure_message"] = error_message or ""
|
|
return payload
|
|
|
|
|
|
class TaskMonitor:
|
|
def __init__(self, worker, task_key: str):
|
|
self.worker = worker
|
|
self.task_key = task_key
|
|
self.task_started_at: Optional[float] = None
|
|
self.stage_started_at: Dict[str, float] = {}
|
|
self.diagnostics = TaskDiagnostics()
|
|
self.final_trace_payload: Optional[Dict[str, Any]] = None
|
|
|
|
def emit_event(
|
|
self,
|
|
event_type: str,
|
|
*,
|
|
stage: Optional[str] = None,
|
|
state: Optional[str] = None,
|
|
status: Optional[str] = None,
|
|
reason: Optional[str] = None,
|
|
metrics: Optional[Dict[str, Any]] = None,
|
|
):
|
|
payload = {
|
|
"event_type": event_type,
|
|
"task_key": self.task_key,
|
|
}
|
|
if stage is not None:
|
|
payload["stage"] = stage
|
|
if state is not None:
|
|
payload["state"] = state
|
|
if status is not None:
|
|
payload["status"] = status
|
|
if reason is not None:
|
|
payload["reason"] = reason
|
|
if metrics:
|
|
payload["metrics"] = metrics
|
|
self.worker.event(payload)
|
|
|
|
def mark(self, key: str, value: Any = True):
|
|
self.diagnostics.mark(key, value)
|
|
|
|
def start_task(self, retry_count: int = 0):
|
|
self.task_started_at = time.time()
|
|
self.emit_event("task_started", metrics={"retry_count": retry_count})
|
|
|
|
def start_stage(self, stage_name: str, state_name: Optional[str] = None):
|
|
self.stage_started_at[stage_name] = time.time()
|
|
self.emit_event("stage_started", stage=stage_name)
|
|
if state_name:
|
|
self.emit_event("worker_state_changed", stage=stage_name, state=state_name)
|
|
|
|
def finish_stage(self, stage_name: str, status: str = "success", **metrics) -> Optional[float]:
|
|
started_at = self.stage_started_at.pop(stage_name, None)
|
|
duration_seconds = round(time.time() - started_at, 2) if started_at else None
|
|
stage_metrics = dict(metrics)
|
|
if duration_seconds is not None and "duration_seconds" not in stage_metrics:
|
|
stage_metrics["duration_seconds"] = duration_seconds
|
|
self.emit_event("stage_finished", stage=stage_name, metrics=stage_metrics)
|
|
return duration_seconds
|
|
|
|
def finish_task(
|
|
self,
|
|
task_status: str,
|
|
*,
|
|
failed_stage: Optional[str] = None,
|
|
error_type: Optional[str] = None,
|
|
error_message: Optional[str] = None,
|
|
retry_count: int = 0,
|
|
download_errors: Any = None,
|
|
) -> Dict[str, Any]:
|
|
total_duration_seconds = round(time.time() - (self.task_started_at or time.time()), 2)
|
|
self.final_trace_payload = self.diagnostics.build_payload(
|
|
task_status,
|
|
error_type=error_type,
|
|
error_message=error_message,
|
|
download_errors=download_errors,
|
|
)
|
|
metrics = {
|
|
"total_duration_seconds": total_duration_seconds,
|
|
"retry_count": retry_count,
|
|
"trace": self.final_trace_payload,
|
|
}
|
|
if task_status != "success":
|
|
metrics["failed_total_duration_seconds"] = total_duration_seconds
|
|
if failed_stage:
|
|
metrics["failed_stage"] = failed_stage
|
|
if error_type:
|
|
metrics["error_type"] = error_type
|
|
if error_message:
|
|
metrics["error_message"] = error_message
|
|
if download_errors:
|
|
metrics["download_errors"] = download_errors
|
|
if self.final_trace_payload.get("failure_domain"):
|
|
metrics["failure_domain"] = self.final_trace_payload["failure_domain"]
|
|
metrics["failure_subtype"] = self.final_trace_payload["failure_subtype"]
|
|
self.emit_event("task_finished", status=task_status, metrics=metrics)
|
|
return self.final_trace_payload
|
|
|
|
def build_report_payload(self, status: str, error: Optional[Any] = None, **extra) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {"status": status}
|
|
if error is not None:
|
|
payload["error"] = error.to_report_dict() if hasattr(error, "to_report_dict") else error
|
|
if self.final_trace_payload:
|
|
payload["trace"] = self.final_trace_payload
|
|
payload.update(extra)
|
|
return payload
|