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 的迁移成本，以及多约束冲突。拓扑调度的目标是让服务在拓扑层级上具备容灾能力，同时避免过度约束导致的调度失败。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
6
0- 0
已为社区贡献99条内容
所有评论(0)