410 lines
15 KiB
Python
410 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from app.config import settings
|
|
from app.enums.celery_runtime import CeleryRuntimeEvent
|
|
from app.services.operation_log_service import log_operation_event
|
|
from app.services.redis_registry_service import (
|
|
RedisExecutionLockUnavailable,
|
|
datetime_to_epoch,
|
|
get_registry_redis,
|
|
)
|
|
|
|
|
|
def _worker_instance_key(worker_instance_id: str) -> str:
|
|
return f"{settings.CELERY_RUNTIME_WORKER_INSTANCE_PREFIX}:{worker_instance_id}"
|
|
|
|
|
|
def _worker_name_instances_key(worker_name: str) -> str:
|
|
return f"{settings.CELERY_RUNTIME_WORKER_NAME_INSTANCE_ZSET_PREFIX}:{worker_name}"
|
|
|
|
|
|
def _worker_task_set_key(worker_instance_id: str) -> str:
|
|
return f"{settings.CELERY_RUNTIME_WORKER_TASK_SET_PREFIX}:{worker_instance_id}"
|
|
|
|
|
|
async def startup_barrier_exists() -> bool:
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable while checking startup barrier")
|
|
return bool(await redis.exists(settings.CELERY_RUNTIME_STARTUP_BARRIER_KEY))
|
|
|
|
|
|
async def set_startup_barrier() -> None:
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable while setting startup barrier")
|
|
await redis.set(
|
|
settings.CELERY_RUNTIME_STARTUP_BARRIER_KEY,
|
|
str(datetime_to_epoch(datetime.now(timezone.utc))),
|
|
ex=max(30, int(settings.CELERY_RUNTIME_STARTUP_BARRIER_TTL_SECONDS or 120)),
|
|
)
|
|
|
|
|
|
async def clear_startup_barrier() -> None:
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
return
|
|
await redis.delete(settings.CELERY_RUNTIME_STARTUP_BARRIER_KEY)
|
|
|
|
|
|
async def guard_periodic_recovery(*, check_global_lock: bool = True) -> dict[str, Any] | None:
|
|
if await startup_barrier_exists():
|
|
log_operation_event(
|
|
domain="celery_runtime",
|
|
event_type=CeleryRuntimeEvent.RECOVERY_BARRIER_SKIPPED.value,
|
|
event_status="skipped",
|
|
source="recovery",
|
|
detail={"barrier_key": settings.CELERY_RUNTIME_STARTUP_BARRIER_KEY},
|
|
)
|
|
return {"skipped": "startup_barrier"}
|
|
|
|
if check_global_lock:
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable while checking global recovery lock")
|
|
if await redis.exists(settings.CELERY_RUNTIME_GLOBAL_RECOVERY_LOCK_KEY):
|
|
log_operation_event(
|
|
domain="celery_runtime",
|
|
event_type=CeleryRuntimeEvent.RECOVERY_BARRIER_SKIPPED.value,
|
|
event_status="skipped",
|
|
source="recovery",
|
|
detail={"global_lock_key": settings.CELERY_RUNTIME_GLOBAL_RECOVERY_LOCK_KEY},
|
|
)
|
|
return {"skipped": "global_recovery_lock"}
|
|
return None
|
|
|
|
|
|
async def garbage_collect_registry_pair(*, hash_key: str, zset_key: str, limit: int = 500) -> dict[str, int]:
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable during registry GC")
|
|
|
|
members = await redis.zrange(zset_key, 0, max(0, int(limit) - 1))
|
|
if not members:
|
|
return {"checked": 0, "removed": 0}
|
|
|
|
values = await redis.hmget(hash_key, members)
|
|
stale = [str(member) for member, value in zip(members, values) if not value]
|
|
if stale:
|
|
await redis.zrem(zset_key, *stale)
|
|
return {"checked": len(members), "removed": len(stale)}
|
|
|
|
|
|
async def mark_stale_worker_instance_candidates(
|
|
*,
|
|
worker_name: str,
|
|
current_worker_instance_id: str,
|
|
supports_targeted_recovery: bool,
|
|
emit_duplicate_log: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""只扫描同一逻辑 Worker 名称下已经失活的旧主实例。
|
|
|
|
同名存在其他活跃实例时保守跳过，避免滚动发布或错误重名导致互相抢占。
|
|
旧实例中的任务只有在业务 String 锁已经不存在时才提前标记为恢复候选；
|
|
真正接管仍由各业务恢复服务执行 DB lease/claim/attempt CAS。
|
|
"""
|
|
if not supports_targeted_recovery:
|
|
return {
|
|
"skipped": "identity_fallback",
|
|
"checked_instances": 0,
|
|
"stale_instances": 0,
|
|
"marked": 0,
|
|
}
|
|
|
|
normalized_name = str(worker_name or "").strip()
|
|
normalized_current = str(current_worker_instance_id or "").strip()
|
|
if not normalized_name or not normalized_current:
|
|
return {
|
|
"skipped": "invalid_identity",
|
|
"checked_instances": 0,
|
|
"stale_instances": 0,
|
|
"marked": 0,
|
|
}
|
|
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable while discovering stale worker instances")
|
|
|
|
index_key = _worker_name_instances_key(normalized_name)
|
|
raw_instances = await redis.zrange(index_key, 0, -1, withscores=True)
|
|
instances = [
|
|
(str(member), int(float(score)))
|
|
for member, score in raw_instances
|
|
if str(member) != normalized_current
|
|
]
|
|
if not instances:
|
|
return {
|
|
"checked_instances": 0,
|
|
"active_instances": 0,
|
|
"stale_instances": 0,
|
|
"marked": 0,
|
|
"live": 0,
|
|
"invalid": 0,
|
|
}
|
|
|
|
keys = [_worker_instance_key(instance_id) for instance_id, _ in instances]
|
|
values = await redis.mget(keys)
|
|
active_instances = [
|
|
instance_id
|
|
for (instance_id, _), value in zip(instances, values)
|
|
if value
|
|
]
|
|
if active_instances:
|
|
if emit_duplicate_log:
|
|
log_operation_event(
|
|
domain="celery_runtime",
|
|
event_type=CeleryRuntimeEvent.WORKER_DUPLICATE_NAME.value,
|
|
event_status="warning",
|
|
source="worker_ready",
|
|
detail={
|
|
"worker_name": normalized_name,
|
|
"current_worker_instance_id": normalized_current,
|
|
"other_active_worker_instance_ids": active_instances,
|
|
},
|
|
)
|
|
return {
|
|
"skipped": "duplicate_active_worker_name",
|
|
"checked_instances": len(instances),
|
|
"active_instances": len(active_instances),
|
|
"stale_instances": 0,
|
|
"marked": 0,
|
|
}
|
|
|
|
# 实例 key 由主进程 heartbeat 独立维护。key 已不存在即可判定旧主实例
|
|
# 不再活跃；任务是否可以恢复仍必须继续检查业务 String 锁和 DB fencing。
|
|
stale_instances = [
|
|
instance_id
|
|
for (instance_id, _heartbeat_at), value in zip(instances, values)
|
|
if not value
|
|
]
|
|
|
|
totals: dict[str, Any] = {
|
|
"checked_instances": len(instances),
|
|
"active_instances": 0,
|
|
"stale_instances": len(stale_instances),
|
|
"checked": 0,
|
|
"marked": 0,
|
|
"live": 0,
|
|
"invalid": 0,
|
|
}
|
|
|
|
for stale_instance_id in stale_instances:
|
|
log_operation_event(
|
|
domain="celery_runtime",
|
|
event_type=CeleryRuntimeEvent.WORKER_STALE_INSTANCE_FOUND.value,
|
|
event_status="success",
|
|
source="worker_ready",
|
|
detail={
|
|
"worker_name": normalized_name,
|
|
"current_worker_instance_id": normalized_current,
|
|
"old_worker_instance_id": stale_instance_id,
|
|
},
|
|
)
|
|
result = await mark_worker_instance_runtime_candidates(
|
|
old_worker_instance_id=stale_instance_id,
|
|
worker_name=normalized_name,
|
|
)
|
|
for key in ("checked", "marked", "live", "invalid"):
|
|
totals[key] += int(result.get(key, 0) or 0)
|
|
|
|
return totals
|
|
|
|
|
|
async def mark_worker_instance_runtime_candidates(
|
|
*,
|
|
old_worker_instance_id: str,
|
|
worker_name: str | None = None,
|
|
) -> dict[str, int]:
|
|
"""按旧 Worker 主实例 ID 精准标记失锁任务。"""
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable while marking worker recovery candidates")
|
|
|
|
normalized_instance = str(old_worker_instance_id or "").strip()
|
|
if not normalized_instance:
|
|
return {"checked": 0, "marked": 0, "live": 0, "invalid": 0}
|
|
|
|
task_set_key = _worker_task_set_key(normalized_instance)
|
|
runtime_ids = [str(value) for value in await redis.smembers(task_set_key)]
|
|
if not runtime_ids:
|
|
return {"checked": 0, "marked": 0, "live": 0, "invalid": 0}
|
|
|
|
log_operation_event(
|
|
domain="celery_runtime",
|
|
event_type=CeleryRuntimeEvent.WORKER_INSTANCE_RECOVERY_START.value,
|
|
event_status="started",
|
|
source="worker_ready",
|
|
detail={
|
|
"worker_name": worker_name,
|
|
"old_worker_instance_id": normalized_instance,
|
|
"candidate_count": len(runtime_ids),
|
|
},
|
|
)
|
|
|
|
raw_locations = await redis.hmget(settings.CELERY_RUNTIME_LOCATION_HASH_KEY, runtime_ids)
|
|
locations: dict[str, dict[str, Any]] = {}
|
|
lock_keys: list[str] = []
|
|
invalid_ids: list[str] = []
|
|
|
|
for runtime_id, raw in zip(runtime_ids, raw_locations):
|
|
try:
|
|
payload = json.loads(raw) if raw else None
|
|
except Exception:
|
|
payload = None
|
|
|
|
if (
|
|
not isinstance(payload, dict)
|
|
or not payload.get("lock_key")
|
|
or not payload.get("hash_key")
|
|
or not payload.get("zset_key")
|
|
or str(payload.get("worker_instance_id") or "") != normalized_instance
|
|
):
|
|
invalid_ids.append(runtime_id)
|
|
continue
|
|
|
|
locations[runtime_id] = payload
|
|
lock_keys.append(str(payload["lock_key"]))
|
|
|
|
lock_values = await redis.mget(lock_keys) if lock_keys else []
|
|
live_by_key = {key: bool(value) for key, value in zip(lock_keys, lock_values)}
|
|
|
|
now_epoch = datetime_to_epoch(datetime.now(timezone.utc))
|
|
marked = 0
|
|
live = 0
|
|
stale_by_hash: dict[str, list[str]] = {}
|
|
|
|
for runtime_id, location in locations.items():
|
|
lock_key = str(location["lock_key"])
|
|
if live_by_key.get(lock_key):
|
|
live += 1
|
|
continue
|
|
stale_by_hash.setdefault(str(location["hash_key"]), []).append(runtime_id)
|
|
|
|
payload_by_runtime: dict[str, dict[str, Any]] = {}
|
|
for hash_key, hash_runtime_ids in stale_by_hash.items():
|
|
raw_payloads = await redis.hmget(hash_key, hash_runtime_ids)
|
|
for runtime_id, raw_payload in zip(hash_runtime_ids, raw_payloads):
|
|
try:
|
|
payload = json.loads(raw_payload) if raw_payload else {}
|
|
except Exception:
|
|
payload = {}
|
|
payload_by_runtime[runtime_id] = payload if isinstance(payload, dict) else {}
|
|
|
|
pipe = redis.pipeline(transaction=False)
|
|
for runtime_id, location in locations.items():
|
|
lock_key = str(location["lock_key"])
|
|
if live_by_key.get(lock_key):
|
|
continue
|
|
|
|
hash_key = str(location["hash_key"])
|
|
zset_key = str(location["zset_key"])
|
|
payload = payload_by_runtime.get(runtime_id, {})
|
|
payload.update(
|
|
{
|
|
"runtime_state": "recovery_candidate",
|
|
"check_at": now_epoch,
|
|
"reason": "worker_instance_inactive_lock_missing",
|
|
"old_worker_instance_id": normalized_instance,
|
|
}
|
|
)
|
|
pipe.hset(hash_key, runtime_id, json.dumps(payload, ensure_ascii=False, default=str))
|
|
pipe.zadd(zset_key, {runtime_id: now_epoch})
|
|
marked += 1
|
|
|
|
if invalid_ids:
|
|
pipe.srem(task_set_key, *invalid_ids)
|
|
|
|
await pipe.execute()
|
|
|
|
result = {
|
|
"checked": len(runtime_ids),
|
|
"marked": marked,
|
|
"live": live,
|
|
"invalid": len(invalid_ids),
|
|
}
|
|
log_operation_event(
|
|
domain="celery_runtime",
|
|
event_type=CeleryRuntimeEvent.WORKER_INSTANCE_RECOVERY_DONE.value,
|
|
event_status="success",
|
|
source="worker_ready",
|
|
detail={
|
|
"worker_name": worker_name,
|
|
"old_worker_instance_id": normalized_instance,
|
|
**result,
|
|
},
|
|
)
|
|
return result
|
|
|
|
|
|
async def garbage_collect_worker_registry(*, limit: int = 500) -> dict[str, int]:
|
|
"""清理 V2 Worker 名称索引和旧实例 Task Set 中的失配成员。"""
|
|
redis = await get_registry_redis()
|
|
if redis is None:
|
|
raise RedisExecutionLockUnavailable("Redis unavailable during worker registry GC")
|
|
|
|
checked_names = 0
|
|
checked_instances = 0
|
|
removed_instances = 0
|
|
removed_task_members = 0
|
|
now = int(time.time())
|
|
grace = max(
|
|
int(settings.CELERY_RUNTIME_WORKER_HEARTBEAT_TTL_SECONDS or 90),
|
|
int(settings.CELERY_RUNTIME_WORKER_STALE_GRACE_SECONDS or 120),
|
|
)
|
|
pattern = f"{settings.CELERY_RUNTIME_WORKER_NAME_INSTANCE_ZSET_PREFIX}:*"
|
|
|
|
async for raw_index_key in redis.scan_iter(match=pattern, count=max(10, min(limit, 500))):
|
|
if checked_names >= limit:
|
|
break
|
|
checked_names += 1
|
|
index_key = str(raw_index_key)
|
|
raw_instances = await redis.zrange(index_key, 0, -1, withscores=True)
|
|
|
|
for raw_instance_id, raw_score in raw_instances:
|
|
if checked_instances >= limit:
|
|
break
|
|
checked_instances += 1
|
|
instance_id = str(raw_instance_id)
|
|
score = int(float(raw_score))
|
|
if now - score < grace:
|
|
continue
|
|
if await redis.exists(_worker_instance_key(instance_id)):
|
|
continue
|
|
|
|
task_set_key = _worker_task_set_key(instance_id)
|
|
runtime_ids = [str(value) for value in await redis.smembers(task_set_key)]
|
|
if runtime_ids:
|
|
raw_locations = await redis.hmget(settings.CELERY_RUNTIME_LOCATION_HASH_KEY, runtime_ids)
|
|
stale_members: list[str] = []
|
|
for runtime_id, raw_location in zip(runtime_ids, raw_locations):
|
|
try:
|
|
location = json.loads(raw_location) if raw_location else None
|
|
except Exception:
|
|
location = None
|
|
if (
|
|
not isinstance(location, dict)
|
|
or str(location.get("worker_instance_id") or "") != instance_id
|
|
):
|
|
stale_members.append(runtime_id)
|
|
if stale_members:
|
|
removed_task_members += int(await redis.srem(task_set_key, *stale_members) or 0)
|
|
|
|
if int(await redis.scard(task_set_key) or 0) == 0:
|
|
await redis.delete(task_set_key)
|
|
removed_instances += int(await redis.zrem(index_key, instance_id) or 0)
|
|
|
|
if int(await redis.zcard(index_key) or 0) == 0:
|
|
await redis.delete(index_key)
|
|
|
|
return {
|
|
"checked_names": checked_names,
|
|
"checked_instances": checked_instances,
|
|
"removed_instances": removed_instances,
|
|
"removed_task_members": removed_task_members,
|
|
}
|