FastheiandClaude Opus 4.8
2b0f2fe6f2
实测 heicode-rd（OSSCluster，2 分片）：默认 keys() 只命中单节点， 漏掉其它分片上的 key → agent_registry/task_queue 枚举不全。 cluster 模式下改用 target_nodes=PRIMARIES，redis-py 合并各节点结果。 附带修 agent_swarm#44 Bug1：Dockerfile.orchestrator 漏 COPY benchmark/ （orchestrator/quality.py 启动即 import benchmark.fixtures/metrics）→ 原镜像 CrashLoopBackOff。新增 .dockerignore 控制构建上下文。 影响范围：仅 agent_swarm orchestrator（连接层 + 构建物料）； 不改契约/计费/审计/密钥落地。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
214 lines
8.7 KiB
Python
214 lines
8.7 KiB
Python
"""Redis client for orchestrator state management.
|
|
|
|
Redis is the canonical store. An in-memory fallback (backed by the in-process
|
|
``fakeredis`` emulator, which faithfully implements the Redis list/hash API) is
|
|
available ONLY for local development and CI, and ONLY when explicitly enabled via
|
|
``REDIS_FAKE=1`` or ``ALLOW_MEMORY_STORE=1``. In production neither flag is set, so a
|
|
Redis outage fails fast on startup instead of silently dropping durability/Manager
|
|
state.
|
|
|
|
Connection config (precedence):
|
|
1. ``REDIS_URL`` — full URL, e.g. ``rediss://:<password>@host:10000/0`` for a
|
|
managed TLS Redis (Azure Cache / Redis Enterprise). ``rediss://`` enables TLS;
|
|
credentials live in the URL. Takes precedence over the discrete vars below.
|
|
2. Discrete vars: ``REDIS_HOST`` / ``REDIS_PORT`` / ``REDIS_DB`` plus optional
|
|
``REDIS_PASSWORD`` and ``REDIS_SSL`` (truthy → TLS).
|
|
|
|
All credentials come from the environment (injected via Secret / secret_ref) and are
|
|
NEVER hardcoded or logged. When neither a password nor TLS is set, behaviour is
|
|
identical to the previous plaintext in-cluster default (``redis-service:6379``).
|
|
"""
|
|
import os
|
|
from typing import Optional
|
|
import logging
|
|
|
|
try: # redis-py is required in production; guarded so dev/CI can run on fakeredis only.
|
|
import redis.asyncio as redis
|
|
except Exception: # pragma: no cover - exercised only when redis-py is absent
|
|
redis = None
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _truthy(value: Optional[str]) -> bool:
|
|
return (value or "").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
class RedisClient:
|
|
"""Async Redis client wrapper for orchestrator operations."""
|
|
|
|
def __init__(self):
|
|
self.client = None
|
|
self.url = (os.getenv("REDIS_URL") or "").strip() or None
|
|
self.host = os.getenv("REDIS_HOST", "redis-service")
|
|
self.port = int(os.getenv("REDIS_PORT", "6379"))
|
|
self.db = int(os.getenv("REDIS_DB", "0"))
|
|
self.password = os.getenv("REDIS_PASSWORD") or None
|
|
self.ssl = _truthy(os.getenv("REDIS_SSL"))
|
|
# Cluster mode is required for endpoints speaking the Redis Cluster protocol
|
|
# (e.g. Azure Redis Enterprise with clusteringPolicy=OSSCluster). A plain
|
|
# client there silently mis-routes keys() / cross-slot ops on multi-shard DBs.
|
|
self.cluster = _truthy(os.getenv("REDIS_CLUSTER"))
|
|
|
|
def _fallback_allowed(self) -> bool:
|
|
return _truthy(os.getenv("REDIS_FAKE")) or _truthy(os.getenv("ALLOW_MEMORY_STORE"))
|
|
|
|
def _make_real_client(self):
|
|
"""Build a real redis-py client from config (URL preferred, else discrete vars).
|
|
|
|
Connection kwargs only — does NOT connect. Credential-bearing values
|
|
(URL/password) are never logged here or by callers. When ``REDIS_CLUSTER``
|
|
is set, a cluster-aware client is built (cluster mode has no DB select, so
|
|
``REDIS_DB`` is ignored — Redis Cluster only exposes logical DB 0).
|
|
"""
|
|
if redis is None:
|
|
raise RuntimeError("redis-py is not installed")
|
|
common = dict(
|
|
decode_responses=True,
|
|
socket_connect_timeout=5,
|
|
)
|
|
if self.cluster:
|
|
from redis.asyncio.cluster import RedisCluster # lazy: only when cluster mode on
|
|
if self.url:
|
|
return RedisCluster.from_url(self.url, **common)
|
|
kwargs = dict(host=self.host, port=self.port, **common)
|
|
if self.password:
|
|
kwargs["password"] = self.password
|
|
if self.ssl:
|
|
kwargs["ssl"] = True
|
|
return RedisCluster(**kwargs)
|
|
|
|
common["socket_keepalive"] = True
|
|
if self.url:
|
|
# from_url honours the scheme: rediss:// → TLS, and any user:pass in the URL.
|
|
return redis.Redis.from_url(self.url, **common)
|
|
kwargs = dict(host=self.host, port=self.port, db=self.db, **common)
|
|
if self.password:
|
|
kwargs["password"] = self.password
|
|
if self.ssl:
|
|
kwargs["ssl"] = True
|
|
return redis.Redis(**kwargs)
|
|
|
|
def _target_desc(self) -> str:
|
|
"""Human-readable target for logs — no credentials. Never logs the URL itself."""
|
|
mode = "cluster" if self.cluster else "standalone"
|
|
if self.url:
|
|
scheme = self.url.split("://", 1)[0] if "://" in self.url else "redis"
|
|
return f"REDIS_URL ({scheme}://, TLS={scheme == 'rediss'}, {mode})"
|
|
return f"{self.host}:{self.port} (TLS={self.ssl}, {mode})"
|
|
|
|
def _make_fake_client(self):
|
|
"""Return an in-process fakeredis client (dev/CI fallback only)."""
|
|
import fakeredis.aioredis as fakeredis # imported lazily; dev/CI dependency
|
|
return fakeredis.FakeRedis(decode_responses=True)
|
|
|
|
async def connect(self):
|
|
"""Establish Redis connection, or a gated in-memory fallback for dev/CI."""
|
|
# Explicit fake mode (used by local runs and tests) short-circuits real Redis.
|
|
if _truthy(os.getenv("REDIS_FAKE")):
|
|
self.client = self._make_fake_client()
|
|
logger.warning("REDIS_FAKE enabled; using in-memory fakeredis (NOT for production)")
|
|
return
|
|
|
|
try:
|
|
self.client = self._make_real_client()
|
|
await self.client.ping()
|
|
logger.info(f"Connected to Redis at {self._target_desc()}")
|
|
except Exception as e:
|
|
if self._fallback_allowed():
|
|
logger.warning(
|
|
f"Redis unavailable ({e}); ALLOW_MEMORY_STORE set, using in-memory fakeredis "
|
|
"fallback (NOT for production)"
|
|
)
|
|
self.client = self._make_fake_client()
|
|
return
|
|
logger.error(f"Failed to connect to Redis: {e}")
|
|
raise
|
|
|
|
async def disconnect(self):
|
|
"""Close Redis connection."""
|
|
if self.client:
|
|
await self.client.close()
|
|
logger.info("Disconnected from Redis")
|
|
|
|
async def set(self, key: str, value: str, ex: Optional[int] = None):
|
|
"""Set key-value pair with optional expiration."""
|
|
await self.client.set(key, value, ex=ex)
|
|
|
|
async def incr(self, key: str) -> int:
|
|
"""Atomically increment an integer counter and return the new value.
|
|
|
|
Used for the per-swarm event sequence (strictly increasing from 1); INCR is
|
|
atomic so concurrent emits on the same swarm never collide on a number.
|
|
"""
|
|
return await self.client.incr(key)
|
|
|
|
async def get(self, key: str) -> Optional[str]:
|
|
"""Get value by key."""
|
|
return await self.client.get(key)
|
|
|
|
async def delete(self, key: str):
|
|
"""Delete key."""
|
|
await self.client.delete(key)
|
|
|
|
async def exists(self, key: str) -> bool:
|
|
"""Check if key exists."""
|
|
return await self.client.exists(key) > 0
|
|
|
|
async def hset(self, name: str, key: str, value: str):
|
|
"""Set hash field."""
|
|
await self.client.hset(name, key, value)
|
|
|
|
async def hget(self, name: str, key: str) -> Optional[str]:
|
|
"""Get hash field."""
|
|
return await self.client.hget(name, key)
|
|
|
|
async def hgetall(self, name: str) -> dict:
|
|
"""Get all hash fields."""
|
|
return await self.client.hgetall(name)
|
|
|
|
async def hdel(self, name: str, *keys: str):
|
|
"""Delete hash fields."""
|
|
await self.client.hdel(name, *keys)
|
|
|
|
async def keys(self, pattern: str) -> list:
|
|
"""Get keys matching pattern.
|
|
|
|
In cluster mode KEYS must fan out to every primary and merge — the default
|
|
routing hits a single node, so on a multi-shard cluster it silently drops
|
|
keys living on other shards (breaks agent/task enumeration). redis-py merges
|
|
the per-node results into one flat list.
|
|
"""
|
|
if self.cluster:
|
|
from redis.asyncio.cluster import RedisCluster
|
|
return await self.client.keys(pattern, target_nodes=RedisCluster.PRIMARIES)
|
|
return await self.client.keys(pattern)
|
|
|
|
async def lpush(self, key: str, *values: str):
|
|
"""Push values to list head."""
|
|
await self.client.lpush(key, *values)
|
|
|
|
async def rpush(self, key: str, *values: str):
|
|
"""Push values to list tail."""
|
|
await self.client.rpush(key, *values)
|
|
|
|
async def rpop(self, key: str) -> Optional[str]:
|
|
"""Pop value from list tail."""
|
|
return await self.client.rpop(key)
|
|
|
|
async def llen(self, key: str) -> int:
|
|
"""Get list length."""
|
|
return await self.client.llen(key)
|
|
|
|
async def lrange(self, key: str, start: int, end: int) -> list:
|
|
"""Get list range."""
|
|
return await self.client.lrange(key, start, end)
|
|
|
|
async def lrem(self, key: str, count: int, value: str) -> int:
|
|
"""Remove values from list."""
|
|
return await self.client.lrem(key, count, value)
|
|
|
|
|
|
# Global Redis client instance
|
|
redis_client = RedisClient()
|