Agent 跨集群调度：推理流量在多 K8s 集群间动态迁移

Agent 跨集群调度：推理流量在多 K8s 集群间动态迁移

一、集群 A 的 A100 被抢光，集群 B 的 A100 在闲着

我们的 AI 推理服务部署在 3 个 K8s 集群上（us-east、us-west、ap-southeast），每个集群有 8 张 A100。下午 3 点（美东的上午），us-east 的推理请求爆满，GPU 利用率 95%；同时 ap-southeast 是凌晨，GPU 利用率 5%。但推理流量没有跨集群调度——地域亲和性绑死了"us-east 用户只能用 us-east 集群"。

跨集群调度的需求很明确：把"位置不敏感"的推理请求（如离线批处理、长文本总结）动态迁移到空闲集群，释放本地集群的 GPU 给实时请求。实现上的挑战是：跨集群网络延迟（200ms+）、推理结果回传、以及 K8s 原生调度器不支持跨集群决策。

二、跨集群推理调度的架构

graph TD A["推理请求"] --> B["全局调度器<br/>Global Scheduler"] B --> C{"请求延迟<br/>容忍度?"} C -->|"< 50ms<br/>(实时)"| D["本地集群<br/>Cluster Local"] C -->|"< 500ms<br/>(交互式)"| E["区域集群<br/>Same Region"] C -->|"不限<br/>(批处理)"| F["全局集群<br/>Any Cluster"] D --> G["Cluster A<br/>(us-east)"] E --> H["Cluster B<br/>(us-west)"] F --> I["Cluster C<br/>(ap-southeast)"] B --> J["集群健康探测<br/>- GPU 利用率<br/>- 排队长度<br/>- 网络延迟"] J --> C K["Pod 生命周期管理"] --> G K --> H K --> I style B fill:#4A90D9,color:#fff style C fill:#F5A623,color:#000 style G fill:#50B86C,color:#fff style H fill:#50B86C,color:#fff style I fill:#50B86C,color:#fff

三层调度策略：

1. 本地优先（Locality-Aware）：实时推理请求（延迟 < 50ms）强制路由到本地集群。不跨集群，避免网络延迟导致超时。
2. 区域弹性（Regional Overflow）：交互式请求（延迟 < 500ms）优先本地，本地 GPU 不足时路由到同一区域的远程集群。
3. 全局分发（Global Batch）：离线批处理请求容许多秒级延迟，可在所有集群间负载均衡。

关键设计决策

• 全局调度器：在所有集群之上运行一个独立的调度器，它不是 K8s 原生调度器，而是根据集群健康状态和请求延迟容忍度做跨集群决策的独立组件。
• 推理 Pod 的跨集群生命周期：不是把单个推理 Pod 从一个集群迁移到另一个（跨集群 Pod migration 不存在），而是在目标集群创建新的推理 Pod，推理完成后销毁。
• 推理结果回流：远程推理的结果通过对象存储（S3/MinIO）或消息队列回传，不依赖跨集群网络长连接。

三、生产级跨集群调度实现

全局调度器

""" 跨集群推理全局调度器 根据集群健康状态和请求延迟容忍度分配推理任务 """ import asyncio import random import time from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, Tuple import aiohttp from kubernetes import client, config class LatencyClass(Enum): """请求延迟容忍度""" REALTIME = "realtime" # < 50ms INTERACTIVE = "interactive" # < 500ms BATCH = "batch" # 不限 @dataclass class ClusterHealth: """集群健康状态""" cluster_name: str region: str # us-east / us-west / ap-southeast gpu_total: int gpu_available: int gpu_utilization_pct: float queue_length: int avg_inference_latency_ms: float # 跨集群网络延迟（不同源集群的值不同） network_latency: Dict[str, float] = field(default_factory=dict) last_updated: float = field(default_factory=time.time) @property def gpu_utilization(self) -> float: return self.gpu_utilization_pct / 100 @property def is_overloaded(self) -> bool: """GPU 利用率 > 85% 或 排队 > 10""" return self.gpu_utilization > 0.85 or self.queue_length > 10 class GlobalScheduler: """ 全局跨集群调度器 调度决策优先级： 1. 实时请求 (REALTIME) → 本地集群 2. 交互式请求 (INTERACTIVE) → 同区域集群 3. 批处理请求 (BATCH) → 全局最少负载集群 """ def __init__(self): # 集群健康状态缓存 self._cluster_health: Dict[str, ClusterHealth] = {} self._health_lock = asyncio.Lock() # 每个集群的 K8s client self._k8s_clients: Dict[str, client.CoreV1Api] = {} # 推理结果存储（跨集群共享） self._result_store = None # S3 client async def start(self, cluster_configs: List[dict]): """初始化所有集群的 K8s 客户端并启动健康探测""" for cfg in cluster_configs: k8s_client = self._init_k8s_client(cfg) self._k8s_clients[cfg["name"]] = k8s_client # 启动后台任务：定期探测集群健康 asyncio.create_task(self._health_probe_loop()) async def schedule_inference( self, request_id: str, model_name: str, input_data: dict, latency_class: LatencyClass, source_cluster: str, priority: int = 0, ) -> Tuple[str, str]: """ 调度推理请求到最优集群 返回：(目标集群名, 推理 Pod 名) """ target_cluster = await self._select_cluster( latency_class, source_cluster ) if target_cluster is None: raise RuntimeError("没有可用集群") # 在目标集群创建推理 Pod pod_name = await self._create_inference_pod( target_cluster, request_id, model_name, input_data ) # 记录调度决策（用于后续分析和优化） await self._log_scheduling_decision( request_id, source_cluster, target_cluster, latency_class ) return target_cluster, pod_name async def _select_cluster( self, latency_class: LatencyClass, source_cluster: str ) -> Optional[str]: """ 根据延迟容忍度选择目标集群 为什么需要 source_cluster： 跨集群延迟不是对称的！ us-east → ap-southeast 可能是 200ms， 但 us-east → us-west 可能只有 60ms。 需要基于实际的网络拓扑做延迟感知路由 """ health_map = await self._get_cluster_health() if not health_map: return None candidates = [] for cluster_name, health in health_map.items(): # 过滤 1: GPU 是否可用 if health.gpu_available <= 0: continue # 过滤 2: 延迟约束 if latency_class == LatencyClass.REALTIME: # 实时：仅本地集群 if cluster_name != source_cluster: continue elif latency_class == LatencyClass.INTERACTIVE: # 交互式：本地 + 同区域 source_region = health_map[source_cluster].region if health.region != source_region: # 非同区域但延迟可接受也可考虑 latency = health.network_latency.get(source_cluster, 999) if latency > 500: continue # 计算选择分数（分数越低越优先） score = self._calculate_cluster_score(health, latency_class) candidates.append((score, cluster_name)) if not candidates: return None # 选分数最低的集群 candidates.sort(key=lambda x: x[0]) return candidates[0][1] def _calculate_cluster_score( self, health: ClusterHealth, latency_class: LatencyClass ) -> float: """ 计算集群选择分数 分数构成： - GPU 利用率权重: 40% - 队列长度权重: 30% - 推理延迟权重: 20% - 随机因子: 10%（微扰动防抖） """ # 基础分数 score = ( health.gpu_utilization * 40 + min(health.queue_length / 20, 1.0) * 30 + min(health.avg_inference_latency_ms / 1000, 1.0) * 20 ) # 随机扰动（防止所有请求同时路由到同一个刚恢复的集群） # 为什么需要扰动： # 如果一个集群刚从过载恢复（score=0），所有请求突然一起路由过去， # 瞬间又过载。扰动让请求分几次路由过去 jitter = random.uniform(0, 0.1) * 100 score += jitter return score async def _create_inference_pod( self, cluster_name: str, request_id: str, model_name: str, input_data: dict, ) -> str: """在目标集群创建推理 Pod""" k8s = self._k8s_clients[cluster_name] pod_name = f"inference-{request_id[:8]}-{int(time.time())}" pod_manifest = { "apiVersion": "v1", "kind": "Pod", "metadata": { "name": pod_name, "labels": { "app": "cross-cluster-inference", "request-id": request_id, "model": model_name, }, }, "spec": { "containers": [{ "name": "inference", "image": f"my-registry/inference-{model_name}:latest", "env": [ {"name": "INPUT_DATA", "value": str(input_data)}, {"name": "RESULT_BUCKET", "value": "s3://inference-results"}, {"name": "RESULT_KEY", "value": f"{request_id}/result.json"}, ], "resources": { "limits": {"nvidia.com/gpu": 1}, }, }], "restartPolicy": "Never", # TTL: 推理完成后 300s 自动清理 # 为什么用 TTL Controller 而非手动删除： # 如果调度器崩溃，Pod 会残留浪费资源 "ttlSecondsAfterFinished": 300, }, } # 提交 Pod 创建请求 await asyncio.to_thread( k8s.create_namespaced_pod, namespace="inference", body=pod_manifest, ) return pod_name async def _health_probe_loop(self): """定期探测所有集群的健康状态""" while True: for cluster_name, k8s_client in self._k8s_clients.items(): try: health = await self._probe_cluster(cluster_name, k8s_client) async with self._health_lock: self._cluster_health[cluster_name] = health except Exception as e: print(f"集群 {cluster_name} 健康探测失败: {e}") await asyncio.sleep(15) # 每 15 秒探测一次 async def _probe_cluster( self, cluster_name: str, k8s_client ) -> ClusterHealth: """探测单个集群的健康状态""" # 查询 GPU 节点信息 # 查询推理队列长度（从 Redis/MQ） # 查询平均推理延迟（从 Prometheus） # 简化实现——实际需要调用监控 API return ClusterHealth( cluster_name=cluster_name, region=self._get_region(cluster_name), gpu_total=8, gpu_available=4, gpu_utilization_pct=45.0, queue_length=3, avg_inference_latency_ms=250, network_latency={}, # 从网络探测服务获取 ) async def _get_cluster_health(self) -> Dict[str, ClusterHealth]: """获取所有集群的健康状态（线程安全）""" async with self._health_lock: return dict(self._cluster_health) def _get_region(self, cluster_name: str) -> str: """从集群名推断区域""" regions = { "us-east-prod": "us-east", "us-west-prod": "us-west", "ap-southeast-prod": "ap-southeast", } return regions.get(cluster_name, "unknown") def _init_k8s_client(self, cfg: dict): """初始化 K8s 客户端""" pass async def _log_scheduling_decision(self, *args): """记录调度决策用于分析""" pass

跨集群推理的 Pod 管理

""" 跨集群推理 Pod 管理器 负责 Pod 创建、监控和清理 """ class CrossClusterPodManager: """跨集群推理 Pod 管理器""" async def wait_for_result( self, cluster_name: str, pod_name: str, timeout_sec: float = 300 ) -> Dict: """ 等待推理结果 为什么不用 gRPC/WebSocket 直接连接 Pod： 跨集群网络安全策略复杂，且长连接在跨区域网络不稳定。 使用 S3 作为中间存储，Pod 写结果到 S3，调度器轮询 S3 """ k8s = self._k8s_clients[cluster_name] deadline = time.time() + timeout_sec while time.time() < deadline: try: # 检查 Pod 状态 pod = await asyncio.to_thread( k8s.read_namespaced_pod, name=pod_name, namespace="inference", ) phase = pod.status.phase if phase == "Succeeded": # Pod 正常完成，从 S3 读取结果 result = await self._fetch_result_from_s3(pod_name) # 清理 Pod await self._delete_pod(cluster_name, pod_name) return result elif phase == "Failed": # Pod 失败：获取错误信息 error = await self._get_pod_error(cluster_name, pod_name) await self._delete_pod(cluster_name, pod_name) raise RuntimeError(f"推理失败: {error}") elif phase in ("Unknown", "CrashLoopBackOff"): # 异常状态 await self._delete_pod(cluster_name, pod_name) raise RuntimeError(f"Pod 异常状态: {phase}") except Exception as e: raise RuntimeError(f"监控 Pod {pod_name} 失败: {e}") await asyncio.sleep(2) # 超时：强制清理 await self._delete_pod(cluster_name, pod_name) raise TimeoutError(f"推理超时 ({timeout_sec}s)") async def _fetch_result_from_s3(self, pod_name: str) -> Dict: """从 S3/MinIO 读取推理结果""" pass async def _get_pod_error(self, cluster_name: str, pod_name: str) -> str: """获取 Pod 错误信息""" k8s = self._k8s_clients[cluster_name] try: logs = await asyncio.to_thread( k8s.read_namespaced_pod_log, name=pod_name, namespace="inference", tail_lines=50, ) return logs except Exception: return "无法获取错误日志" async def _delete_pod(self, cluster_name: str, pod_name: str): """删除推理 Pod""" k8s = self._k8s_clients.get(cluster_name) if k8s: try: await asyncio.to_thread( k8s.delete_namespaced_pod, name=pod_name, namespace="inference", ) except Exception: pass

四、跨集群调度的边界

缺点：

1. 跨区域网络延迟不可消除：us-east→ap-southeast 的 RTT 约 200ms，如果交互式请求的延迟容忍度是 500ms，留给你推理的时间只有 300ms。
2. 双写存储成本：推理结果需要通过 S3 回传，每个结果额外占用存储和网络带宽。对于 50MB 的音频推理结果，S3 上传+下载可能增加 5-10s。
3. 不一致的多集群状态：全局调度器维护各集群的"实时"健康状态，但由于网络延迟，状态可能滞后 10-15 秒。滞后期间可能做出次优调度决策。

禁用场景：

• 延迟 < 50ms 的实时推理：跨集群通信的延迟本身已超过容忍度。
• 数据本地性强的推理（如大型向量数据库）：跨集群推理需要跨集群数据传输，成本过高。
• 安全合规要求数据不出区域的场景。

五、总结

Agent 跨集群调度的核心，是引入一个全局调度器，根据请求延迟容忍度将推理任务分配到不同集群。三层路由策略（实时→本地、交互→区域、批处理→全局）平衡了延迟和资源利用率。工程实现要点：通过集群健康探测维护全局状态、用 S3 做跨集群结果传递、TTL 自动清理推理 Pod 防止资源泄漏。跨集群的物理延迟是硬约束，调度器只能优化，不能消除。