Kubernetes 资源拓扑调度：从亲和性到拓扑扩展的调度策略
Kubernetes 资源拓扑调度：从亲和性到拓扑扩展的调度策略
一、K8s 调度的"盲区"：跨可用区部署的隐性成本
Kubernetes 默认调度器在分配 Pod 时考虑资源请求、亲和性和污点容忍，但对网络拓扑的感知有限。某在线教育平台将 100 个 Pod 调度到 3 个可用区，默认调度器随机分配，导致同一服务的多个副本集中在同一可用区。当该可用区故障时，服务可用性从 99.99% 骤降至 66%。更隐蔽的问题是跨可用区的网络延迟：同一可用区内延迟 < 0.5ms，跨可用区延迟 2-5ms，数据库访问跨可用区后 P99 延迟增加 300%。
拓扑感知调度要求调度器理解节点间的拓扑关系（可用区、机架、NUMA 节点），并根据业务需求做出合理的分布决策。
二、K8s 拓扑调度的层级与策略
flowchart TB subgraph 拓扑层级["拓扑层级"] direction TB L1[区域 Region<br/>跨地域容灾] L2[可用区 Zone<br/>电力/网络隔离] L3[机架 Rack<br/>交换机隔离] L4[NUMA 节点<br/>内存访问延迟] end subgraph 调度策略["调度策略"] direction LR S1[Pod 拓扑分布约束<br/>topologySpreadConstraints<br/>均匀分布] S2[节点亲和性<br/>nodeAffinity<br/>指定拓扑域] S3[服务亲和性<br/>serviceAffinity<br/>同拓扑域优先] end subgraph 扩展机制["扩展机制"] direction LR E1[调度框架<br/>Scheduler Framework<br/>Plugin 扩展] E2[调度器配置<br/>Profile + Plugin<br/>多调度器] E3[Descheduler<br/>事后重平衡<br/>违反约束时迁移] end L1 --> S1 & S2 L2 --> S1 & S3 L3 --> S3 L4 --> E1 S1 --> E1 S2 --> E2 S3 --> E3 style 拓扑层级 fill:#eef,stroke:#333 style 调度策略 fill:#fee,stroke:#333 style 扩展机制 fill:#efe,stroke:#333三、K8s 拓扑调度的代码实现
from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple from enum import Enum from collections import defaultdict import math class TopologyLevel(Enum): REGION = "region" ZONE = "zone" RACK = "rack" NODE = "node" @dataclass class NodeInfo: """节点信息""" name: str zone: str region: str rack: str cpu_capacity: int # CPU 核数 cpu_allocatable: int memory_capacity: int # MB memory_allocatable: int labels: Dict[str, str] = field(default_factory=dict) @dataclass class PodInfo: """Pod 信息""" name: str namespace: str app_label: str cpu_request: int # millicores memory_request: int # MB preferred_zone: Optional[str] = None current_node: Optional[str] = None @dataclass class TopologySpreadConstraint: """拓扑分布约束""" topology_key: str # topology.kubernetes.io/zone max_skew: int # 最大偏差 when_unsatisfiable: str # DoNotSchedule / ScheduleAnyway label_selector: Dict # 匹配的 Pod 标签 class TopologyAwareScheduler: """ 拓扑感知调度器：实现 Pod 拓扑分布约束 """ def __init__(self): self._nodes: Dict[str, NodeInfo] = {} self._pods: List[PodInfo] = [] def add_node(self, node: NodeInfo): self._nodes[node.name] = node def add_pod(self, pod: PodInfo): self._pods.append(pod) # ============ 拓扑分布计算 ============ def get_topology_distribution(self, app_label: str, topology_key: str) -> Dict[str, int]: """获取指定应用在指定拓扑域的分布""" distribution = defaultdict(int) for pod in self._pods: if pod.app_label != app_label or not pod.current_node: continue node = self._nodes.get(pod.current_node) if not node: continue if topology_key == "topology.kubernetes.io/zone": domain = node.zone elif topology_key == "topology.kubernetes.io/region": domain = node.region elif topology_key == "rack": domain = node.rack else: domain = node.labels.get(topology_key, "unknown") distribution[domain] += 1 return dict(distribution) def calculate_skew(self, distribution: Dict[str, int]) -> int: """计算最大偏差""" if not distribution: return 0 return max(distribution.values()) - min(distribution.values()) # ============ 调度决策 ============ def schedule(self, pod: PodInfo, constraint: TopologySpreadConstraint) -> Optional[str]: """ 为 Pod 选择最优节点 核心逻辑：选择使拓扑偏差最小的域中的可用节点 """ # Step 1: 获取当前分布 distribution = self.get_topology_distribution( pod.app_label, constraint.topology_key ) # Step 2: 获取所有拓扑域 all_domains = self._get_all_domains(constraint.topology_key) # 补全分布（无 Pod 的域也要考虑） for domain in all_domains: if domain not in distribution: distribution[domain] = 0 # Step 3: 选择 Pod 数最少的域 min_count = min(distribution.values()) candidate_domains = [ d for d, c in distribution.items() if c == min_count ] # Step 4: 检查偏差约束 if constraint.when_unsatisfiable == "DoNotSchedule": # 严格模式：如果调度后偏差超过 max_skew，拒绝调度 for domain in candidate_domains: new_distribution = dict(distribution) new_distribution[domain] += 1 new_skew = self.calculate_skew(new_distribution) if new_skew <= constraint.max_skew: # 在该域中选择资源最充足的节点 node = self._select_node_in_domain( domain, pod, constraint.topology_key ) if node: return node return None # 无法满足约束 else: # 宽松模式：优先选择偏差最小的域，但不拒绝 # 按域的 Pod 数升序排列 sorted_domains = sorted( distribution.items(), key=lambda x: x[1] ) for domain, _ in sorted_domains: node = self._select_node_in_domain( domain, pod, constraint.topology_key ) if node: return node return None def _get_all_domains(self, topology_key: str) -> List[str]: """获取所有拓扑域""" domains = set() for node in self._nodes.values(): if topology_key == "topology.kubernetes.io/zone": domains.add(node.zone) elif topology_key == "topology.kubernetes.io/region": domains.add(node.region) elif topology_key == "rack": domains.add(node.rack) else: domains.add(node.labels.get(topology_key, "unknown")) return list(domains) def _select_node_in_domain(self, domain: str, pod: PodInfo, topology_key: str) -> Optional[str]: """在指定拓扑域中选择资源最充足的节点""" candidates = [] for node in self._nodes.values(): # 检查节点是否属于目标域 if topology_key == "topology.kubernetes.io/zone": if node.zone != domain: continue elif topology_key == "topology.kubernetes.io/region": if node.region != domain: continue elif topology_key == "rack": if node.rack != domain: continue # 检查资源是否充足 if (node.cpu_allocatable >= pod.cpu_request and node.memory_allocatable >= pod.memory_request): # 计算可用资源分数 score = ( node.cpu_allocatable * 10 + node.memory_allocatable / 1024 ) candidates.append((node.name, score)) if not candidates: return None # 选择分数最高的节点 candidates.sort(key=lambda x: x[1], reverse=True) return candidates[0][0] # ============ K8s Manifest 生成 ============ class TopologyManifestGenerator: """生成 K8s 拓扑调度相关的 Manifest""" @staticmethod def generate_deployment_with_spread( app_name: str, replicas: int, image: str, zones: List[str], max_skew: int = 1, ) -> Dict: """生成带拓扑分布约束的 Deployment""" return { "apiVersion": "apps/v1", "kind": "Deployment", "metadata": {"name": app_name}, "spec": { "replicas": replicas, "selector": { "matchLabels": {"app": app_name} }, "template": { "metadata": { "labels": {"app": app_name} }, "spec": { "topologySpreadConstraints": [{ "maxSkew": max_skew, "topologyKey": "topology.kubernetes.io/zone", "whenUnsatisfiable": "DoNotSchedule", "labelSelector": { "matchLabels": {"app": app_name} }, }], "affinity": { "podAntiAffinity": { "preferredDuringSchedulingIgnoredDuringExecution": [{ "weight": 100, "podAffinityTerm": { "labelSelector": { "matchLabels": {"app": app_name} }, "topologyKey": "kubernetes.io/hostname", }, }], }, }, "containers": [{ "name": app_name, "image": image, "resources": { "requests": { "cpu": "100m", "memory": "128Mi", }, }, }], }, }, }, } @staticmethod def generate_descheduler_policy() -> Dict: """生成 Descheduler 策略：定期重平衡""" return { "apiVersion": "descheduler/v1alpha1", "kind": "DeschedulerPolicy", "strategies": { "RemoveDuplicates": { "enabled": True, }, "LowNodeUtilization": { "enabled": True, "params": { "nodeResourceUtilizationThresholds": { "thresholds": { "cpu": 40, "memory": 40, }, "targetThresholds": { "cpu": 70, "memory": 70, }, }, }, }, "PodLifeTime": { "enabled": True, "params": { "maxPodLifeTimeSeconds": 86400, }, }, }, } # ============ 模拟与验证 ============ class TopologySimulator: """拓扑调度模拟器：验证分布效果""" def __init__(self): self._scheduler = TopologyAwareScheduler() def simulate(self, nodes: List[NodeInfo], pods: List[PodInfo], constraint: TopologySpreadConstraint) -> Dict: """模拟调度并输出分布结果""" for node in nodes: self._scheduler.add_node(node) results = {"scheduled": [], "failed": [], "distribution": {}} for pod in pods: node_name = self._scheduler.schedule(pod, constraint) if node_name: pod.current_node = node_name self._scheduler.add_pod(pod) results["scheduled"].append({ "pod": pod.name, "node": node_name, }) else: results["failed"].append(pod.name) # 最终分布 results["distribution"] = self._scheduler.get_topology_distribution( pods[0].app_label if pods else "", constraint.topology_key, ) # 计算偏差 results["skew"] = self._scheduler.calculate_skew( results["distribution"] ) return results四、K8s 拓扑调度的 Trade-offs
均匀分布与资源利用率的矛盾。严格的拓扑分布约束（maxSkew=1）确保均匀分布，但可能导致资源碎片化——某可用区资源充足但 Pod 数已达上限，新 Pod 被迫调度到资源紧张的可用区。建议对核心服务使用严格约束，对非核心服务使用宽松约束（ScheduleAnyway）。
Pod 反亲和性的爆炸效应。podAntiAffinity要求同一服务的 Pod 不在同一节点上，当副本数超过节点数时，调度会失败。在大规模集群中，反亲和性的计算复杂度随 Pod 数量二次增长，调度延迟显著增加。
Descheduler 的迁移成本。Descheduler 通过驱逐 Pod 来重平衡分布，但每次驱逐都会触发 Pod 重建，增加服务中断风险。建议仅在偏差严重时触发（如 skew > 3），并配置 PDB（PodDisruptionBudget）限制并发驱逐数。
多约束冲突。同时设置拓扑分布约束、节点亲和性和 Pod 反亲和性时，约束之间可能冲突。例如节点亲和性要求调度到 zone-a，但拓扑分布约束要求均匀分布到所有可用区。K8s 调度器按优先级处理，但调试约束冲突是运维中的常见痛点。
五、总结
K8s 拓扑感知调度通过 topologySpreadConstraints 实现跨可用区的均匀分布，通过节点亲和性指定拓扑域偏好，通过 Descheduler 事后重平衡违反约束的分布。调度决策的核心逻辑是选择使拓扑偏差最小的域中的可用节点。关键权衡在于均匀分布与资源利用率、Pod 反亲和性的爆炸效应、Descheduler 的迁移成本，以及多约束冲突。拓扑调度的目标是让服务在拓扑层级上具备容灾能力，同时避免过度约束导致的调度失败。