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 --| 50msbr/(实时)| D[本地集群br/Cluster Local] C --| 500msbr/(交互式)| E[区域集群br/Same Region] C --|不限br/(批处理)| F[全局集群br/Any Cluster] D -- G[Cluster Abr/(us-east)] E -- H[Cluster Bbr/(us-west)] F -- I[Cluster Cbr/(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三层调度策略本地优先Locality-Aware实时推理请求延迟 50ms强制路由到本地集群。不跨集群避免网络延迟导致超时。区域弹性Regional Overflow交互式请求延迟 500ms优先本地本地 GPU 不足时路由到同一区域的远程集群。全局分发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_factorydict) last_updated: float field(default_factorytime.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(keylambda 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 ) # 随机扰动防止所有请求同时路由到同一个刚恢复的集群 # 为什么需要扰动 # 如果一个集群刚从过载恢复score0所有请求突然一起路由过去 # 瞬间又过载。扰动让请求分几次路由过去 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 finference-{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: fmy-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, namespaceinference, bodypod_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_namecluster_name, regionself._get_region(cluster_name), gpu_total8, gpu_available4, gpu_utilization_pct45.0, queue_length3, avg_inference_latency_ms250, 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, namepod_name, namespaceinference, ) 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(fPod 异常状态: {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, namepod_name, namespaceinference, tail_lines50, ) 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, namepod_name, namespaceinference, ) except Exception: pass四、跨集群调度的边界缺点跨区域网络延迟不可消除us-east→ap-southeast 的 RTT 约 200ms如果交互式请求的延迟容忍度是 500ms留给你推理的时间只有 300ms。双写存储成本推理结果需要通过 S3 回传每个结果额外占用存储和网络带宽。对于 50MB 的音频推理结果S3 上传下载可能增加 5-10s。不一致的多集群状态全局调度器维护各集群的实时健康状态但由于网络延迟状态可能滞后 10-15 秒。滞后期间可能做出次优调度决策。禁用场景延迟 50ms 的实时推理跨集群通信的延迟本身已超过容忍度。数据本地性强的推理如大型向量数据库跨集群推理需要跨集群数据传输成本过高。安全合规要求数据不出区域的场景。五、总结Agent 跨集群调度的核心是引入一个全局调度器根据请求延迟容忍度将推理任务分配到不同集群。三层路由策略实时→本地、交互→区域、批处理→全局平衡了延迟和资源利用率。工程实现要点通过集群健康探测维护全局状态、用 S3 做跨集群结果传递、TTL 自动清理推理 Pod 防止资源泄漏。跨集群的物理延迟是硬约束调度器只能优化不能消除。