Kubernetes 源码研习社是由 云原生社区 组织的 Kubernetes 源码特别兴趣小组（SIG），由热爱学习、注重个人成长的一帮小伙伴们自由、自愿成立的小组。每个人都非常希望从 Kubernetes 上学到知识，帮助自己实现成长和进步。欢迎加入，一起坚持，一起克服，一起成长。

源码研习社第二期（进行中）

本期主题：kube-scheduler 源码剖析

活动时间：2020.10.12 开始

如何报名：报名方式

活动介绍

Kubernetes 源码 scheduler 剖析，干就完事了。每周学习目标：

• 每周写笔记做总结。笔记链接：https://docs.qq.com/sheet/DR01kdWZkUmFLc0Jh?tab=o3eynn

• 每周六晚 7-10 点固定在线研讨 Kubernetes 调度器问题。腾讯会议号：4967324951

• 每日讨论 Kubernetes 源码问题

• 参阅本项目推荐的 kubernetes 相关文章

本期学习计划

坚持就是胜利

推荐资料

scheduler 原理解析

Kubernetes Scheduler 设计与实现 【 Go 夜读 】

scheduler 源码剖析

k8s scheduler 源码阅读 【 Go 夜读 】

kube-scheduler 源码分析 -- 作者 田雨飞

Kubernetes Scheduler 源码全解析-全景图

调度器核心数据结构与算法分析 -- 云原生学习笔记 -- 作者 baxiaoshi

Kubernetes 源码分析之 kube-scheduler

非常感谢社区abserari整理的以下的学习资料 源码研习社特供版图解scheduler

scheduler 扩展开发

自定义 Kubernetes 调度器

scheduler 落地实践

scheduler 其他好料

调度器 SIG

scheduler 包括其他的一些组件相关解析

调度系统设计精要

阅读源码的建议

如何高效阅读 Kubernetes 源码？

欢迎贡献

合适的文章有非常多，如果有合适的欢迎提交 PR（Pull Request）合入推荐阅读文章。

你能收获什么？

• 对 Kubernetes 核心源码有更深刻的理解

• 一群热爱云原生的志同道合的朋友

如何报名

进入报名 excel 表，填写自己信息即被认为是报名参加活动，每周按要求完成总结笔记，参与每周周末的讨论即可

报名链接：https://docs.qq.com/sheet/DR01kdWZkUmFLc0Jh?tab=o3eynn

加入我们

源码研习社也有自己的微信群，如何加入？

扫描下面的二维码，添加 Jimmy Song 好友，备注姓名-公司，留言“加入源码研习社”即可。

嘉宾介绍

郑东旭（Derek Zheng） BFE（万亿流量转发引擎）开源项目的作者之一，《Kubernetes 源码剖析》作者，擅长 Linux 下高性能服务器的开发，对云计算、区块链相关技术领域有深刻的理解。

源码研习社 SIG 小组成员

SIG 的全称是 Special Interests Group, 或称 Super Intellectual Genius。 源码研习社 SIG 小组负责源码研习社活动的日常维护，目前的核心成员包括：

• 厉辉 yousali

• 金润森

• 王文虎

• 赵卫国

• 王冬

• 梁远鹏

Q&A:

关于本次活动有任何问题，请在 Issue 中提问，也可以参与 社区会议 反馈。

欢迎来到云原生社区 Kubernetes SIG（特别兴趣小组）！

加入 Kubernetes SIG

请填写申请表加入微信交流群。

如何提问

有关于 Kubernetes 的问题请提交 Issue 后将链接发到微信群中讨论。

Parallelize

Stop Channel Mode

Group

Group 增强了 Go 标准库中 sync.Group 的能力。用于执行通过 Context 或 channel 来控制终止条件的方法，并将该方法在一个独立 Go Routine 中执行。

Forever & Until

JitterUntil

从代码的注释看，我们需要解决两个关键问题，sliding 如何起作用，jitterFactor 如何起作用，解决了这两个问题，一切就清晰了。

Sliding

从代码中，不难看出，所谓 sliding，就是时间间隔是否包含执行时间。按 170 行的代码，并不难猜出，Backoff() 返回的是一个 Timer 类型，那么该 Timer 启动时间即关键。 这个代码需要注意一个细节：从 167 行开始的 select 并不能保证顺序，换句话说，如果 timer 已经触发且 stopCh 已经关闭时，未必能保证一定退出。但进入下轮循环后，由于第 144 行代码，一定能确保程序正常退出。

JitterFactor

jitterFactor 是依赖于 BackoffManager 起作用的，先看一下创建过程，简单的保留了配置参数及其他关联对象。

再看其 Backoff 方法的实现，注意 379 ～ 383 行，确保只有一个 Timer 在工作。

继续看 Jitter 实现，在固定的 duration 上添加了一个动态值，至此，两个问题都解决了。

Design

• Sandbox: 协议栈，可包含多个 Endpoint，可通过 Namespace、Jail 等实现

• Endpoint: 将 Sandbox 与 Network 连接

• Network: 可直接通信的 Endpoint 的集合，可使用 Bridge、VLAN 等实现

Docker Architecture

Network Controller

Initialize Network Controllers

Docker Daemon 管理可用的 NetworkController。在启动 Daemon 时，会创建当前操作系统下全部可用的 NetworkController，以 daemon_unix.go 为例，创建了 none、host、bridge 三种模式的网络控制器。

func (daemon *Daemon) initNetworkController(config *config.Config, activeSandboxes map[string]interface{}) (libnetwork.NetworkController, error) {
    netOptions, err := daemon.networkOptions(config, daemon.PluginStore, activeSandboxes)
    if err != nil {
        return nil, err
    }

    controller, err := libnetwork.New(netOptions...)
    if err != nil {
        return nil, fmt.Errorf("error obtaining controller instance: %v", err)
    }

    if len(activeSandboxes) > 0 {
        logrus.Info("There are old running containers, the network config will not take affect")
        return controller, nil
    }

    // Initialize default network on "null"
    if n, _ := controller.NetworkByName("none"); n == nil {
        if _, err := controller.NewNetwork("null", "none", "", libnetwork.NetworkOptionPersist(true)); err != nil {
            return nil, fmt.Errorf("Error creating default \"null\" network: %v", err)
        }
    }

    // Initialize default network on "host"
    if n, _ := controller.NetworkByName("host"); n == nil {
        if _, err := controller.NewNetwork("host", "host", "", libnetwork.NetworkOptionPersist(true)); err != nil {
            return nil, fmt.Errorf("Error creating default \"host\" network: %v", err)
        }
    }

    // Clear stale bridge network
    if n, err := controller.NetworkByName("bridge"); err == nil {
        if err = n.Delete(); err != nil {
            return nil, fmt.Errorf("could not delete the default bridge network: %v", err)
        }
        if len(config.NetworkConfig.DefaultAddressPools.Value()) > 0 && !daemon.configStore.LiveRestoreEnabled {
            removeDefaultBridgeInterface()
        }
    }

    if !config.DisableBridge {
        // Initialize default driver "bridge"
        if err := initBridgeDriver(controller, config); err != nil {
            return nil, err
        }
    } else {
        removeDefaultBridgeInterface()
    }

    // Set HostGatewayIP to the default bridge's IP  if it is empty
    if daemon.configStore.HostGatewayIP == nil && controller != nil {
        if n, err := controller.NetworkByName("bridge"); err == nil {
            v4Info, v6Info := n.Info().IpamInfo()
            var gateway net.IP
            if len(v4Info) > 0 {
                gateway = v4Info[0].Gateway.IP
            } else if len(v6Info) > 0 {
                gateway = v6Info[0].Gateway.IP
            }
            daemon.configStore.HostGatewayIP = gateway
        }
    }
    return controller, nil
}

NetworkController Implementation

controller 是 libnetwork 中对 NetworkController 的实现。可以看到，controller 通过驱动表来区分不同类型的网络，使用驱动创建 Network 及 Endpoint，并将 Endpoint 加入 Sandbox 或移除出 Sandbox。 Container 通过 SandboxID 以及 SandboxKey 来找到对应的 Sandbox。Sandbox 可以使用 containerID 来确定是否归属于某个 Container。

OS Layer Sandbox

Namespace

Sandbox 接口没有列举出全部功能，只是能看出其能力边界的部分功能。后续以 Namespace 方式实现的 Sandbox 为例。 通过上图，并不难看出，路由、接口等功能应该是由 netlink 提供的，Namespace 获取 netlink 方式如下，需要注意，Namespace 内 netlink 配置，仅在 Namespace 内有效。根据 Namespace 获取 netlink 的关键方法如下

func GetFromPath(path string) (NsHandle, error) {
    fd, err := syscall.Open(path, syscall.O_RDONLY, 0)
    if err != nil {
        return -1, err
    }
    return NsHandle(fd), nil
}

使用返回的 NsHandle 就可以创建具体的 SocketHandle 了，方法如下

func NewHandleAt(ns netns.NsHandle, nlFamilies ...int) (*Handle, error) {
    return newHandle(ns, netns.None(), nlFamilies...)
}

// NewHandleAtFrom works as NewHandle but allows client to specify the
// new and the origin netns Handle.
func NewHandleAtFrom(newNs, curNs netns.NsHandle) (*Handle, error) {
    return newHandle(newNs, curNs)
}

func newHandle(newNs, curNs netns.NsHandle, nlFamilies ...int) (*Handle, error) {
    h := &Handle{sockets: map[int]*nl.SocketHandle{}}
    fams := nl.SupportedNlFamilies
    if len(nlFamilies) != 0 {
        fams = nlFamilies
    }
    for _, f := range fams {
        s, err := nl.GetNetlinkSocketAt(newNs, curNs, f)
        if err != nil {
            return nil, err
        }
        h.sockets[f] = &nl.SocketHandle{Socket: s}
    }
    return h, nil
}

Add Interface

Bridge Network

Create Network

根据配置文件中 BridgeName 查找系统中已存在的 Link 实例，如果 BridgeName 为空，使用默认网桥 docker0。

func newInterface(nlh *netlink.Handle, config *networkConfiguration) (*bridgeInterface, error) {
    var err error
    i := &bridgeInterface{nlh: nlh}

    // Initialize the bridge name to the default if unspecified.
    if config.BridgeName == "" {
        config.BridgeName = DefaultBridgeName
    }

    // Attempt to find an existing bridge named with the specified name.
    i.Link, err = nlh.LinkByName(config.BridgeName)
    if err != nil {
        logrus.Debugf("Did not find any interface with name %s: %v", config.BridgeName, err)
    } else if _, ok := i.Link.(*netlink.Bridge); !ok {
        return nil, fmt.Errorf("existing interface %s is not a bridge", i.Link.Attrs().Name)
    }
    return i, nil
}

创建 bridgeNetwork 实例，并存入 networks

// Create and set network handler in driver
network := &bridgeNetwork{
    id:         config.ID,
    endpoints:  make(map[string]*bridgeEndpoint),
    config:     config,
    portMapper: portmapper.New(d.config.UserlandProxyPath),
    bridge:     bridgeIface,
    driver:     d,
}

d.Lock()
d.networks[config.ID] = network
d.Unlock()

如果获取的 bridgeInterface 中不存在有效网桥设备，则将创建设备、sysctl 方法加入设置队列；如果使用 docker0，仅将 sysctl 方法加入设置队列

bridgeAlreadyExists := bridgeIface.exists()
if !bridgeAlreadyExists {
    bridgeSetup.queueStep(setupDevice)
    bridgeSetup.queueStep(setupDefaultSysctl)
}

// For the default bridge, set expected sysctls
if config.DefaultBridge {
    bridgeSetup.queueStep(setupDefaultSysctl)
}

根据配置文件参数，将对应的设置方法加入设置队列

for _, step := range []struct {
        Condition bool
        Fn        setupStep
}{
    // Enable IPv6 on the bridge if required. We do this even for a
    // previously  existing bridge, as it may be here from a previous
    // installation where IPv6 wasn't supported yet and needs to be
    // assigned an IPv6 link-local address.
    {config.EnableIPv6, setupBridgeIPv6},

    // We ensure that the bridge has the expectedIPv4 and IPv6 addresses in
    // the case of a previously existing device.
    {bridgeAlreadyExists && !config.InhibitIPv4, setupVerifyAndReconcile},

    // Enable IPv6 Forwarding
    {enableIPv6Forwarding, setupIPv6Forwarding},

    // Setup Loopback Addresses Routing
    {!d.config.EnableUserlandProxy, setupLoopbackAddressesRouting},

    // Setup IPTables.
    {d.config.EnableIPTables, network.setupIPTables},

    //We want to track firewalld configuration so that
    //if it is started/reloaded, the rules can be applied correctly
    {d.config.EnableIPTables, network.setupFirewalld},

    // Setup DefaultGatewayIPv4
    {config.DefaultGatewayIPv4 != nil, setupGatewayIPv4},

    // Setup DefaultGatewayIPv6
    {config.DefaultGatewayIPv6 != nil, setupGatewayIPv6},

    // Add inter-network communication rules.
    {d.config.EnableIPTables, setupNetworkIsolationRules},

    //Configure bridge networking filtering if ICC is off and IP tables are enabled
    {!config.EnableICC && d.config.EnableIPTables, setupBridgeNetFiltering},
} {
    if step.Condition {
        bridgeSetup.queueStep(step.Fn)
    }
}

将设备启动设置方法加入设置队列，并返回执行结果

bridgeSetup.queueStep(setupDeviceUp)
return bridgeSetup.apply()

Setup Device

创建 netlink.Bridge 结构体，LinkAttrs 中使用配置中的 BridgeName，然后，使用 netlink 方法创建网桥设备，如果需要设置 MAC 则随机生成 MAC 地址。

func setupDevice(config *networkConfiguration, i *bridgeInterface) error {
    var setMac bool

    // We only attempt to create the bridge when the requested device name is
    // the default one.
    if config.BridgeName != DefaultBridgeName && config.DefaultBridge {
        return NonDefaultBridgeExistError(config.BridgeName)
    }

    // Set the bridgeInterface netlink.Bridge.
    i.Link = &netlink.Bridge{
        LinkAttrs: netlink.LinkAttrs{
            Name: config.BridgeName,
        },
    }

    // Only set the bridge's MAC address if the kernel version is > 3.3, as it
    // was not supported before that.
    kv, err := kernel.GetKernelVersion()
    if err != nil {
        logrus.Errorf("Failed to check kernel versions: %v. Will not assign a MAC address to the bridge interface", err)
    } else {
        setMac = kv.Kernel > 3 || (kv.Kernel == 3 && kv.Major >= 3)
    }

    if setMac {
        hwAddr := netutils.GenerateRandomMAC()
        i.Link.Attrs().HardwareAddr = hwAddr
        logrus.Debugf("Setting bridge mac address to %s", hwAddr)
    }

    if err = i.nlh.LinkAdd(i.Link); err != nil {
        logrus.Debugf("Failed to create bridge %s via netlink. Trying ioctl", config.BridgeName)
        return ioctlCreateBridge(config.BridgeName, setMac)
    }

    return err
}

Bridge 设备创建、配置等，最终均通过 netlink 接口完成。

Networking Configuration

• System Control

  • /proc/sys/net/ipv6/conf/BridgeName/accept_ra -> 0：不接受路由建议

  • /proc/sys/net/ipv4/conf/BridgeName/route_localnet -> 1：将外部流量重定向至 loopback，需要配合 iptables 使用

IPTABLES

• INTERNAL

  • filter

    • DOCKER-ISOLATION-STAGE-1 -i BridgeInterface ! -d Network -j DROP

    • DOCKER-ISOLATION-STAGE-1 -o BridgeInterface ! -s Network -j DROP

• NON INTERNAL

  • nat

    • DOCKER -t nat -i BridgeInterface -j RETURN

  • filter

    • FORWARD -i BridgeInterface ! -o BridgeInterface -j ACCEPT

  • HOST IP != nil

    • nat

      • POSTROUTING -t nat -s BridgeSubnet ! -o BridgeInterface -j SNAT --to-source HOSTIP

      • POSTROUTING -t nat -m addrtype --src-type LOCAL -o BridgeInterface -j SNAT --to-source HOSTIP

  • HOST IP == nil

    • nat

      • POSTROUTING -t nat -s BridgeSubnet ! -o BridgeInterface -j MASQUERADE

      • POSTROUTING -t nat -m addrtype --src-type LOCAL -o BridgeInterface -j MASQUERADE

  • Inter Container Communication Enabled

    • filter

      • FORWARD -i BridgeInterface -o __BridgeInterface -j ACCEPT

  • Inter Container Communication Disabled

    • filter

      • FORWARD -i BridgeInterface -o __BridgeInterface -j DROP

  • nat

    • PREROUTING -m addrtype --dst-type LOCAL -j DOCKER

    • OUTPUT -m addrtype --dst-type LOCAL -j DOCKER

  • filter

    • FORWARD -o BridgeInterface -j DOCKER

    • FORWARD -o BridgeInterface -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT

  • filter

    • -I FORWARD -j DOCKER-ISOLATION-STAGE-1

Create Endpoint

全局有一个默认 Bridge 设备 docker0，每个 Container 有自己独立的网络协议栈，容器网络和通过 veth 对与 Bridge 设备互通。 同一节点上不同 Container 间，通过 ARP 协议，即可进行 3 层通信；Container 出 Node 网络可以通过默认网关设备 docker0，再经过 IPTABLES 重定向至 eth0。

Runner

Landscape

References

简介

大家好，这一次给大家带来的是 Kubernetes 的源码阅读。

源码设计图需要配合源码使用，除展示关键设计外，还会有问题性的留白。阅读图的时候发现存在问题的，带着问题去看源码，看别人如何设计解决的，会提高设计能力。

如果发现图片模糊的情况，是图片压缩所致，右键另存为下载原图或者联系我（abser9216）发 svg 原图。

图作者 GitHub：，

Architecture

Container

Components

Definition

Creation

timeBudget 定义如下

创建代码如下

Process

Bonus

timeBudget 开始运行后，创建了一个工作协程。在这个协程中，每秒会触发一次增加预算的操作，如果预算大于上限，则取上限。

Take & Return

takeAvailable 一次性获取全部预算，并重置预算；returnUnused 归还剩余的预算。

Definition

Relationships

ObjectConvertor

New

AddKnownTypes

Type Default Function

SchemeBuilder

Buffer

Ring Buffer

Overview

从命名来看，CacheableObject 用于保存 Object 实例，在 Kubernetes 版本 1.18 中，只要一个结构 cachingObject 实现了该接口，关系大致如下图

再结合 CacheableObject 定义，会发现：向 CacheableObject 存入 Object 时，指定了 Identifier，那是否意味着可以存入多个 Object 呢？而 GetObject 方法，确只返回了一个 Object 实例，虽然不能排除 slice、map 等容器类结构实现 Object 接口的可能性，但也是一个需要深入了解并解决的问题。让我们继续，并从尝试解决这两个问题开始，看看能有什么收获。

cachingObject

Overview

每个 cachingObject 实际存储一个 Object，即 metaRuntimeInterface 实例。根据不同的 Identifier，将这个对象编码为不同格式，并缓存在 map 中。

Implementation

Definition

metaRuntimeInterface 同时实现了 runtime.Object 与 metav1.Object 接口

metav1.Object 接口定义如下，用于描述 Kubernetes 核心对象

Creation

从 new 方法，可以看到，一个 cachingObject 存入的确实是一个实例，并且存入实例时，存储的并不是原始对象，而是一份儿深度拷贝。

GetObject 方法获取存入的 metaRuntimeInterface 实例的一份儿深度拷贝。

Implementation for runtime.Object

cachingObject 本身也满足 runtime.Object 接口，实现如下。这里需要注意，在 DeepCopyObject 方法中，创建了新的 serializationCache，且没有复制旧内容。

Implementation For CacheableObject

重点是更换 atomic.Value 操作过程，如下所示

Object

Overview

Object 实例如下所示，基本都在 pkg/apis 目录下，自己查找即可。

Unstructured

Unstructured 与 Object 配合使用的场景如下

实例图如下

Scheme

Builder

Scheme

AddKnownTypes

AddUnversionedTypes

nameFunc

Others

Conversion

Landscape

Core Function Definitions

DefaultNameFunc 实现如下所示

ConversionFunc 声明如下所示

FieldMappingFunc 将 key 转换为源结构及目标结构中的 Field

ConversionFuncs

Converter

将下列方法做简要说明：

doConversion

Scope

defaultConvert

然后根据 dv.Kind() 分别进行处理。

dv.Kind() -> reflect.Struct

dv.Kind() -> reflect.Slice

dv.Kind() -> reflect.Ptr

dv.Kind() -> reflect.Interface

dv.Kind() -> reflect.Map

Overview

大家好，这一次给大家带来的是 ETCD 的源码阅读。本文写就时是三部分，方便大家阅读，合成一篇，分别是 Server 篇， Storage 篇和 Utility 篇。配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Server

Single

Landscape

Clients 中包含了 etcd 服务器要监听的地址，地址可为 TCP、Unix Socket 形式，且支持 http 与 https。serverCtx 与一个 net.Listener 匹配，独立运行于一个 goroutine。

Serve Procedure

Backend

Landscape

Storage

BoltDB Backend

Landscape

run

启动 Timer，定时提交；或者在收到停止信号时，提交并退出。代码很简单，如下所示

func (b *backend) run() {
    defer close(b.donec)
    t := time.NewTimer(b.batchInterval)
    defer t.Stop()
    for {
        select {
        case <-t.C:
        case <-b.stopc:
            b.batchTx.CommitAndStop()
            return
        }
        b.batchTx.Commit()
        t.Reset(b.batchInterval)
    }
}

Transaction Relationship

Buffer

MVCC

Store

References

• ReadView

• WriteView

• KV

Watchable

Landscape

Watcher Creation

Nofity Waiter

Utility

CMux

• soheilhy/cmux: Connection multiplexer for GoLang: serve different services on the same port!

Landscape

Implementation

Scheduler

Landscape

WAL

Landscape

致谢一下响哥（李响）在我阅读 bboltDB 的源码的时候给予我的鼓励，使我对 ETCD 的源码产生兴趣（可惜到现在还没有见过一面）。本篇实际是在阅读 API Server 的时候完成的，本意也是为了更好的理解 API Server，API Server 部分图较多，也经历过多次重画，最近整理完毕会逐步放出，有意阅读交流的朋友请持续关注，或者催更（笑）。另外我时常会组织一起进行源码阅读并画图的活动（通常是我和我对象，本篇的 Server 部分就是她画的），如果大家有兴趣的话也可以加入进来对一些优秀设计的开源项目进行源码阅读和画图分享，表示欢迎。

本文研究了 Route 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

APIServerHandler

Overview

下图为 APIServerHandler 核心组件间关联。主要分为 Restful 和 NonRestful 两部分，director 优先使用 Restful 部分，如果处理成功，则退出，不执行 NonRestful 部分；如果 Restful 部分没有目标功能，则执行 NonRestful 部分。FullHandlerChain 用于 HTTP 处理入口点，链接了中间件功能，并将请求引导至 Director 进行处理。

下面为 APIServer 默认的 HandlerChain 构建过程

本文研究了 API Group 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Resource Management

Storage

VersionedResourcesStorageMap 保存 Version -> Resource -> rest.Storage 的映射，第一级映射为 Version，二级为 Resource，Storage 用于解决资源对象的创建、更改、删除等操作。

Install on API Server

getResourceNamesForGroup

APIGroupVersion

Install APIGroup

InstallREST

APIInstaller

Install

registerResourceHandlers

将 rest.Storage 接口，转换为各种操作的接口，代码如下所示。从这里可以看出，rest.Storage 接口是关键，后续再深入探讨。

以 creater 为例，最终，将 creater 或 namedCreater 注册在 Post 方法上

Discovery

在注册代码中，我们可以看到，注册 API 时，返回了可用的 Resources、restful.WebService。随后，马上将该 WebService 可获取的 Resources 注册在该 WebService 的根请求上，动作为 GET。

本文研究了 Storage 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

StorageFactory

StorageFactory 的作用是封装并简化对资源的操作。StorageFactory 最主要的作用是根据传入的 GroupResource 获取对应于该资源的存储配置 Config。

在 API Server 中，StorageFactory 由 StorageFactoryConfig 来生成，StorageFactoryConfig 又通过 EtcdOption 来生成，毕竟无论如何变化，etcd 存储才是最终的目的地。

DefaultStorageFactory

DefaultStorageFactory 是到 1.18 版本前，K8S 内部 StorageFactory 的唯一实现，下面我们来详细分析 DefaultStorageFactory 的模式。

Cohabitating Resources

DefaultStorageFactory 将关联的 GroupResource 组织在一起，从上图可以看到，每一个传入的 GroupResource 是按顺序处理的，因此，关联的 GroupResource 间也是有优先级问题存在的。下图为创建 kube-apiserver 时，使用的 StorageFactory 中关联资源的配置情况

RESTOptionsGetter

Etcd 配置与 StorageFactory 最终都汇入 RESTOptionsGetter 中。RESTOptionsGetter 做为核心配置项，用于通过 GroupResource 找到最终的存储。

创建 storage.Interface 的过程如下图所示

StorageFactoryRestOptionFactory

以 StorageFactoryRestOptionFactory 为例，GetRESTOptions 方法步骤如下

• 使用 StorageFactory 生成 Storage Config

• 创建 RESTOptions 结构体，并保存生成的 Storage Config

• 默认使用 generic.UndecoratedStorage 方法作为 Decorator

• 如果开启了 EnableWatchCache 选项，会修改 Decorator

UndecoratedStorage

UndecoratedStorage 只使用了传入的 storagebackend.Config 参数

直接调用 factory.Create 创建后端存储

本文研究了 Cacher 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Overview

Cacher 包含了 storage.Interface 实例，这个实例是真正的存储后端实例。同时，Cacher 也实现了 storage.Interface 接口，这个是典型的装饰器模式。Kubernetes 源码中，有大量优雅的设计模式的运用，在阅读时可多加注意。简单追踪了下代码，目前猜测的关系如下所示。

registry 包位置如下

storage 位置如下

Store 初始化代码中，设置 DryRunnableStorage 位置

Store

Interface Definition

Store 接口定义在 k8s.io/client-go 中，注意接口中的 Add/Update/Delete，作用是向 Store 中添加对象。那么这个接口的作用就是：API Server 与 Etcd 间的粘合剂。

Event Main Cycle

Cacher 结构定义如下，它包含了一个 watchCache 实例。

再看一下 Cacher 初始化方法，373 行用于创建 watchCache 实例，其中传入的 EventHandler 是 Cacher 的方法，这样，watchCache 就有了向 Cacher 注入事件的通道。

上面代码中 dispatchEvents 方法看起来是对从 watchCache 方法发送过来的 Event 进行处理的部分，我们继续，看起来我们马上就要解决事件来源问题了。

跟踪一下 incoming，那么 processEvent 是否似曾相识呢？

到 watchCache 结构中，找到使用 eventHandler 的位置。

继续挖掘，至此，我们找到了事件完整来源，且事件只有三种类型：Add/Update/Delete。

watchCache.processEvent

原始事件到最终事件的生成如下图所示，使用的 keyFunc、getAttrsFunc、Indexer 等均通过配置传入。

事件创建完毕后，刷新缓存

Event Generation

Cache Watcher

Cacher 中关于 cacheWatcher 的相关结构如下图所示

cacheWatcher 实现了 watch.Interface 接口，用于监听事件。watch.Interface 声明如下所示

watch.Event 定义如下所示

cacheWatcher 核心处理流程如下所示

Watch

Cacher

triggerValue、triggerSupported 判定过程如下

cacheWatcher 的 input channel 缓存大小计算如下

具体添加代码如下

forgetWatcher 如下所示，从 Cacher 中清理 watcher。

Event Dispatching

Bookmark Event

在 Cacher 事件分发过程中，创建了一个 Timer，这个 Timer 每次触发时，有可能会产生一个 Bookmark Event 事件，并分发这个事件。源码如下所示

Dispatch

Bookmark Event 创建后，通过 Versioner 更新事件对象的 ResourceVersion 信息，然后将这个事件进行分发，接下来，我们具体看一下如何分发。

Bookmark Event 分发流程如下图所示，可以看到，事件已分发至全部 ID 小于当前时间的 cacheWatcher 中。

到达 cacheWatcher 后，处理非常简单，只是返回原始对象而已。

General Event

Dispatch

从上图可以看到，当 watchersBuffer 长度大于等于 3 时，将对象缓存起来进行发送。发送 event 时，如果有失败，则获取一个可用时间片，在这个时间片内，尝试阻塞的发送该事件。如果全部发送成功，则等待时间片消耗完毕。

如果在时间片内发送失败，则删除剩余的 cacheWatcher

References

本文研究了 Aggregator Server 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Service Registration

Workflow

通过 Informer 监控 APIService 资源变更，通过 ResourceEventHandler 放入 Controller 队列。Controller 内部处理逻辑与其他 Controller 一致，最终将 APIService 资源变更情况，反映至 Aggregator Server 的 HTTP 处理部分。

Available Condition Controller

Rebuild Service Cache

• 监听的是 APIService 资源变更

• 无论是 Add/Update/Delete，重建 cache 方法一致，使用的是从 API Server 获取的服务列表

Change Condition

AvailableConditionController 的运行协程从 queue 中取出内容，并检查该服务状态后，将服务当前上报至 API Server。

本文研究了 Generic API Server 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Delegation Chain

Overview

Server Chain

HTTP Server

Handler Chain

HandlerChainBuilderFn 类型定义如下，传入一个 http.Handler 实例，并返回一个 http.Handler 实例，这样，可达到类似中间件的效果。

创建 APIServerHandler 时，使用的是下面的方法

Start

最终生成的 preparedAPIAggregator 的启动代码如下，只是简单的调用了 runnable 的 Run 方法，在 Server Chain 中，我们知道，runnable 是 APIAggregator 包含的 GenericAPIServer 生成的 preparedGenericAPIServer 实例。

preparedGenericAPIServer 的 Run 方法如下所示

本文研究了 CRD 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

ResourceConfig

Default Configuration

开启的资源配置及禁用的版本

Extend

开启选型如下

Runtime Support

三者如下图所示

Storage

Custom Resource Definitions

Store 展开后如下图所示

State Transition

Landscape

SharedInformerFactory 用于创建 SharedIndexInformer，后者会周期性的使用 Clientset 连接版本为 v1beta1 或 v1 的 API Extension Services，获取到状态变更后，通知各自的 ResourceEventHandler。在此，还有一些问题需要深入挖掘：

• SharedInformerFactory 如何区分不同类型的资源状态变更

• ResourceEventHandler 是否能同时关注不同类型资源状态的变更

• 资源状态变更是如何获取到的

Clientset

Clientset 功能相对简单，将可用的 API Extension Services 进行封装，每个 RESTClient 都连接在 "Loopback" 地址上，并向不同的服务发送请求。

SharedInformerFactory

Relationship

Add Informers

Management

EstablishingController

EstablishingController 启动后，会启动一个定时执行任务，这个任务每秒检查队列里是否有新的 Key 值，如果有，则更新 Server 端对应资源状态为 Established。

sync 代码如下

CRD Handler

CRD Handler 向 SharedIndexInformer 注册事件处理，Watch 的对象类型 Update 时，则有可能是状态变为 Established 状态，需要向 EstablingController 发送。

CRD Handler 处理请求时，首先检查缓存是否包含请求对象，如果有，返回缓存对象；如果没有，则向 Server 请求，并更改缓存状态。

CRD Controller

本文研究了 Kubernetes 中 Client Shared Informer 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Workflow

从接口间关系可以看出，SharedInformer 是核心组件，它通过 Controller 执行操作，并将结果存入 Store 中。SharedIndexInformer 为 SharedInformer 添加了 Index 功能。

Procedure

Run

Add Handler

ListAndWatch

Indexer

[1] cache 根据 Object 生成 Key 的方式如下

[2] items 根据 Key 获取老对象，并设置新对象

[3] updateIndices 代码如下

[4] sharedIndexInformer 在创建 processorListener 时，如果处于工作状态，会调用 indexer 的 List 方法将全部缓存的 object 取出，并发送给新添加的 processorListener。

最终获取全部事件对象位置

本文研究了 Kubernetes 中 Client Shared Informer 部分的源码，是 Client 篇的第一部分，下面是全系列的链接。

本文研究了 Kubernetes 中 Client Shared Informer 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Workflow

从接口间关系可以看出，SharedInformer 是核心组件，它通过 Controller 执行操作，并将结果存入 Store 中。SharedIndexInformer 为 SharedInformer 添加了 Index 功能。

Procedure

Run

Add Handler

ListAndWatch

Indexer

[1] cache 根据 Object 生成 Key 的方式如下

[2] items 根据 Key 获取老对象，并设置新对象

[3] updateIndices 代码如下

[4] sharedIndexInformer 在创建 processorListener 时，如果处于工作状态，会调用 indexer 的 List 方法将全部缓存的 object 取出，并发送给新添加的 processorListener。

最终获取全部事件对象位置

本文研究了 Kubernetes 中 Client Shared Informer 部分的源码，是 Client 篇的第一部分，下面是全系列的链接。

本文研究了 Master Server 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Server Handler

Serve HTTP Procedure

Add Route

Resource Handler

Install Legacy Resource

首先，判断是否开启了 v1 版本的资源配置，如果开启，才会安装对应的资源处理 API。注意两个核心组件 StorageFactory 与 RESTOptionsGetter，此前都有较为详细的说明。

创建 LegacyRESTStorageProvider 对象，保存 StorageFactory 及其他必要的信息，然后传入方法 InstallLegacyAPI，伴随传入的还有 RESTOptionsGetter。

InstallLegacyAPI 使用传入的参数，创建 APIGroupInfo，并安装。

NewLegacyRESTStorage

• 创建 APIGroupInfo

• 创建各种类型的 RESTStorage，下图没有列举全部

• 构建资源到 Storage 的映射

• 将资源到 Storage 的映射，关联在版本 v1 上

REST

每个资源类型，都有自己的 REST 封装。一般说来，REST 只需要简单的封装一个 Store 即可。创建时，将注册与该资源类型匹配的 NewFunc、NewListFunc 以及行为策略。

要注意，REST 里未必只包含一个 Store，比如 PosStorage

RESTStorageProvider

RESTStorageProvider 配合 Resource Config 与 REST Options 创建 APIGroupInfo，用于向 API Server 注册资源处理方法。

RESTOptionsGetter 根据 APIResourceConfigSource 中的版本、资源检查方法，向 Storage Map 注册 Store，并最终将 Storage Map 挂载到 APIGroupInfo 上。以 Auto Scaling 为例，代码如下所示

创建 v1 版本的 Storage 的代码如下，其他部分大同小异。

不难看出，RESTStorageProvider 是承接配置到 API Group 的核心组件。这样的设计，可以非常明确的划分各个结构、接口的边界，并设定了合理的流程。

Cluster Authentication

Controller Runner

Listener 只有一个 Enqueue 方法，并通过 Notifier 注册到某处。ControllerRunner 控制某一任务的执行，执行过程中如果需要通知外部，则通过已注册的 Listener 列表，广播（或单播）至目标方任务队列。队列拥有方，可能是一个正在等待队列输出的任务。

通过这样的设计，利用队列特性，将两个关联的任务隔离开来，划分好各自边界。Listener 接口的 Enqueue 方法没有参数，因此，Listener 的实现更关注于事件发生，而不是事件内容的具体细节，这种思路值得借鉴。

Dynamic CA

References

Setup

Events

• 创建 Broadcaster

• 使用 Broadcaster 创建 Recorder

• 各组件根据自己需要，使用 Recorder 向 Broadcaster 发送 Event

Informers

Service Informer 是一定存在的 Informer；EndpointSlice 及 Endpoint 二选一，1.18 默认使用 Ednpoit Informer；Node Informer 需要开启 ServiceTopology 选项。最终，都由 Provider 来处理各类事件。 具体来说，每一类型资源都会创建独自的 ResourceEventHandler。各自的 Handler 都是一个 slice 类型，slice 内存储真正的资源 Handler，如下图所示

Bounded Frequency Runner

大家好，这一次给大家带来的是 Controllers 的源码阅读。

Starting Controller

Controller 启动过程是类似的，首先创建到 API Server 的客户端连接 clientset.Interface，它包含了访问 API Server 不同类型资源的客户端。

然后，启动 SharedInformer 接口实例，伴随其启动的，还有一个 Controller 实例。Controller 定期从 API Server 获取资源变更，并存入 Store 实例中。Controller 的 processLoop 协程，从 Store 中顺序读取资源变更事件，并交由 sharedIndexInformer 实例处理，最终到达 ResourceEventHandler。

Controller 实现的核心，在于其对监听资源的变更处理方法上。

简化后的工作流如下图

Controller Manager

Controller Manager 负责启动 Controllers。通过注册不同类型 Controller 的初始化方法，并创建 ControllerContext，隔离了 Controller 具体实现。

Controller Manager ---Create--> ControllerContext ---Pass--> Initialization Function

1.18 版本，注册的 Controller

源码设计图需要配合源码使用，除展示关键设计外，还会有问题性的留白。阅读图的时候发现存在问题的，带着问题去看源码，看别人如何设计解决的，会提高设计能力。

Design

RelationShip

Interface implementation

通过对上图的分析，可以知道 Type 实例中三个容器的作用如下：

• queue： 保存任意类型实例，其中保存的对象不存在有相同内容的对象在 dirty 或 processing 中

• processing： 对象正在处理中，处理完成后移除

• dirty： 与要添加的对象内容相同的对象正在处理中

DelayingInterface

• 添加对象时，如果延时条件已达成，直接进入 Queue

• 延时条件未达成，进入 waitForPriorityQueue，其满足 Heap 接口，没进入一个对象，都会以延迟到达的绝对时刻进行重排

• 如果 waitForPriorityQueue 的“最小元素”不满足延时条件，其他不可能满足延时条件

RateLimiter

• MaxOfRateLimiter 是 RateLimiter 的 Controller，包含了一个 RateLimiter 列表，可填入随意数量的 RateLimiter 具体实现

• MaxOfRateLimiter 本身也是 RateLimiter，其实现方法是使用全部存储的 RateLimiter 分别调用同名方法

• MaxOfRateLimiter 中保存的独立的 RateLimiter 可提供不完整的 RateLimiter 能力，由其他 RateLimiter 补足

IPVS

References

System Control Configuration

• /proc/sys/net/ipv4/conf/all/route_localnet ---> 1

• /proc/sys/net/bridge/bridge-nf-call-iptables ---> 1

• /proc/sys/net/ipv4/vs/conntrack ---> 1

• /proc/sys/net/ipv4/vs/conn_reuse_mode ---> 0

• /proc/sys/net/ipv4/vs/expire_nodest_conn ---> 1

• /proc/sys/net/ipv4/vs/expire_quiescent_template ---> 1

• /proc/sys/net/ipv4/ip_forward ---> 1

• StrictARP

  • /proc/sys/net/ipv4/conf/all/arp_ignore ---> 1

• /proc/sys/net/ipv4/conf/all/arp_announce ---> 2

Sync Services

Basic Chains & Jump

Cluster IP Handling

Create Virtual Server

使用 Service 的 ClusterIP、Port 等信息，创建 VirtualServer，通过 netlink 来查询、创建、更新、删除 VirtualServer。

External IP Handling

VirtualServer 通过 netlink 接口创建 libipvs.Service 与处理 ClusterIP 类型服务时相同。需要注意，如果开启特性 ExternalPolicyForExternalIP，并且当前处理的 Service 的 Endpoint 只存在与当前 Node，那么使用 KUBE-EXTERNAL-IP-LOCAL 存储 Entry 值。

Query Destinations

在同步 Endpoints 前，需要获取当前服务的全部 RealServer，可通过 netlink 根据 VirtualServer 获取其对应的 Destination 列表，再根据 Destination 转化为 RealServer

Synchronize Endpoints

处理过程比较简单，先处理 newEndpoints 中新增或更新的 Endpoint。处理完毕后，将处理删除的 Endpoints，被移除的 IP 及 Port 可通过 curEndpoints.Difference(newEndpoints) 获取。遍历删除列表，如果 IP、Port 已存在于 termination list，则不需要任何处理；如果不存在，将该 IP、Port 存入 termination list，同时存入的还有其对应的 netlink 创建的 Server。

Load Balancer Handling

Node Port

找到本地地址，并根据当前 Service 使用的 NodePort 情况，创建监听的端口。遍历时，如果遇到零地址段，则退出循环，因为这意味着本机全部 IP 都要监听该 NodePort。 创建 Entry 结构时，与当前 Service 的协议相关，如果为 SCTP 协议，类型为：HashIPPort。如果当前 Service 不是 Node Only，则只需要添加至 KUBE-NODE-PORT-protocol 中即可。

Synchronize IPSet

在之前的处理中，在 ipsetList 中各种类型的 IPSet 上，添加了各自的 IP、端口信息，在这个方法中将其应用。utilipset.Interface 实现是基于 ipset 命令集的，ListEntries 方法中传入的 Name 是内部的 utilipset.IPSet 中 Name 域，注意区分。

IPTables Rules

在之前的处理中 RealServer 已创建完毕，但是，每个 Node 只创建了一个 dummy 类型的网络设备 kube-ipvs0，那么，需要通过 iptables 及 ipset 配置，将流量合理引导。 首先创建的是如下的 NAT 规则，因为使用 -A 选择，下图中顺序即规则顺序，现在需要添加规则，从系统内置链中跳转至不同 Chain 中处理。

跳转规则如下所示

Iptables

References

• Man Page of Iptables

• Iptables Extensions Manual

• How To Implement a Basic Firewall Template with Iptables on Ubuntu 14.04

• How To Forward Ports through a Linux Gateway with Iptables

• Iptables HowTo

Packet Flow

图片来源：维基百科 iptables

Table & Chain

Chain 如上图中虚线方框所示，代表了包处理过程中，实际的规则执行路径。Table 是可简单理解为规则的分类：

• Raw：Connection Tracking 相关操作

• Mangle：修改 IP 包

• Nat：地址转换

• Filter：过滤

Proxier

Monitoring Routine

创建 Proxier 结构时，启动了一个协程，用于确保上图的 Table、Chain 都是存在的，然后执行同步规则方法。如果检查存在性时失败，则删除全部的 Table 及 Chain。创建 Table、Chain 成功后，才会执行 syncProxyRules 方法。 Proxier 的 SyncLoop 启动时，注册了 syncProxyRules 的 BoundedFrequencyRunner 也同时启动，可以通过 Sync 方法，触发 syncProxyRules 方法执行，需要注意 BoundedFrequencyRunner 通过令牌桶算法，限制其运行方法的执行频率。

Service Change Handling

ServiceMap

Service 到 ServiceMap 过程如上图所示，LoadBalancer 部分没有在图中标注。如果需要定制化操作，可以通过自定义 makeServiceInfo 方法来实现。ServiceMap 是 iptables 模式下的核心结构之一。

Service Change Track

Service Informer 触发后，由 proxy.Provider 来处理。在 iptables 模式下，处理最终落在 ServiceChangeTracker 的 OnServiceAdd/Update/Delete 方法上。三个方法使用简单的技巧，统一为 Update：

• Create 时调用 OnServiceUpdate(nil, service)

• Delete 时调用 OnServiceDelete(service, nil)

在 ServiceChangeTracker 的 Update 方法中，将当前 Service 对象与前次的 Service 对象分别对应的 ServiceMap 做比较，决定其在 ServiceMap 中的去留，规则如下

• 同一 Service 对应的 NamespacedName 对象相同

• 如果 previous 与 current 保存的 ServiceMap 内容相同，则删除，否则更新

• Create 时，current 为 Service 对应的 ServiceMap，previous 为 nil

• Update 时，更新 current，不改变 previous

• Delete 时，更新 current，不改变 previous

Endpoint Change Handling

Endpoints 处理逻辑与 Service 基本一致，不同指出在于 Endpoints 中包含的 Ports 与 Addresses 可自由组合。同样的，可以通过自定义 makeEndpointInfo 获取 Endpoint 接口对象，这里的 Endpoint 接口，是 Proxy 中使用的，不是 Kubernetes 的资源对象。

处理 Endpoints 变更方法与 Service 并无本质区别，只是将 previous 与 current 指向的对象更根为 EndpointsMap。

Node Change Handling

Node 资源变更处理相对简单，只要变更 Proxier 的 nodeLabels 即可，以 OnNodeAdd 为例

OnNodeUpdate 使用新 Node 对象的 Labels；OnNodeDelete 则将 nodeLabels 设置为 nil。

Sync Proxy Rules

原则上，Service、Endpoints(EndpointSlice) 任何一个的变更都会触发 syncProxyRules 的执行，但是，不要忘记 BoundedFrequencyRunner 限制调用频率的存在，因此，在执行规则同步时，有可能存在 Service、Endpoints(EnpointSlice) 同时变更的可能性。 Service、Endpoints、Node 的变更，全部 kube-proxy 都会收到，后续的处理如果没有特殊说明，每个 Node 都会处理。

ServiceMap

将检测到的 Service 变更情况更新至 ServiceMap 中，已删除的 Service 如果包含 UDP 协议的端口，保留下来。 最终得到了所有 Created、Updated 的 ServiceMap、已删除的 UDP 端口及到目前为止仍然存活的 Service 的健康检查端口。

EndpoinsMap

将 Endpoints 变更应用到 EndpointsMap 上，上图是没有开启 EndpointSlice 特性时的情况。不同于 ServiceMap 处理之处在于保留的是本地健康的 Endpoint IP 数量。

然后，将 staleServiceNames 合并至由 Service 变更引起的 staleServicesClusterIP 中，这样，全部变更的 Cluster IP 获取完毕。

Jumping Chains

创建了如下的自定义链并将默认链处理对应至自定义链

• Nat

  • KUBE-SERVICES

  • KUBE-POSTROUTING

• Filter

  • KUBE-SERVICES

  • KUBE-EXTERNAL-SERVICES

  • KUBE-FORWARD

Iptables Save Format

# Generated by iptables-save v1.3.1 on Sun Apr 23 05:32:09 2006
*filter
:INPUT ACCEPT [273:55355]
:FORWARD ACCEPT [0:0]
:LOGNDROP - [0:0]
:OUTPUT ACCEPT [92376:20668252]
-A INPUT -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT
-A INPUT -i eth0 -p tcp -m tcp --dport 22 -j ACCEPT
-A INPUT -i eth0 -p tcp -m tcp --dport 80 -j ACCEPT
-A INPUT -i lo -j ACCEPT
-A INPUT -j LOGNDROP
-A LOGNDROP -p tcp -m limit --limit 5/min -j LOG --log-prefix "Denied TCP: " --log-level 7
-A LOGNDROP -p udp -m limit --limit 5/min -j LOG --log-prefix "Denied UDP: " --log-level 7
-A LOGNDROP -p icmp -m limit --limit 5/min -j LOG --log-prefix "Denied ICMP: " --log-level 7
-A LOGNDROP -j DROP
COMMIT
# Completed on Sun Apr 23 05:32:09 2006

Prepare Chains & Rules

在根据 ServiceMap 处理规则前，先使用 iptables-save 格式确保以下的 Chain 存在

• Filter

  • KUBE-SERVICES

  • KUBE-EXTERNAL-SERVICES

  • KUBE-FORWARD

• Nat

  • KUBE-SERVICES

  • KUBE-NODEPORTS

  • KUBE-POSTROUTING

  • KUBE-MARK-MASQ

添加 nat 规则

Handling ServiceMaps

Cluster IP

无 Endpoints 的服务，Node 在接收到请求后，直接拒绝，因此规则添加在 Filter 表；Cluster IP 要根据配置进行 NAT 转化。

External IPs

Load Balancer Ingress

是否只有本节点 Endpoints 判断代码如下所示

Node Ports

处理至此，与服务相关的基本规则均已建立，如果本次循环处理的 Service 没有 Endpoints，那么继续处理下一个 Service；如果有，则继续向下建立 Endpoints 规则。

Endpoint Chains

处理到此处，Service 对应的 Endpoints 如果不是 NodeOnly，则处理下一个 Service。后续处理，仅对 NodeOnly 的服务起作用。

然后，对 Node 上每一个网络接口地址设置如下规则

最后，设置 KUBE-FORWARD 上规则如下

本文研究了 Kubernetes 中 Scheduler Cache 部分的源码，通过画图表现其设计思想，希望读者能自行配备源码进行进一步理解，学会自己进行相关设计。

Nodes

Nodes 中保存了 Node.Name 到 nodeInfoListItem 链表的映射。每个 nodeInfoListItem 对应一个 NodeInfo 实例，实例中保存了 Node 信息及相关的 Pod 信息。

AddNode

AddNode 方法执行时，需要传入一个 Node 实例。首先，根据 Node.Name 是否存在于 nodes 中来判断执行路径。 如果 Node.Name 不存在，那么创建一个新的 nodeInfoListItem 并存入 nodes 中。如果已经存在，那么获取对应的链表第一个对象，使用该对象包含的 Node 节点进行镜像清理，需要注意，这里并没有删除镜像，只是在 imageStates 中移除镜像名。

然后，将最近修改的 Node 对应的链表项移动至 headNode 表头，如下图所示，这样也解释了为什么一个 Node 对应的 Key 会关联一个链表。事实上，一个 Key 只有一个链表项，通过 headNode 关联起来的是最近使用顺序。

接着，将 Node 添加至 nodeTree 中，过程如下图

完成后，将 Node 中关联的镜像添加至 imageStates 中，关于 imageState 的清理操作，前面已详细说明，添加操作不再深入。

Pod

AddPod

随后，再执行 addPod，这里不再描述，前面的图中已详细绘制。

UpdatePod

Schedule

Update Snapshot

Summary

本文研究了 Kubernetes 中 Scheduler Cache 部分的源码，进度在 1/5，接下来将整理 Kubernetes 1.18 版本下的全部源码设计图。 预计会有五个大模块，分别是 API Server，Client，Proxy，Controllers 和 Scheduler，和一些辅助工具如 Docker，Go Basic 和 Network 方面统共 123 张源码设计图。敬请期待吧。

大家好，这一次给大家带来的是 Node Controllers 部分的源码阅读。

IPAM Controller

IPAM Controller 在 K8S 1.18 中有四种类型的 CIDRAllocator，分别为：RangeAllocatorType、CloudAllocatorType、IPAMFromClusterAllocatorType 及 IPAMFromClusterAllocatorType。

Controller 在 RangeAllocatorType 或 CloudAllocatorType 模式下，cidrAllocator 有具体指向的结构体负责处理 IPAM 逻辑。

Controller 在 IPAMFromClusterAllocatorType 或 IPAMFromClusterAllocatorType 模式下时，启动 ipam.Controller 来处理，但 K8S 1.18 代码中，仅支持 GCE。

再来看一下 Controller 的 Run 方法，如果类型为 RangeAllocatorType、CloudAllocatorType 时，启动 cidrAllocator 的 Run 方法。

Range Allocator

Node List Initialization

Range Allocator 启动时，Node 有可能已启动完毕，因此，需要获取全部 Node，并标注使用的 IP 地址。根据 Node.Spec.PodCIDRs 获取 IPNet 后，使用如下方法获取索引值。

获取到起始位置后，设置 CidrSet 中对应位，防止重复使用。

On Node Created

nodesInProcessing 中存储正在处理的 Node 名称，新 Node 到达时，如果在其中已经存在该名称，说明该节点正在处理中，直接退出处理程序。

如果 Node 的Spec 中 PodCIDRs 不为空，直接更新 rangeAllocator 中 IP 使用位图即可；如果为空，则创建 nodeReservedCIDRs 结构体，并使用 rangeAllocator 分配 IP，只要有一个 IP 分配成功，那么从 nodesInProcessing 移除该 Node。最后，将新创建的 nodeReservedCIDRs 发送至 Worker 协程处理。

On Node Update

根据 Node 最新状态下 Spec 中 PodCIDRs 长度，决定是要执行的操作，如果长度为零，等同于创建操作，否则直接退出。

On Node Delete

创建 Node 的逆操作，将占用的 IP 资源释放，根据 Node 配置，遍历 PodCIDRs，将占用的资源逐一释放。

相对于创建时的置 1 操作，释放资源时使用清 0 操作

Update CIDR Allocation

在 Node 创建事件回调方法中，曾创建一个 nodeReservedCIDRs 并被发送至 Channel 中。在 Worker 协程中，会捕获该结构，并做处理。具体来说，先根据 Name 从 API Server 端获取 Node 最新信息，根据最新信息中的 PodCIDRs 与 该结构体中预留的 CIDR 对比：

• 长度相同，且每个 IP 信息均相同，则处理完毕

• 上述条件不满足，且 PodCIDRs 长度不为 0，则释放全部 CIDR 资源，并退出

• PodCIDRs 长度为零，那么意味着该 Node 尚未包含 CIDR 资源，分配 CIDR，并通知 API Server，如果成功，则退出；如果 API Server 响应超时，则释放预留 CIDR 资源，退出

Worker 协程中，如果上述处理返回错误，则将 nodeReservedCIDRs 重新发送至 Channel，以待下次处理。

Node Lifecycle Controller

Pod Change Handling

监听到 Pod 变化时，将 Pod 变化情况包装为 podUpdateItem，并放入队列。Pod 变更处理协程从队列中获取到该实例，并处理 Pod 变更情况。如果处理成功，在队列中移除该实例，如果失败，重新将该实例放入队列，以待下次处理。

变更监听代码如下所示，将 Create/Update/Delete 统一至相同方法 podUpdate 中进行后续处理，类似手法已经在之前章节有详细说明，这里不再赘述。

Node Change Handler

Node Processing

No Scheduler Taint

遍历 Node.Status.Conditions，并根据 Condition.Type 在 nodeConditionToKeyStatusMap 中获取到相应信息。nodeConditionToKeyStatusMap 内容如下

获取时如下所示

Procedure

NamespaceController 结构体中包含 RateLimitingInterface 示例，当从 Informer 监听到时间变更时，触发的 Event Handler 将事件对象转换为 key 值，并存入 RateLimitingInterface 实例中。

NamespaceController 在启动运行时，会根据要求，启动多个协程，这些协程都执行相同的功能：从 RateLimitingInterface 实例中获取 key 值，并做处理。

NameResourcesDeleterInterface

Initialize Unsupported Operations

使用 Discover Client 获取服务端支持的全部 Resource 列表，并根据 Resource 是否需要归属于Namespace 来进行过滤，不需要关联于 Namespace 的资源被过滤。

将获取到的 Resources 进行遍历，根据 GroupVersion 进行分类，并获取该资源支持的操作列表(Verbs)，将不支持 list、deletecollection 操作的 Resource 记录下来。

Delete

• Source

Delete All Content

本文研究了 Kubernetes 中 Priority Queue 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Internal Data Structures

Heap

上图为 PriorityQueue 中 activeQ 域，它是一个 Heap 实例。Heap 中核心结构为 data，包含一个字符串到 heapItem 的映射，heapItem 存储了实际对象以及该对象的 Key 在 queue 中位置的变量。

Pods Containers

PriorityQueue 中与 Pod 关联的两个数据结构如上图所示。UnschedulablePodsMap 中保存了从 Pod 信息到 key 值的方法。

nominatedPodMap

Add

优先使用传入的 nodeName，若 nodeName 为空时，使用 UID，若 UID 也为空，处理完毕。否则，按上图示意，添加对应的 map。

Pod Handling

Added/Update

kube-scheduler 接收到添加 Pod 事件后，会将 Pod 添加进 SchedulerCache 中，随后，执行上图操作。Pod 的添加、更新操作执行相同代码，区别为状态不同，分别对应为 AssignedPodAdd 与 AssignedPodUpdate。

SchedulerCache 注

// State Machine of a pod's events in scheduler's cache:
//
//
//   +-------------------------------------------+  +----+
//   |                            Add            |  |    |
//   |                                           |  |    | Update
//   +      Assume                Add            v  v    |
//Initial +--------> Assumed +------------+---> Added <--+
//   ^                +   +               |       +
//   |                |   |               |       |
//   |                |   |           Add |       | Remove
//   |                |   |               |       |
//   |                |   |               +       |
//   +----------------+   +-----------> Expired   +----> Deleted
//         Forget             Expire

Delete

将 podInfoMap 中全部 PodInfo 移动至 podBackoffQ 或 activeQ 中，并删除 podInfoMap 中对应 K/V 对，标记状态为 AssignedPodDelete。

Handling Routines

Flush Backoff Queue

定时从 backoff queue 中获取一个 PodInfo 对象，检查其 backoff time 是否到期，如果没有到期，直接返回，等待下次触发，此时，PodInfo 对象仍然存在于 backoff queue 中。如果到期，则弹出该对象，并存入 active queue 中，此时，PodInfo 对象被移除出 backoff queue。

Flush Unscheduable Queue

本文研究了 Kubernetes 中 Priority Queue 部分的源码，是 Scheduler 的第二部分。

本文研究了 Kubernetes 中 Plugins 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Basic Structures

Plugins

Plugins 包含了一组 PluginSet，每个 PluginSet 又包含了两组 Plugin，一组为激活状态，一组为禁用状态。Plugin 包含一个唯一标识符和该 Plugin 的权重。

Quick Sort Plugin

接口方法 Less 如上图所示，传入两个 PodInfo 实例，通过 Less 方法判定二者在排序时先后位置。由于 Plugin 接口只有通用方法 Name，因此，每个特定功能的 Plugin 原则上可随意定制自己需要的方法。

InterPodAffinity

将与当前调度 Pod 的相关 Node 数据存储在 CycleState 的 PreFilterInterPodAffinity 关键字中，供后续调度使用。

Configurator

Profile

Registry 用于组织 provider 与 Plugins 间关联关系，保存的是默认的 Plugins。在创建 Profile 结构时，会使用这些默认的 Plugins。

Extender

相关代码如下图所示

Profile Map

本文研究了 Kubernetes 中 Plugins 部分的源码，是 Scheduler 的第三部分。

Landscape

References

EventBroadcaster

Implementation

EventCollerator

Event Recorder

References

Tunner

Landscape

References

本文研究了 Kubernetes 中 Schedule 部分的源码，配备源码进行进一步理解，可以加深理解,增强相关设计能力。

Scheduling

NextPod

Scheduler 中封装了具体调度算法，并与调度算法共享相同的 SchedulingQueue 实例对象。在执行调度时，首先需要通过 NextPod 方法获取待调度的 PodInfo。

Find Profile for Pod

Profile 中保存着调度框架基本信息。根据选中的当前调度 PodInfo 中的 Pod 实例，选中合适的调度 Profile。根据 Profile、Pod 来判断是否需要跳过当前 Pod 调度，如果判断结果不需要调度当前 Pod，本次调度完成；如果需要调度，则通过 Algorithm 进行调度。

Algorithm

Overview

ScheduleAlgorithm 定义了关于调度的两个核心方法，同时，也通过 Extenders 方法预留出了扩展空间。

Schedule

genericScheduler 实现了 Scheduler 接口，接下来，我们详细看下 genericScheduler 的核心调度方法的实现。

Pod Basic Check

Run PreFilter

根据 Pod 选择 Profile 后，执行 Profile 的 RunPreFilterPlugins 方法，该方法由 framework 结构提供。执行过程对每个注册的 PreFilterPlugin 接口，执行其 PreFilter 方法，如果执行中有错误发生，那么创建 Status 结构，记录错误信息，并返回，不再执行后续的 PreFilterPlugin 接口。如果全部 PreFilterPlugin 接口都执行成功，返回 nil。Status 结构定义非常简洁，如下

Find Nodes Passed Filters

首先根据 Snapshot 中 nodeInfoList 的长度来计算最大 Node 数量，并预先分配 Node 切片。然后根据当前要调度的 NodeInfo，获取其 Node 的名称，在 SchedulingQueue 中查找对应的 Pod。遍历全部 Pod 数组，使用当前 Profile，并执行其 RunPreFilterExtensionAddPod 方法，如果通过，则将该 Pod 存入 Node 中。最后，对 Node 执行 Profile 的 RunFilterPlugins 方法。

Find Nodes Passed Extenders

进一步筛选已通过检查的 Node 列表，筛选出满足 SchedulerExtender 要求的 Node 列表。至此，确认了当前调度的 Pod 可选择的 Node 列表。筛选 Node 完成后，使用选中的 Profile 进行 PreScore 操作。

完成 PreScore 操作后，如果仍然存在多于一个可选 Node 的情况，将执行优先级运算，最终根据优先级运算结果经由 selectHost 方法确认要使用的 Host 名称，一次调度过程完成。

Scheduler Volume Binder

AssumeCache

PodBindingCache

本文研究了 Kubernetes 中 Schedule 部分的源码，是 Scheduler 最后一部分。

大家好，这一次给大家带来的是 Endpoint Controller 部分的源码阅读。

ResourceEventHandler for Service

Selector Cache

EndpointController 在收到 v1.Service 资源变更时，将服务 Spec 中指定的 Selector 对象，保存在一个 map 中。该 map 的 key 通过 DeletionHandlingMetaNamespaceKeyFunc 方法生成，如下所示：

使用的 MetaNamespaceKeyFunc 如下所示：

最后，将生成的 key 放入 EndpointController 的工作队列，待后续处理。附上 Kubernetes 官方的 Service 对象配置示例。

Synchronize Service

上图为 syncService 完整的流程。第一步，先根据通过 Service 获取的 Key 值，还原 Namespace 与 Name，这两个值是后续向 API Server 查询的核心依据。

• 服务删除，那么通知 API Server 删除该服务相关的 Endpoint

• 如果是创建、更新服务

  • 从 API Server 根据 (Namespace, Name) 拉取与现有与 Service.Spec.Selector 相关的 Pod 列表

  • 根据 Pod 列表获取子网地址列表

  • 获取当前与 Service 关联的 Endpoints

  • DeepCopy 返回的 Endpoints 对象，并更新 Copy 的 Endpoints 的子网地址列表

  • 使用新 Endpoints，通知 API Server 创建或更新 Endpoints 资源

  • 创建/更新过程发生错误时，记录事件

Subnets

从 API Server 获取的全部 Pod 都会做如下处理来生成 EndpointAddress，以下两种情况下，Pod 被跳过，继续处理下个 Pod：

• Pod.PodStatus.IP 长度为 0

• 没有设置 tolerateUnreadyEndpoints 且 Pod 将被删除

如果设置了 IPv6DualStack，则根据 Service.Spec.ClusterIP 的类型（IPv4 或 IPv6），从 Pod.PodStatus.PodIPs 中存在同类型的地址，找到即刻返回。如果同类型地址不存在，则报错。

Set Hostname

获取到 IP 地址的 EndpointAddress 结构，会根据下图条件，设置 EndpointAddress 结构的 Hostname 域。

EndpointSubset

生成 EndpointAddress 后，根据 Service.ServiceSpec.Ports 配置，生成 EndpointSubset 列表，并存入全局的列表中。

• 如果设置了 tolerateUnreadyEndpoints 或当前遍历的 Pod 处于 Ready 状态，Ready 计数 +1

• 如果不满足上述情况，且当前遍历的 Pod 应该归属于 Endpoint，那么 Unready 计数 +1

ResourceEventHandler for Pod

On Add Pod

从 API Server 获取 Pod.Namespace 下所有 Service。遍历 Service，如果缓存中没有该 Service 存在，更新缓存。从缓存中获取 Service 使用的 Selector，并与 Pod 的 Labels 比对，如果一致，说明该服务受到 Pod 影响，添加进队列等待处理。

On Update Pod

On Delete Pod

删除的 Pod 无法从 API Server 上获取，那么从传入的 obj 中获取即可。找到被删除节点后，处理方式就和添加 Pod 一样了。

获取 Pod 对象的方法如下