1
# kubectl 生成的 Pod 定义
2
apiVersion: v1
3
kind: Pod
4
metadata:
5
  name: nginx
6
  labels:
7
    run: nginx
8
spec:
9
  containers:
10
    - name: nginx
11
      image: nginx
12
      resources: {}
13
  dnsPolicy: ClusterFirst
14
  restartPolicy: Always

1
// kubectl 创建资源流程（简化）
2
func (o *RunOptions) Run() error {
3
    // 1. 构建 Pod 对象
4
    pod := o.createPod()
5

6
    // 2. 客户端验证
7
    if errs := validation.ValidatePod(pod); len(errs) > 0 {
8
        return errs.ToAggregate()
9
    }
10

11
    // 3. 发送到 API Server
12
    result, err := o.Client.CoreV1().Pods(namespace).Create(
13
        context.TODO(), pod, metav1.CreateOptions{})
14

15
    return result, err
16
}

1
// API Server 存储接口（简化）
2
type Registry interface {
3
    Create(ctx context.Context, obj runtime.Object) error
4
    Update(ctx context.Context, obj runtime.Object) error
5
    Get(ctx context.Context, name string) (runtime.Object, error)
6
    Delete(ctx context.Context, name string) error
7
}
8

9
// etcd 存储实现
10
func (s *store) Create(ctx context.Context, obj runtime.Object) error {
11
    key := s.keyFunc(obj)
12
    data, _ := s.codec.Encode(obj)
13
    return s.client.Put(ctx, key, string(data))
14
}

1
// Deployment Controller（简化）
2
func (dc *DeploymentController) reconcileDeployment(d *apps.Deployment) error {
3
    // 1. 获取关联的 ReplicaSet
4
    rsList := dc.getReplicaSetsForDeployment(d)
5

6
    // 2. 计算需要的 Pod 数量
7
    desiredReplicas := int(*d.Spec.Replicas)
8
    currentReplicas := len(rsList)
9

10
    // 3. 创建/删除 ReplicaSet
11
    if currentReplicas < desiredReplicas {
12
        dc.createReplicaSet(d)
13
    }
14

15
    return nil
16
}
17

18
// ReplicaSet Controller（简化）
19
func (rsc *ReplicaSetController) syncReplicaSet(rs *apps.ReplicaSet) error {
20
    // 1. 获取关联的 Pod
21
    podList := rsc.getPodsForReplicaSet(rs)
22

23
    // 2. 计算差异
24
    diff := len(podList) - int(*rs.Spec.Replicas)
25

26
    // 3. 创建/删除 Pod
27
    if diff < 0 {
28
        for i := 0; i < -diff; i++ {
29
            rsc.createPod(rs)
30
        }
31
    }
32

33
    return nil
34
}

1
// 预选算法（部分）
2
var predicates = map[string]Predicate{
3
    "PodFitsResources":        PodFitsResources,      // 资源充足
4
    "PodFitsHostPorts":        PodFitsHostPorts,      // 端口可用
5
    "PodMatchNodeSelector":    PodMatchNodeSelector,  // 节点选择器
6
    "PodToleratesNodeTaints":  PodToleratesNodeTaints,// 污点容忍
7
    "CheckNodeCondition":      CheckNodeCondition,    // 节点状态
8
    "CheckNodeUnschedulable":  CheckNodeUnschedulable,// 不可调度标记
9
}

1
# Pod 资源请求
2
resources:
3
  requests:
4
    cpu: 500m
5
    memory: 512Mi
6
  limits:
7
    cpu: 1
8
    memory: 1Gi
9

10
# 节点资源
11
Allocatable:
12
  cpu: 4
13
  memory: 8Gi
14
Allocated:
15
  cpu: 2.5
16
  memory: 4Gi
17
# 可用: cpu 1.5, memory 4Gi -> 满足要求

1
// 优选算法（部分）
2
var priorities = map[string]Priority{
3
    "LeastRequestedPriority":  LeastRequestedPriority,  // 资源使用最少
4
    "BalancedResourceAllocation": BalancedResourceAllocation, // 资源均衡
5
    "NodeAffinityPriority":    NodeAffinityPriority,    // 节点亲和性
6
    "InterPodAffinityPriority": InterPodAffinityPriority, // Pod 亲和性
7
    "TaintTolerationPriority": TaintTolerationPriority, // 污点容忍
8
    "ImageLocalityPriority":   ImageLocalityPriority,   // 镜像本地化
9
}

1
节点评分计算：
2
Node A:
3
  - LeastRequested: 80 分
4
  - BalancedResource: 70 分
5
  - ImageLocality: 90 分
6
  - 综合得分: 80 * 1 + 70 * 1 + 90 * 2 = 330
7

8
Node B:
9
  - LeastRequested: 60 分
10
  - BalancedResource: 80 分
11
  - ImageLocality: 50 分
12
  - 综合得分: 60 * 1 + 80 * 1 + 50 * 2 = 240
13

14
选择 Node A

1
// 调度绑定
2
func (sched *Scheduler) bind(assumed *v1.Pod, targetNode string) error {
3
    binding := &v1.Binding{
4
        ObjectMeta: assumed.ObjectMeta,
5
        Target: v1.ObjectReference{
6
            Kind: "Node",
7
            Name: targetNode,
8
        },
9
    }
10

11
    return sched.Client.CoreV1().Pods(assumed.Namespace).Bind(
12
        context.TODO(), binding, metav1.CreateOptions{})
13
}

1
// Kubelet Pod 同步（简化）
2
func (kl *Kubelet) syncPod(pod *v1.Pod) error {
3
    // 1. 检查 Pod 是否可以运行
4
    if err := kl.canRunPod(pod); err != nil {
5
        return err
6
    }
7

8
    // 2. 创建 Pod 目录
9
    kl.makePodDataDirs(pod)
10

11
    // 3. 挂载卷
12
    if err := kl.volumeManager.WaitForAttachAndMount(pod); err != nil {
13
        return err
14
    }
15

16
    // 4. 拉取镜像
17
    for _, container := range pod.Spec.Containers {
18
        kl.imageManager.PullImage(container.Image)
19
    }
20

21
    // 5. 创建 Pod Sandbox
22
    podSandboxID := kl.runtimeService.CreatePodSandbox(pod)
23

24
    // 6. 创建容器
25
    for _, container := range pod.Spec.Containers {
26
        kl.createContainer(pod, container, podSandboxID)
27
    }
28

29
    // 7. 启动容器
30
    for _, container := range pod.Spec.Containers {
31
        kl.runtimeService.StartContainer(containerID)
32
    }
33

34
    return nil
35
}

1
// PLEG 定期检查容器状态
2
func (pleg *GenericPLEG) relist() {
3
    // 1. 获取所有容器状态
4
    containers := pleg.runtime.GetPods()
5

6
    // 2. 与缓存状态对比
7
    for _, pod := range containers {
8
        oldPod := pleg.podRecords[pod.ID]
9

10
        // 3. 检测变化
11
        events := pleg.generateEvents(oldPod, pod)
12

13
        // 4. 发送事件
14
        for _, event := range events {
15
            pleg.eventChannel <- event
16
        }
17
    }
18
}

1
// CRI 服务定义（简化）
2
service RuntimeService {
3
    rpc CreatePodSandbox(CreatePodSandboxRequest) returns (CreatePodSandboxResponse);
4
    rpc StopPodSandbox(StopPodSandboxRequest) returns (StopPodSandboxResponse);
5
    rpc RemovePodSandbox(RemovePodSandboxRequest) returns (RemovePodSandboxResponse);
6

7
    rpc CreateContainer(CreateContainerRequest) returns (CreateContainerResponse);
8
    rpc StartContainer(StartContainerRequest) returns (StartContainerResponse);
9
    rpc StopContainer(StopContainerRequest) returns (StopContainerResponse);
10
    rpc RemoveContainer(RemoveContainerRequest) returns (RemoveContainerResponse);
11

12
    rpc ListContainers(ListContainersRequest) returns (ListContainersResponse);
13
    rpc ContainerStatus(ContainerStatusRequest) returns (ContainerStatusResponse);
14
}

1
# Pause 容器是 Pod 的根容器
2
# 也称为 Sandbox 容器
3

4
# 作用：
5
# 1. 持有 Pod 的网络 namespace
6
# 2. 作为 Pod 中其他容器的父进程
7
# 3. 回收孤儿进程
8

9
# pause 镜像内容
10
FROM scratch
11
ARG ARCH
12
ADD bin/pause-${ARCH} /pause
13
ENTRYPOINT ["/pause"]

1
// Kubelet 状态更新
2
func (kl *Kubelet) updateStatus(pod *v1.Pod) error {
3
    // 1. 收集容器状态
4
    containerStatuses := kl.getContainerStatuses(pod)
5

6
    // 2. 构建 Pod 状态
7
    podStatus := &v1.PodStatus{
8
        Phase:             kl.getPhase(pod, containerStatuses),
9
        Conditions:        kl.getConditions(pod),
10
        ContainerStatuses: containerStatuses,
11
    }
12

13
    // 3. 更新到 API Server
14
    _, err := kl.kubeClient.CoreV1().Pods(pod.Namespace).UpdateStatus(
15
        context.TODO(), pod, metav1.UpdateOptions{})
16

17
    return err
18
}